Merge branch 'master' into synap5e/feat/experiment-models-extensions

tests-unit/app_test/test_migrations.py

@@ -8,6 +8,7 @@ upgrade/downgrade for 0003+.
 """
 
 import os
+import sqlite3
 
 import pytest
 from alembic import command
@@ -30,6 +31,12 @@ def _make_config(db_path: str) -> Config:
     return cfg
 
 
+def _sqlite_path(cfg: Config) -> str:
+    url = cfg.get_main_option("sqlalchemy.url")
+    assert url is not None and url.startswith("sqlite:///")
+    return url.removeprefix("sqlite:///")
+
+
 @pytest.fixture
 def migration_db(tmp_path):
     """Yield an alembic Config pre-upgraded to the baseline revision."""
@@ -55,3 +62,26 @@ def test_upgrade_downgrade_cycle(migration_db):
     command.upgrade(migration_db, "head")
     command.downgrade(migration_db, _BASELINE)
     command.upgrade(migration_db, "head")
+
+
+def test_case_sensitive_tags_downgrade_normalizes_existing_tags(migration_db):
+    """Downgrading 0005 folds mixed-case tag vocabulary before restoring CHECK."""
+    command.upgrade(migration_db, "0005_allow_case_sensitive_tags")
+
+    db_path = _sqlite_path(migration_db)
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("INSERT INTO tags(name) VALUES (?)", ("NewTag",))
+        conn.execute("INSERT INTO tags(name) VALUES (?)", ("newtag",))
+        conn.execute("INSERT INTO tags(name) VALUES (?)", ("model_type:LLM",))
+
+    command.downgrade(migration_db, "0004_drop_tag_type")
+
+    with sqlite3.connect(db_path) as conn:
+        tags = {row[0] for row in conn.execute("SELECT name FROM tags")}
+        assert "newtag" in tags
+        assert "model_type:llm" in tags
+        assert "NewTag" not in tags
+        assert "model_type:LLM" not in tags
+
+        with pytest.raises(sqlite3.IntegrityError):
+            conn.execute("INSERT INTO tags(name) VALUES (?)", ("Upper",))

tests-unit/assets_test/conftest.py

@@ -234,7 +234,7 @@ def seeded_asset(request: pytest.FixtureRequest, http: requests.Session, api_bas
     p = getattr(request, "param", {}) or {}
     tags: Optional[list[str]] = p.get("tags")
     if tags is None:
-        tags = ["models", "checkpoints", "unit-tests", "alpha"]
+        tags = ["models", "model_type:checkpoints", "unit-tests", "alpha"]
     meta = {"purpose": "test", "epoch": 1, "flags": ["x", "y"], "nullable": None}
     # Unique content per test so the seed always creates a fresh asset (201).
     # Delete is now always a soft delete, so content from a prior test survives

tests-unit/assets_test/queries/test_asset_info.py

@@ -133,6 +133,66 @@ class TestListReferencesPage:
         assert total == 1
         assert refs[0].name == "tagged"
 
+    def test_include_tags_filter_ands_persisted_model_tags(self, session: Session):
+        asset = _make_asset(session, "hash-model-tags")
+        checkpoint = _make_reference(session, asset, name="checkpoint")
+        lora = _make_reference(session, asset, name="lora")
+        input_ref = _make_reference(session, asset, name="input")
+        ensure_tags_exist(
+            session,
+            ["models", "model_type:checkpoints", "model_type:loras", "unit-tests"],
+        )
+        add_tags_to_reference(
+            session,
+            reference_id=checkpoint.id,
+            tags=["models", "model_type:checkpoints", "unit-tests"],
+            origin="automatic",
+        )
+        add_tags_to_reference(
+            session,
+            reference_id=lora.id,
+            tags=["models", "model_type:loras", "unit-tests"],
+            origin="automatic",
+        )
+        add_tags_to_reference(
+            session,
+            reference_id=input_ref.id,
+            tags=["unit-tests"],
+        )
+        session.commit()
+
+        refs, _, total = list_references_page(
+            session,
+            include_tags=["models", "model_type:checkpoints", "unit-tests"],
+        )
+
+        assert total == 1
+        assert refs[0].id == checkpoint.id
+
+    def test_include_tags_filter_preserves_model_type_case(self, session: Session):
+        asset = _make_asset(session, "hash-model-case")
+        ref = _make_reference(session, asset, name="llm")
+        ensure_tags_exist(session, ["models", "model_type:LLM"])
+        add_tags_to_reference(
+            session,
+            reference_id=ref.id,
+            tags=["models", "model_type:LLM"],
+            origin="automatic",
+        )
+        session.commit()
+
+        refs, _, total = list_references_page(
+            session, include_tags=["models", "model_type:LLM"]
+        )
+        refs_lower, _, total_lower = list_references_page(
+            session, include_tags=["models", "model_type:llm"]
+        )
+
+        assert total == 1
+        assert refs[0].id == ref.id
+        assert total_lower == 0
+        assert refs_lower == []
+
     def test_exclude_tags_filter(self, session: Session):
         asset = _make_asset(session, "hash1")
         _make_reference(session, asset, name="keep")

tests-unit/assets_test/queries/test_cache_state.py

@@ -176,6 +176,39 @@ class TestUpsertReference:
         ref = session.query(AssetReference).filter_by(file_path=file_path).one()
         assert ref.mtime_ns == final_mtime
 
+    def test_upsert_refreshes_loader_path_on_existing_reference(self, session: Session):
+        """Re-ingesting an existing reference writes the loader_path computed
+        by that ingest, healing NULL or stale values even when nothing else
+        about the row changed."""
+        asset = _make_asset(session, "hash1")
+        file_path = "/models/checkpoints/sub/model.safetensors"
+
+        upsert_reference(
+            session, asset_id=asset.id, file_path=file_path, name="model",
+            mtime_ns=100, loader_path=None,
+        )
+        session.commit()
+
+        created, updated = upsert_reference(
+            session, asset_id=asset.id, file_path=file_path, name="model",
+            mtime_ns=100, loader_path="sub/model.safetensors",
+        )
+        session.commit()
+
+        assert created is False
+        assert updated is True
+        ref = session.query(AssetReference).filter_by(file_path=file_path).one()
+        assert ref.loader_path == "sub/model.safetensors"
+
+        # Identical loader_path is a no-op, not a spurious update.
+        created, updated = upsert_reference(
+            session, asset_id=asset.id, file_path=file_path, name="model",
+            mtime_ns=100, loader_path="sub/model.safetensors",
+        )
+        session.commit()
+        assert created is False
+        assert updated is False
+
     def test_upsert_restores_missing_reference(self, session: Session):
         """Upserting a reference that was marked missing should restore it."""
         asset = _make_asset(session, "hash1")

tests-unit/assets_test/queries/test_tags.py

@@ -58,7 +58,7 @@ class TestEnsureTagsExist:
         session.commit()
 
         tags = session.query(Tag).all()
-        assert {t.name for t in tags} == {"alpha", "beta"}
+        assert {t.name for t in tags} == {"ALPHA", "Beta", "alpha"}
 
     def test_empty_list_is_noop(self, session: Session):
         ensure_tags_exist(session, [])
@@ -258,6 +258,16 @@ class TestListTagsWithUsage:
         tag_names = {name for name, _ in rows}
         assert tag_names == {"alpha", "alphabet"}
 
+    def test_prefix_filter_is_case_sensitive(self, session: Session):
+        ensure_tags_exist(session, ["model_type:LLM", "model_type:llm"])
+        session.commit()
+
+        rows, total = list_tags_with_usage(session, prefix="model_type:L")
+
+        tag_names = {name for name, _ in rows}
+        assert tag_names == {"model_type:LLM"}
+        assert total == 1
+
     def test_order_by_name(self, session: Session):
         ensure_tags_exist(session, ["zebra", "alpha", "middle"])
         session.commit()

tests-unit/assets_test/services/test_asset_response_loader_path.py

@@ -0,0 +1,83 @@
+"""Tests for how _build_asset_response derives the response `loader_path`.
+
+Guards the persist-and-read contract: the response reads the stored
+`loader_path` verbatim, with no read-time recomputation. Like tags, the
+value is a seed-time derivative healed by the scan lifecycle.
+"""
+
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import patch
+
+from app.assets.api.routes import _build_asset_response
+from app.assets.services.schemas import AssetDetailResult, ReferenceData
+
+_TS = datetime(2024, 1, 1, 0, 0, 0)
+
+
+def _make_result(
+    *, file_path: str | None, loader_path: str | None
+) -> AssetDetailResult:
+    ref = ReferenceData(
+        id="ref-1",
+        name="model.safetensors",
+        file_path=file_path,
+        loader_path=loader_path,
+        user_metadata=None,
+        preview_id=None,
+        created_at=_TS,
+        updated_at=_TS,
+        last_access_time=_TS,
+    )
+    return AssetDetailResult(ref=ref, asset=None, tags=[])
+
+
+def test_uses_persisted_loader_path_without_recomputing():
+    """A stored loader_path is returned verbatim, not re-derived from file_path.
+
+    The sentinel value could never be produced by compute_loader_path for this
+    file_path, so seeing it in the response proves the stored column is read.
+    """
+    result = _make_result(
+        file_path="/unmatched/root/model.safetensors",
+        loader_path="SENTINEL/stored.safetensors",
+    )
+
+    resp = _build_asset_response(result)
+
+    assert resp.loader_path == "SENTINEL/stored.safetensors"
+
+
+def test_null_stored_loader_path_is_served_as_null(tmp_path: Path):
+    """No read-time recomputation: a NULL column is served as null even when
+    the path would resolve."""
+    models = tmp_path / "models"
+    ckpt = models / "checkpoints"
+    ckpt.mkdir(parents=True)
+    f = ckpt / "bar.safetensors"
+    f.touch()
+
+    with patch("app.assets.services.path_utils.folder_paths") as mock_fp, patch(
+        "app.assets.services.path_utils.get_comfy_models_folders",
+        return_value=[("checkpoints", [str(ckpt)], {".safetensors"})],
+    ):
+        mock_fp.get_input_directory.return_value = str(tmp_path / "in")
+        mock_fp.get_output_directory.return_value = str(tmp_path / "out")
+        mock_fp.get_temp_directory.return_value = str(tmp_path / "tmp")
+        mock_fp.models_dir = str(models)
+
+        result = _make_result(file_path=str(f), loader_path=None)
+        resp = _build_asset_response(result)
+
+        assert resp.loader_path is None
+        assert resp.display_name == "checkpoints/bar.safetensors"
+
+
+def test_all_path_fields_null_without_file_path():
+    """API-created / hash-only references (no file_path) expose no paths."""
+    result = _make_result(file_path=None, loader_path=None)
+
+    resp = _build_asset_response(result)
+
+    assert resp.loader_path is None
+    assert resp.display_name is None

tests-unit/assets_test/services/test_bulk_ingest.py

@@ -1,10 +1,14 @@
 """Tests for bulk ingest services."""
 
+import os
 from pathlib import Path
+from unittest.mock import patch
 
 from sqlalchemy.orm import Session
 
 from app.assets.database.models import Asset, AssetReference
+from app.assets.database.queries import get_reference_tags
+from app.assets.scanner import build_asset_specs
 from app.assets.services.bulk_ingest import SeedAssetSpec, batch_insert_seed_assets
 
 
@@ -101,6 +105,184 @@ class TestBatchInsertSeedAssets:
             asset = session.query(Asset).filter_by(id=ref.asset_id).first()
             assert asset.mime_type == expected_mime, f"Expected {expected_mime} for {filename}, got {asset.mime_type}"
 
+    def test_duplicate_paths_merge_tags_before_insert(
+        self, session: Session, temp_dir: Path
+    ):
+        """Overlapping model-folder registrations can emit the same path twice."""
+        file_path = temp_dir / "shared.safetensors"
+        file_path.write_bytes(b"shared model")
+
+        specs: list[SeedAssetSpec] = [
+            {
+                "abs_path": str(file_path),
+                "size_bytes": 12,
+                "mtime_ns": 1234567890000000000,
+                "info_name": "Shared Model",
+                "tags": ["models", "model_type:checkpoints"],
+                "fname": "shared.safetensors",
+                "metadata": None,
+                "hash": None,
+                "mime_type": "application/safetensors",
+            },
+            {
+                "abs_path": str(file_path),
+                "size_bytes": 12,
+                "mtime_ns": 1234567890000000000,
+                "info_name": "Shared Model",
+                "tags": ["models", "model_type:diffusion_models"],
+                "fname": "shared.safetensors",
+                "metadata": None,
+                "hash": None,
+                "mime_type": "application/safetensors",
+            },
+        ]
+
+        result = batch_insert_seed_assets(session, specs=specs, owner_id="")
+
+        assert result.inserted_refs == 1
+        assert result.won_paths == 1
+        refs = session.query(AssetReference).all()
+        assert len(refs) == 1
+        assert set(get_reference_tags(session, reference_id=refs[0].id)) == {
+            "models",
+            "model_type:checkpoints",
+            "model_type:diffusion_models",
+        }
+
+    def test_duplicate_paths_are_merged_after_abspath_normalization(
+        self, session: Session, temp_dir: Path, monkeypatch
+    ):
+        """The scanner may emit equivalent paths with different spelling."""
+        file_path = temp_dir / "same-file.safetensors"
+        file_path.write_bytes(b"shared model")
+        monkeypatch.chdir(temp_dir)
+        relative_path = file_path.name
+        absolute_path = os.path.abspath(relative_path)
+
+        specs: list[SeedAssetSpec] = [
+            {
+                "abs_path": relative_path,
+                "size_bytes": 12,
+                "mtime_ns": 1234567890000000000,
+                "info_name": "Shared Model",
+                "tags": ["models", "model_type:checkpoints"],
+                "fname": "same-file.safetensors",
+                "metadata": None,
+                "hash": None,
+                "mime_type": "application/safetensors",
+            },
+            {
+                "abs_path": absolute_path,
+                "size_bytes": 12,
+                "mtime_ns": 1234567890000000000,
+                "info_name": "Shared Model",
+                "tags": ["models", "model_type:diffusion_models"],
+                "fname": "same-file.safetensors",
+                "metadata": None,
+                "hash": None,
+                "mime_type": "application/safetensors",
+            },
+        ]
+
+        result = batch_insert_seed_assets(session, specs=specs, owner_id="")
+
+        assert result.inserted_refs == 1
+        assert result.won_paths == 1
+        refs = session.query(AssetReference).all()
+        assert len(refs) == 1
+        assert refs[0].file_path == absolute_path
+        # loader_path is persisted from the spec's fname (compute_loader_path).
+        assert refs[0].loader_path == "same-file.safetensors"
+        assert set(get_reference_tags(session, reference_id=refs[0].id)) == {
+            "models",
+            "model_type:checkpoints",
+            "model_type:diffusion_models",
+        }
+
+    def test_scanner_duplicate_shared_model_paths_keep_all_model_type_tags(
+        self, session: Session, temp_dir: Path
+    ):
+        """Shared extra model roots make scanner collection emit duplicate paths."""
+        shared_root = temp_dir / "shared"
+        input_dir = temp_dir / "input"
+        output_dir = temp_dir / "output"
+        temp_root = temp_dir / "temp"
+        for directory in (shared_root, input_dir, output_dir, temp_root):
+            directory.mkdir()
+        file_path = shared_root / "dual_use_model.safetensors"
+        file_path.write_bytes(b"shared model")
+
+        with (
+            patch("app.assets.services.path_utils.folder_paths") as mock_fp,
+            patch(
+                "app.assets.services.path_utils.get_comfy_models_folders",
+                return_value=[
+                    ("checkpoints", [str(shared_root)], {".safetensors"}),
+                    ("diffusion_models", [str(shared_root)], {".safetensors"}),
+                ],
+            ),
+        ):
+            mock_fp.get_input_directory.return_value = str(input_dir)
+            mock_fp.get_output_directory.return_value = str(output_dir)
+            mock_fp.get_temp_directory.return_value = str(temp_root)
+
+            specs, tag_pool, skipped = build_asset_specs(
+                paths=[str(file_path), str(file_path)],
+                existing_paths=set(),
+                enable_metadata_extraction=False,
+                compute_hashes=False,
+            )
+
+        assert skipped == 0
+        assert len(specs) == 2
+        assert tag_pool == {
+            "models",
+            "model_type:checkpoints",
+            "model_type:diffusion_models",
+        }
+
+        result = batch_insert_seed_assets(session, specs=specs, owner_id="")
+
+        assert result.inserted_refs == 1
+        assert result.won_paths == 1
+        refs = session.query(AssetReference).all()
+        assert len(refs) == 1
+        assert set(get_reference_tags(session, reference_id=refs[0].id)) == {
+            "models",
+            "model_type:checkpoints",
+            "model_type:diffusion_models",
+        }
+
+    def test_loader_path_persisted_as_null_when_fname_is_none(
+        self, session: Session, temp_dir: Path
+    ):
+        """A file with no in-root loader path (fname=None, e.g. an orphan under
+        models_root) persists loader_path as NULL rather than a synthesized value."""
+        file_path = temp_dir / "orphan.bin"
+        file_path.write_bytes(b"x")
+
+        specs: list[SeedAssetSpec] = [
+            {
+                "abs_path": str(file_path),
+                "size_bytes": 1,
+                "mtime_ns": 1234567890000000000,
+                "info_name": "orphan.bin",
+                "tags": [],
+                "fname": None,
+                "metadata": None,
+                "hash": None,
+                "mime_type": None,
+            }
+        ]
+
+        result = batch_insert_seed_assets(session, specs=specs, owner_id="")
+
+        assert result.inserted_refs == 1
+        refs = session.query(AssetReference).all()
+        assert len(refs) == 1
+        assert refs[0].file_path == str(file_path)
+        assert refs[0].loader_path is None
+
 
 class TestMetadataExtraction:
     def test_extracts_mime_type_for_model_files(self, temp_dir: Path):

tests-unit/assets_test/services/test_ingest.py

@@ -94,6 +94,47 @@ class TestIngestFileFromPath:
         ref_tags = get_reference_tags(session, reference_id=result.reference_id)
         assert set(ref_tags) == {"models", "checkpoints"}
 
+    def test_path_derived_tags_use_automatic_origin(
+        self, mock_create_session, temp_dir: Path, session: Session
+    ):
+        input_dir = temp_dir / "input"
+        output_dir = temp_dir / "output"
+        temp_root = temp_dir / "temp"
+        for directory in (input_dir, output_dir, temp_root):
+            directory.mkdir()
+        file_path = input_dir / "pasted" / "tagged.png"
+        file_path.parent.mkdir()
+        file_path.write_bytes(b"data")
+
+        with (
+            patch("app.assets.services.path_utils.folder_paths") as mock_fp,
+            patch(
+                "app.assets.services.path_utils.get_comfy_models_folders",
+                return_value=[],
+            ),
+        ):
+            mock_fp.get_input_directory.return_value = str(input_dir)
+            mock_fp.get_output_directory.return_value = str(output_dir)
+            mock_fp.get_temp_directory.return_value = str(temp_root)
+
+            result = _ingest_file_from_path(
+                abs_path=str(file_path),
+                asset_hash="blake3:pathorigin",
+                size_bytes=4,
+                mtime_ns=1234567890000000000,
+                info_name="Tagged Asset",
+                tags=["input", "manual-label"],
+            )
+
+        assert result.reference_id is not None
+        links = session.query(AssetReferenceTag).filter_by(
+            asset_reference_id=result.reference_id
+        )
+        origin_by_tag = {link.tag_name: link.origin for link in links}
+        assert origin_by_tag["input"] == "automatic"
+        assert origin_by_tag["pasted"] == "automatic"
+        assert origin_by_tag["manual-label"] == "manual"
+
     def test_idempotent_upsert(self, mock_create_session, temp_dir: Path, session: Session):
         file_path = temp_dir / "dup.bin"
         file_path.write_bytes(b"content")
@@ -288,6 +329,45 @@ class TestIngestExistingFileTagFK:
             assert "output" in ref_tag_names
 
 
+class TestIngestExistingFileLoaderPath:
+    """Outputs saved into a subfolder must persist the subfolder-qualified
+    loader path, not the bare basename (regression: spec["fname"] was
+    os.path.basename)."""
+
+    def test_subfoldered_output_persists_relative_loader_path(
+        self, mock_create_session, temp_dir: Path, session: Session
+    ):
+        input_dir = temp_dir / "input"
+        output_dir = temp_dir / "output"
+        temp_root = temp_dir / "temp"
+        for directory in (input_dir, output_dir, temp_root):
+            directory.mkdir()
+        file_path = output_dir / "sub" / "img_00001_.png"
+        file_path.parent.mkdir()
+        file_path.write_bytes(b"image data")
+
+        with (
+            patch("app.assets.services.path_utils.folder_paths") as mock_fp,
+            patch(
+                "app.assets.services.path_utils.get_comfy_models_folders",
+                return_value=[],
+            ),
+        ):
+            mock_fp.get_input_directory.return_value = str(input_dir)
+            mock_fp.get_output_directory.return_value = str(output_dir)
+            mock_fp.get_temp_directory.return_value = str(temp_root)
+
+            assert ingest_existing_file(abs_path=str(file_path)) is True
+
+        ref = (
+            session.query(AssetReference)
+            .filter_by(file_path=str(file_path))
+            .one()
+        )
+        assert ref.loader_path == "sub/img_00001_.png"
+        assert (ref.user_metadata or {}).get("filename") == "sub/img_00001_.png"
+
+
 class TestIngestImageDimensions:
     """system_metadata should carry {kind, width, height} for image assets."""
 

tests-unit/assets_test/services/test_path_utils.py

@@ -6,7 +6,16 @@ from unittest.mock import patch
 
 import pytest
 
-from app.assets.services.path_utils import get_asset_category_and_relative_path
+from app.assets.services.path_utils import (
+    compute_display_name,
+    compute_loader_path,
+    compute_logical_path,
+    get_asset_category_and_relative_path,
+    get_known_input_subfolder_tags_from_path,
+    get_known_subfolder_tags,
+    get_name_and_tags_from_asset_path,
+    resolve_destination_from_tags,
+)
 
 
 @pytest.fixture
@@ -17,7 +26,8 @@ def fake_dirs():
         input_dir = root_path / "input"
         output_dir = root_path / "output"
         temp_dir = root_path / "temp"
-        models_dir = root_path / "models" / "checkpoints"
+        models_root = root_path / "models"
+        models_dir = models_root / "checkpoints"
         for d in (input_dir, output_dir, temp_dir, models_dir):
             d.mkdir(parents=True)
 
@@ -25,15 +35,17 @@ def fake_dirs():
             mock_fp.get_input_directory.return_value = str(input_dir)
             mock_fp.get_output_directory.return_value = str(output_dir)
             mock_fp.get_temp_directory.return_value = str(temp_dir)
+            mock_fp.models_dir = str(models_root)
 
             with patch(
                 "app.assets.services.path_utils.get_comfy_models_folders",
-                return_value=[("checkpoints", [str(models_dir)])],
+                return_value=[("checkpoints", [str(models_dir)], {".safetensors"})],
             ):
                 yield {
                     "input": input_dir,
                     "output": output_dir,
                     "temp": temp_dir,
+                    "models_root": models_root,
                     "models": models_dir,
                 }
 
@@ -76,6 +88,538 @@ class TestGetAssetCategoryAndRelativePath:
         cat, rel = get_asset_category_and_relative_path(str(f))
         assert cat == "models"
 
+    def test_model_path_tags_include_registered_model_type_only(self, fake_dirs):
+        f = fake_dirs["models"] / "subdir" / "model.safetensors"
+        f.parent.mkdir()
+        f.touch()
+
+        _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert "model_type:checkpoints" in tags
+        assert "checkpoints" not in tags
+        assert "subdir" not in tags
+
+    def test_model_type_preserves_registered_folder_case(self, fake_dirs):
+        llm_dir = fake_dirs["models"].parent / "LLM"
+        llm_dir.mkdir()
+        f = llm_dir / "model.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("LLM", [str(llm_dir)], {".safetensors"})],
+        ):
+            _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert "model_type:LLM" in tags
+        assert "model_type:llm" not in tags
+
+    def test_path_components_do_not_create_model_type_tags(self, fake_dirs):
+        f = fake_dirs["models"] / "loras" / "model.safetensors"
+        f.parent.mkdir()
+        f.touch()
+
+        _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert "model_type:checkpoints" in tags
+        assert "loras" not in tags
+        assert "model_type:loras" not in tags
+
+    def test_shared_root_returns_all_matching_model_type_tags(self, fake_dirs):
+        shared_root = fake_dirs["models"].parent / "shared"
+        shared_root.mkdir()
+        f = shared_root / "foo.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[
+                ("checkpoints", [str(shared_root)], {".safetensors"}),
+                ("loras", [str(shared_root)], {".safetensors"}),
+            ],
+        ):
+            _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert "model_type:checkpoints" in tags
+        assert "model_type:loras" in tags
+
+    def test_shared_root_model_type_tags_respect_bucket_extensions(self, fake_dirs):
+        """Buckets sharing a base dir only tag files matching their extensions."""
+        shared_root = fake_dirs["models"].parent / "unet"
+        shared_root.mkdir()
+        safetensors_file = shared_root / "wan.safetensors"
+        gguf_file = shared_root / "wan.gguf"
+        safetensors_file.touch()
+        gguf_file.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[
+                ("diffusion_models", [str(shared_root)], {".safetensors"}),
+                ("unet_gguf", [str(shared_root)], {".gguf"}),
+            ],
+        ):
+            _name, safetensors_tags = get_name_and_tags_from_asset_path(str(safetensors_file))
+            _name, gguf_tags = get_name_and_tags_from_asset_path(str(gguf_file))
+
+        assert "model_type:diffusion_models" in safetensors_tags
+        assert "model_type:unet_gguf" not in safetensors_tags
+        assert "model_type:unet_gguf" in gguf_tags
+        assert "model_type:diffusion_models" not in gguf_tags
+
+    def test_empty_extension_set_tags_any_extension(self, fake_dirs):
+        """Custom buckets registered without extensions accept every file."""
+        custom_root = fake_dirs["models"].parent / "custom_bucket"
+        custom_root.mkdir()
+        f = custom_root / "weights.bin"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("custom_bucket", [str(custom_root)], set())],
+        ):
+            _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert "model_type:custom_bucket" in tags
+
+    def test_no_extension_match_keeps_models_tag_without_model_type(self, fake_dirs):
+        f = fake_dirs["models"] / "notes.txt"
+        f.touch()
+
+        _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert not any(tag.startswith("model_type:") for tag in tags)
+
+    def test_output_backed_registered_folder_gets_model_and_output_tags(self, fake_dirs):
+        output_checkpoints_dir = fake_dirs["output"] / "checkpoints"
+        output_checkpoints_dir.mkdir()
+        f = output_checkpoints_dir / "saved.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("checkpoints", [str(output_checkpoints_dir)], {".safetensors"})],
+        ):
+            _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "models" in tags
+        assert "model_type:checkpoints" in tags
+        assert "output" in tags
+
+    def test_temp_path_tags_include_temp_not_output_or_preview(self, fake_dirs):
+        f = fake_dirs["temp"] / "preview.png"
+        f.touch()
+
+        _name, tags = get_name_and_tags_from_asset_path(str(f))
+
+        assert "temp" in tags
+        assert "output" not in tags
+        assert "preview:true" not in tags
+
+    def test_known_subfolder_tags_are_centralized(self):
+        assert get_known_subfolder_tags("pasted") == ["pasted"]
+        assert get_known_subfolder_tags("arbitrary") == []
+
+    def test_known_input_subfolder_tags_are_path_derived_for_direct_children(self, fake_dirs):
+        f = fake_dirs["input"] / "pasted" / "image.png"
+        f.parent.mkdir()
+        f.touch()
+
+        assert get_known_input_subfolder_tags_from_path(str(f)) == ["pasted"]
+
+        _name, tags = get_name_and_tags_from_asset_path(str(f))
+        assert "input" in tags
+        assert "pasted" in tags
+
+    def test_known_input_subfolder_tags_do_not_apply_to_nested_or_other_roots(self, fake_dirs):
+        nested = fake_dirs["input"] / "pasted" / "session" / "image.png"
+        output = fake_dirs["output"] / "pasted" / "image.png"
+        for path in (nested, output):
+            path.parent.mkdir(parents=True)
+            path.touch()
+
+        assert get_known_input_subfolder_tags_from_path(str(nested)) == []
+        assert get_known_input_subfolder_tags_from_path(str(output)) == []
+
     def test_unknown_path_raises(self, fake_dirs):
         with pytest.raises(ValueError, match="not within"):
             get_asset_category_and_relative_path("/some/random/path.png")
+
+
+class TestResponseStoragePaths:
+    def test_input_file_path_and_display_name_include_subfolder(self, fake_dirs):
+        sub = fake_dirs["input"] / "some" / "folder"
+        sub.mkdir(parents=True)
+        f = sub / "image.png"
+        f.touch()
+
+        assert compute_logical_path(str(f)) == "input/some/folder/image.png"
+        assert compute_display_name(str(f)) == "some/folder/image.png"
+
+    def test_output_file_path_and_display_name_include_subfolder(self, fake_dirs):
+        sub = fake_dirs["output"] / "renders"
+        sub.mkdir()
+        f = sub / "ComfyUI_00001_.png"
+        f.touch()
+
+        assert compute_logical_path(str(f)) == "output/renders/ComfyUI_00001_.png"
+        assert compute_display_name(str(f)) == "renders/ComfyUI_00001_.png"
+
+    def test_temp_file_path_and_display_name(self, fake_dirs):
+        f = fake_dirs["temp"] / "preview.png"
+        f.touch()
+
+        assert compute_logical_path(str(f)) == "temp/preview.png"
+        assert compute_display_name(str(f)) == "preview.png"
+
+    def test_exact_storage_root_has_no_display_name(self, fake_dirs):
+        assert compute_logical_path(str(fake_dirs["input"])) == "input"
+        assert compute_display_name(str(fake_dirs["input"])) is None
+
+    def test_longest_matching_builtin_root_wins(self, fake_dirs, tmp_path: Path):
+        nested_output = fake_dirs["input"] / "nested-output"
+        nested_output.mkdir()
+        f = nested_output / "image.png"
+        f.touch()
+
+        with patch("app.assets.services.path_utils.folder_paths") as mock_fp:
+            mock_fp.get_input_directory.return_value = str(fake_dirs["input"])
+            mock_fp.get_output_directory.return_value = str(nested_output)
+            mock_fp.get_temp_directory.return_value = str(tmp_path / "temp")
+            mock_fp.models_dir = str(fake_dirs["models_root"])
+
+            assert compute_logical_path(str(f)) == "output/image.png"
+            assert compute_display_name(str(f)) == "image.png"
+
+    def test_model_file_path_is_relative_to_physical_models_root(self, fake_dirs):
+        sub = fake_dirs["models"] / "flux"
+        sub.mkdir()
+        f = sub / "model.safetensors"
+        f.touch()
+
+        assert compute_logical_path(str(f)) == "models/checkpoints/flux/model.safetensors"
+        assert compute_display_name(str(f)) == "checkpoints/flux/model.safetensors"
+
+        name, tags = get_name_and_tags_from_asset_path(str(f))
+        assert name == "model.safetensors"
+        assert "models" in tags
+        assert "model_type:checkpoints" in tags
+        assert "checkpoints" not in tags
+        assert "flux" not in tags
+
+    @pytest.mark.parametrize(
+        "folder_name",
+        ["checkpoints", "clip", "vae", "diffusion_models", "loras"],
+    )
+    def test_output_model_folder_uses_output_storage_file_path(self, fake_dirs, folder_name):
+        output_model_dir = fake_dirs["output"] / folder_name
+        output_model_dir.mkdir(exist_ok=True)
+        default_model_dir = fake_dirs["models_root"] / folder_name
+        default_model_dir.mkdir(exist_ok=True)
+        f = output_model_dir / "saved.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[
+                (folder_name, [str(default_model_dir), str(output_model_dir)], {".safetensors"})
+            ],
+        ):
+            assert compute_logical_path(str(f)) == f"output/{folder_name}/saved.safetensors"
+            assert compute_display_name(str(f)) == f"{folder_name}/saved.safetensors"
+
+            name, tags = get_name_and_tags_from_asset_path(str(f))
+            assert name == "saved.safetensors"
+            assert "output" in tags
+            assert "models" in tags
+            assert f"model_type:{folder_name}" in tags
+            assert folder_name not in tags
+
+    def test_output_model_subfolder_uses_output_storage_file_path(self, fake_dirs):
+        folder_name = "loras"
+        output_model_dir = fake_dirs["output"] / folder_name
+        subdir = output_model_dir / "experiments"
+        subdir.mkdir(parents=True)
+        f = subdir / "my_lora.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[(folder_name, [str(output_model_dir)], {".safetensors"})],
+        ):
+            assert (
+                compute_logical_path(str(f))
+                == "output/loras/experiments/my_lora.safetensors"
+            )
+            assert compute_display_name(str(f)) == "loras/experiments/my_lora.safetensors"
+
+            name, tags = get_name_and_tags_from_asset_path(str(f))
+            assert name == "my_lora.safetensors"
+            assert "output" in tags
+            assert "models" in tags
+            assert "model_type:loras" in tags
+            assert "loras" not in tags
+            assert "experiments" not in tags
+
+    def test_external_model_folder_without_provenance_has_no_file_path(self, tmp_path: Path):
+        external_checkpoints_dir = tmp_path / "external" / "not_named_like_category"
+        external_checkpoints_dir.mkdir(parents=True)
+        f = external_checkpoints_dir / "external.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("checkpoints", [str(external_checkpoints_dir)], {".safetensors"})],
+        ):
+            assert compute_logical_path(str(f)) is None
+            assert compute_display_name(str(f)) is None
+
+            name, tags = get_name_and_tags_from_asset_path(str(f))
+            assert name == "external.safetensors"
+            assert "models" in tags
+            assert "model_type:checkpoints" in tags
+
+    def test_same_relative_model_file_under_multiple_external_roots_has_no_storage_file_path(
+        self, tmp_path: Path
+    ):
+        foo_dir = tmp_path / "foo"
+        bar_dir = tmp_path / "bar"
+        foo_dir.mkdir()
+        bar_dir.mkdir()
+        foo_file = foo_dir / "baz.safetensors"
+        bar_file = bar_dir / "baz.safetensors"
+        foo_file.touch()
+        bar_file.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("checkpoints", [str(foo_dir), str(bar_dir)], {".safetensors"})],
+        ):
+            assert compute_logical_path(str(foo_file)) is None
+            assert compute_logical_path(str(bar_file)) is None
+            assert compute_display_name(str(foo_file)) is None
+            assert compute_display_name(str(bar_file)) is None
+
+    def test_output_clip_folder_uses_output_storage_and_text_encoder_tag(self, fake_dirs):
+        output_clip_dir = fake_dirs["output"] / "clip"
+        output_clip_dir.mkdir()
+        f = output_clip_dir / "clip_l.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("text_encoders", [str(output_clip_dir)], {".safetensors"})],
+        ):
+            assert compute_logical_path(str(f)) == "output/clip/clip_l.safetensors"
+            assert compute_display_name(str(f)) == "clip/clip_l.safetensors"
+
+            name, tags = get_name_and_tags_from_asset_path(str(f))
+            assert name == "clip_l.safetensors"
+            assert "output" in tags
+            assert "models" in tags
+            assert "model_type:text_encoders" in tags
+            assert "clip" not in tags
+
+    def test_physical_unet_folder_uses_storage_path_and_diffusion_models_tag(self, fake_dirs):
+        unet_dir = fake_dirs["models_root"] / "unet"
+        diffusion_models_dir = fake_dirs["models_root"] / "diffusion_models"
+        unet_dir.mkdir()
+        diffusion_models_dir.mkdir()
+        f = unet_dir / "wan.safetensors"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[
+                ("diffusion_models", [str(unet_dir), str(diffusion_models_dir)], {".safetensors"})
+            ],
+        ):
+            assert compute_logical_path(str(f)) == "models/unet/wan.safetensors"
+            assert compute_display_name(str(f)) == "unet/wan.safetensors"
+
+            name, tags = get_name_and_tags_from_asset_path(str(f))
+            assert name == "wan.safetensors"
+            assert "models" in tags
+            assert "model_type:diffusion_models" in tags
+            assert "unet" not in tags
+
+    def test_unregistered_file_under_physical_models_root_still_has_storage_file_path(self, fake_dirs):
+        f = fake_dirs["models_root"] / "not_registered" / "orphan.bin"
+        f.parent.mkdir()
+        f.touch()
+
+        assert compute_logical_path(str(f)) == "models/not_registered/orphan.bin"
+        assert compute_display_name(str(f)) == "not_registered/orphan.bin"
+
+    def test_output_checkpoint_folder_without_registration_has_only_output_tag(self, fake_dirs):
+        f = fake_dirs["output"] / "checkpoints" / "saved.safetensors"
+        f.parent.mkdir(exist_ok=True)
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[],
+        ):
+            assert compute_logical_path(str(f)) == "output/checkpoints/saved.safetensors"
+            assert compute_display_name(str(f)) == "checkpoints/saved.safetensors"
+
+            name, tags = get_name_and_tags_from_asset_path(str(f))
+            assert name == "saved.safetensors"
+            assert "output" in tags
+            assert "models" not in tags
+            assert not any(tag.startswith("model_type:") for tag in tags)
+
+    def test_unknown_path_returns_none(self):
+        assert compute_logical_path("/some/random/path.png") is None
+        assert compute_display_name("/some/random/path.png") is None
+
+
+class TestLoaderPath:
+    """In-root loader path: relative to the storage root, model category dropped."""
+
+    def test_model_loader_path_drops_category(self, fake_dirs):
+        sub = fake_dirs["models"] / "flux"
+        sub.mkdir()
+        f = sub / "model.safetensors"
+        f.touch()
+
+        # logical_path keeps the category, file_path (loader) drops it
+        assert compute_logical_path(str(f)) == "models/checkpoints/flux/model.safetensors"
+        assert compute_loader_path(str(f)) == "flux/model.safetensors"
+
+    def test_model_loader_path_flat_file(self, fake_dirs):
+        f = fake_dirs["models"] / "model.safetensors"
+        f.touch()
+
+        assert compute_loader_path(str(f)) == "model.safetensors"
+
+    def test_input_loader_path_keeps_subfolders(self, fake_dirs):
+        sub = fake_dirs["input"] / "some" / "folder"
+        sub.mkdir(parents=True)
+        f = sub / "image.png"
+        f.touch()
+
+        assert compute_loader_path(str(f)) == "some/folder/image.png"
+
+    def test_temp_loader_path(self, fake_dirs):
+        f = fake_dirs["temp"] / "preview.png"
+        f.touch()
+
+        assert compute_loader_path(str(f)) == "preview.png"
+
+    def test_unregistered_file_under_models_root_has_no_loader_path(self, fake_dirs):
+        # Under models_root but not within any registered category base.
+        f = fake_dirs["models_root"] / "not_registered" / "orphan.bin"
+        f.parent.mkdir()
+        f.touch()
+
+        # It still has a namespaced logical_path, but no loader path.
+        assert compute_logical_path(str(f)) == "models/not_registered/orphan.bin"
+        assert compute_loader_path(str(f)) is None
+
+    def test_extension_mismatch_in_registered_bucket_has_no_loader_path(self, fake_dirs):
+        # Inside a registered bucket, but the bucket's extension set cannot
+        # load it: no model_type tag, and no loader path either.
+        f = fake_dirs["models"] / "notes.txt"
+        f.touch()
+
+        assert compute_logical_path(str(f)) == "models/checkpoints/notes.txt"
+        assert compute_loader_path(str(f)) is None
+
+    def test_shared_base_loader_path_uses_extension_matching_bucket(self, fake_dirs):
+        shared_root = fake_dirs["models"].parent / "unet"
+        shared_root.mkdir()
+        f = shared_root / "wan.gguf"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[
+                ("diffusion_models", [str(shared_root)], {".safetensors"}),
+                ("unet_gguf", [str(shared_root)], {".gguf"}),
+            ],
+        ):
+            assert compute_loader_path(str(f)) == "wan.gguf"
+
+    def test_match_all_bucket_provides_loader_path_for_any_extension(self, fake_dirs):
+        custom_root = fake_dirs["models"].parent / "custom_bucket"
+        custom_root.mkdir()
+        f = custom_root / "weights.bin"
+        f.touch()
+
+        with patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("custom_bucket", [str(custom_root)], set())],
+        ):
+            assert compute_loader_path(str(f)) == "weights.bin"
+
+    def test_extra_path_model_has_loader_path_but_no_logical_path(self, tmp_path: Path):
+        """Registered category base outside models_dir (extra_model_paths style).
+
+        Loadable, so loader_path resolves; but it is not under any canonical
+        storage root, so logical_path/display_name are None. This asymmetry is
+        intentional: loader_path resolves every registered model-folder base,
+        logical_path only resolves the canonical storage roots.
+        """
+        extra = tmp_path / "extra_ckpts"
+        extra.mkdir()
+        f = extra / "foo.safetensors"
+        f.touch()
+
+        with patch("app.assets.services.path_utils.folder_paths") as mock_fp, patch(
+            "app.assets.services.path_utils.get_comfy_models_folders",
+            return_value=[("checkpoints", [str(extra)], {".safetensors"})],
+        ):
+            mock_fp.get_input_directory.return_value = str(tmp_path / "in")
+            mock_fp.get_output_directory.return_value = str(tmp_path / "out")
+            mock_fp.get_temp_directory.return_value = str(tmp_path / "tmp")
+            mock_fp.models_dir = str(tmp_path / "models")  # extra is NOT under this
+
+            assert compute_loader_path(str(f)) == "foo.safetensors"
+            assert compute_logical_path(str(f)) is None
+            assert compute_display_name(str(f)) is None
+
+    def test_unknown_path_returns_none(self):
+        assert compute_loader_path("/some/random/path.png") is None
+
+
+class TestResolveDestinationFromTags:
+    def test_extra_tags_are_not_path_components(self, fake_dirs):
+        base_dir, subdirs = resolve_destination_from_tags(["input", "unit-tests", "foo"])
+
+        assert base_dir == os.path.abspath(fake_dirs["input"])
+        assert subdirs == []
+
+    def test_model_upload_rejects_non_writable_registered_folders(self):
+        with tempfile.TemporaryDirectory() as root:
+            root_path = Path(root)
+            checkpoints_dir = root_path / "models" / "checkpoints"
+            configs_dir = root_path / "models" / "configs"
+            custom_nodes_dir = root_path / "custom_nodes"
+            for path in (checkpoints_dir, configs_dir, custom_nodes_dir):
+                path.mkdir(parents=True)
+
+            with patch("app.assets.services.path_utils.folder_paths") as mock_fp:
+                mock_fp.folder_names_and_paths = {
+                    "checkpoints": ([str(checkpoints_dir)], set()),
+                    "configs": ([str(configs_dir)], set()),
+                    "custom_nodes": ([str(custom_nodes_dir)], set()),
+                }
+
+                base_dir, subdirs = resolve_destination_from_tags(
+                    ["models", "model_type:checkpoints"]
+                )
+                assert base_dir == os.path.abspath(checkpoints_dir)
+                assert subdirs == []
+
+                for folder_name in ("configs", "custom_nodes"):
+                    with pytest.raises(ValueError, match="unknown model category"):
+                        resolve_destination_from_tags(
+                            ["models", f"model_type:{folder_name}"]
+                        )

tests-unit/assets_test/test_assets_missing_sync.py

@@ -19,7 +19,8 @@ def test_seed_asset_removed_when_file_is_deleted(
     """Asset without hash (seed) whose file disappears:
        after triggering sync_seed_assets, Asset + AssetInfo disappear.
     """
-    # Create a file directly under input/unit-tests/<case> so tags include "unit-tests"
+    # Create a file directly under input/unit-tests/<case>. Backend tags only
+    # classify the root; nested path components are not exposed as tags.
     case_dir = comfy_tmp_base_dir / root / "unit-tests" / "syncseed"
     case_dir.mkdir(parents=True, exist_ok=True)
     name = f"seed_{uuid.uuid4().hex[:8]}.bin"
@@ -32,7 +33,7 @@ def test_seed_asset_removed_when_file_is_deleted(
     # Verify it is visible via API and carries no hash (seed)
     r1 = http.get(
         api_base + "/api/assets",
-        params={"include_tags": "unit-tests,syncseed", "name_contains": name},
+        params={"include_tags": root, "name_contains": name},
         timeout=120,
     )
     body1 = r1.json()
@@ -54,7 +55,7 @@ def test_seed_asset_removed_when_file_is_deleted(
     # It should disappear (AssetInfo and seed Asset gone)
     r2 = http.get(
         api_base + "/api/assets",
-        params={"include_tags": "unit-tests,syncseed", "name_contains": name},
+        params={"include_tags": root, "name_contains": name},
         timeout=120,
     )
     body2 = r2.json()
@@ -132,7 +133,7 @@ def test_hashed_asset_two_asset_infos_both_get_missing(
     second_id = b2["id"]
 
     # Remove the single underlying file
-    p = comfy_tmp_base_dir / "input" / "unit-tests" / "multiinfo" / get_asset_filename(b2["asset_hash"], ".png")
+    p = comfy_tmp_base_dir / "input" / get_asset_filename(created["asset_hash"], ".png")
     assert p.exists()
     p.unlink()
 
@@ -250,8 +251,7 @@ def test_missing_tag_clears_on_fastpass_when_mtime_and_size_match(
 
     a = asset_factory(name, [root, "unit-tests", scope], {}, data)
     aid = a["id"]
-    base = comfy_tmp_base_dir / root / "unit-tests" / scope
-    p = base / get_asset_filename(a["asset_hash"], ".bin")
+    p = comfy_tmp_base_dir / root / get_asset_filename(a["asset_hash"], ".bin")
     st0 = p.stat()
     orig_mtime_ns = getattr(st0, "st_mtime_ns", int(st0.st_mtime * 1_000_000_000))
 

tests-unit/assets_test/test_crud.py

@@ -290,7 +290,7 @@ def test_metadata_filename_is_set_for_seed_asset_without_hash(
 
     r1 = http.get(
         api_base + "/api/assets",
-        params={"include_tags": f"unit-tests,{scope}", "name_contains": name},
+        params={"include_tags": root, "name_contains": name},
         timeout=120,
     )
     body = r1.json()

tests-unit/assets_test/test_downloads.py

@@ -23,7 +23,7 @@ def test_download_svg_forced_to_attachment(http: requests.Session, api_base: str
     svg = b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>'
     files = {"file": ("evil.svg", svg, "image/svg+xml")}
     form_data = {
-        "tags": json.dumps(["models", "checkpoints", "unit-tests", "svgxss"]),
+        "tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "svgxss"]),
         "name": "evil.svg",
     }
     up = http.post(api_base + "/api/assets", files=files, data=form_data, timeout=120)
@@ -131,7 +131,7 @@ def test_download_chooses_existing_state_and_updates_access_time(
         assert t1 > t0
 
 
-@pytest.mark.parametrize("seeded_asset", [{"tags": ["models", "checkpoints"]}], indirect=True)
+@pytest.mark.parametrize("seeded_asset", [{"tags": ["models", "model_type:checkpoints"]}], indirect=True)
 def test_download_missing_file_returns_404(
     http: requests.Session, api_base: str, comfy_tmp_base_dir: Path, seeded_asset: dict
 ):

tests-unit/assets_test/test_list_cursor.py

@@ -13,7 +13,7 @@ def _seed(asset_factory, make_asset_bytes, count: int, tag: str) -> list[str]:
     for n in names:
         asset_factory(
             n,
-            ["models", "checkpoints", "unit-tests", tag],
+            ["models", "model_type:checkpoints", "unit-tests", tag],
             {},
             make_asset_bytes(n, size=2048),
         )
@@ -208,7 +208,7 @@ def test_cursor_walks_for_non_name_sorts(sort_field, http: requests.Session, api
     names = []
     for i in range(4):
         n = f"cursor_{sort_field}_{i:02d}.safetensors"
-        asset_factory(n, ["models", "checkpoints", "unit-tests", f"cursor-{sort_field}"], {}, make_asset_bytes(n, size=2048 + i))
+        asset_factory(n, ["models", "model_type:checkpoints", "unit-tests", f"cursor-{sort_field}"], {}, make_asset_bytes(n, size=2048 + i))
         names.append(n)
 
     params = {

tests-unit/assets_test/test_list_filter.py

@@ -11,7 +11,7 @@ def test_list_assets_paging_and_sort(http: requests.Session, api_base: str, asse
     for n in names:
         asset_factory(
             n,
-            ["models", "checkpoints", "unit-tests", "paging"],
+            ["models", "model_type:checkpoints", "unit-tests", "paging"],
             {"epoch": 1},
             make_asset_bytes(n, size=2048),
         )
@@ -45,8 +45,8 @@ def test_list_assets_paging_and_sort(http: requests.Session, api_base: str, asse
 
 
 def test_list_assets_include_exclude_and_name_contains(http: requests.Session, api_base: str, asset_factory):
-    a = asset_factory("inc_a.safetensors", ["models", "checkpoints", "unit-tests", "alpha"], {}, b"X" * 1024)
-    b = asset_factory("inc_b.safetensors", ["models", "checkpoints", "unit-tests", "beta"], {}, b"Y" * 1024)
+    a = asset_factory("inc_a.safetensors", ["models", "model_type:checkpoints", "unit-tests", "alpha"], {}, b"X" * 1024)
+    b = asset_factory("inc_b.safetensors", ["models", "model_type:checkpoints", "unit-tests", "beta"], {}, b"Y" * 1024)
 
     r = http.get(
         api_base + "/api/assets",
@@ -81,7 +81,7 @@ def test_list_assets_include_exclude_and_name_contains(http: requests.Session, a
 
 
 def test_list_assets_sort_by_size_both_orders(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-size"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-size"]
     n1, n2, n3 = "sz1.safetensors", "sz2.safetensors", "sz3.safetensors"
     asset_factory(n1, t, {}, make_asset_bytes(n1, 1024))
     asset_factory(n2, t, {}, make_asset_bytes(n2, 2048))
@@ -108,7 +108,7 @@ def test_list_assets_sort_by_size_both_orders(http, api_base, asset_factory, mak
 
 
 def test_list_assets_sort_by_updated_at_desc(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-upd"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-upd"]
     a1 = asset_factory("upd_a.safetensors", t, {}, make_asset_bytes("upd_a", 1200))
     a2 = asset_factory("upd_b.safetensors", t, {}, make_asset_bytes("upd_b", 1200))
 
@@ -131,7 +131,7 @@ def test_list_assets_sort_by_updated_at_desc(http, api_base, asset_factory, make
 
 
 def test_list_assets_sort_by_last_access_time_desc(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-access"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-access"]
     asset_factory("acc_a.safetensors", t, {}, make_asset_bytes("acc_a", 1100))
     time.sleep(0.02)
     a2 = asset_factory("acc_b.safetensors", t, {}, make_asset_bytes("acc_b", 1100))
@@ -154,14 +154,14 @@ def test_list_assets_sort_by_last_access_time_desc(http, api_base, asset_factory
 
 
 def test_list_assets_include_tags_variants_and_case(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-include"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-include"]
     a = asset_factory("incvar_alpha.safetensors", [*t, "alpha"], {}, make_asset_bytes("iva"))
     asset_factory("incvar_beta.safetensors", [*t, "beta"], {}, make_asset_bytes("ivb"))
 
-    # CSV + case-insensitive
+    # CSV tag filters are whitespace-trimmed and case-sensitive.
     r1 = http.get(
         api_base + "/api/assets",
-        params={"include_tags": "UNIT-TESTS,LF-INCLUDE,alpha"},
+        params={"include_tags": "unit-tests,lf-include,alpha"},
         timeout=120,
     )
     b1 = r1.json()
@@ -196,14 +196,14 @@ def test_list_assets_include_tags_variants_and_case(http, api_base, asset_factor
 
 
 def test_list_assets_exclude_tags_dedup_and_case(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-exclude"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-exclude"]
     a = asset_factory("ex_a_alpha.safetensors", [*t, "alpha"], {}, make_asset_bytes("exa", 900))
     asset_factory("ex_b_beta.safetensors", [*t, "beta"], {}, make_asset_bytes("exb", 900))
 
-    # Exclude uppercase should work
+    # Exclude filters are case-sensitive.
     r1 = http.get(
         api_base + "/api/assets",
-        params={"include_tags": "unit-tests,lf-exclude", "exclude_tags": "BETA"},
+        params={"include_tags": "unit-tests,lf-exclude", "exclude_tags": "beta"},
         timeout=120,
     )
     b1 = r1.json()
@@ -225,7 +225,7 @@ def test_list_assets_exclude_tags_dedup_and_case(http, api_base, asset_factory,
 
 
 def test_list_assets_name_contains_case_and_specials(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-name"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-name"]
     a1 = asset_factory("CaseMix.SAFE", t, {}, make_asset_bytes("cm", 800))
     a2 = asset_factory("case-other.safetensors", t, {}, make_asset_bytes("co", 800))
 
@@ -261,7 +261,7 @@ def test_list_assets_name_contains_case_and_specials(http, api_base, asset_facto
 
 
 def test_list_assets_offset_beyond_total_and_limit_boundary(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "lf-pagelimits"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "lf-pagelimits"]
     asset_factory("pl1.safetensors", t, {}, make_asset_bytes("pl1", 600))
     asset_factory("pl2.safetensors", t, {}, make_asset_bytes("pl2", 600))
     asset_factory("pl3.safetensors", t, {}, make_asset_bytes("pl3", 600))
@@ -319,7 +319,7 @@ def test_list_assets_name_contains_literal_underscore(
       - foobar.safetensors       (must NOT match)
     """
     scope = f"lf-underscore-{uuid.uuid4().hex[:6]}"
-    tags = ["models", "checkpoints", "unit-tests", scope]
+    tags = ["models", "model_type:checkpoints", "unit-tests", scope]
 
     a = asset_factory("foo_bar.safetensors", tags, {}, make_asset_bytes("a", 700))
     b = asset_factory("fooxbar.safetensors", tags, {}, make_asset_bytes("b", 700))

tests-unit/assets_test/test_metadata_filters.py

@@ -5,7 +5,7 @@ def test_meta_and_across_keys_and_types(
     http, api_base: str, asset_factory, make_asset_bytes
 ):
     name = "mf_and_mix.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-and"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-and"]
     meta = {"purpose": "mix", "epoch": 1, "active": True, "score": 1.23}
     asset_factory(name, tags, meta, make_asset_bytes(name, 4096))
 
@@ -41,7 +41,7 @@ def test_meta_and_across_keys_and_types(
 
 def test_meta_type_strictness_int_vs_str_and_bool(http, api_base, asset_factory, make_asset_bytes):
     name = "mf_types.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-types"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-types"]
     meta = {"epoch": 1, "active": True}
     asset_factory(name, tags, meta, make_asset_bytes(name))
 
@@ -95,7 +95,7 @@ def test_meta_type_strictness_int_vs_str_and_bool(http, api_base, asset_factory,
 
 def test_meta_any_of_list_of_scalars(http, api_base, asset_factory, make_asset_bytes):
     name = "mf_list_scalars.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-list"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-list"]
     meta = {"flags": ["red", "green"]}
     asset_factory(name, tags, meta, make_asset_bytes(name, 3000))
 
@@ -134,7 +134,7 @@ def test_meta_none_semantics_missing_or_null_and_any_of_with_none(
     http, api_base, asset_factory, make_asset_bytes
 ):
     # a1: key missing; a2: explicit null; a3: concrete value
-    t = ["models", "checkpoints", "unit-tests", "mf-none"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "mf-none"]
     a1 = asset_factory("mf_none_missing.safetensors", t, {"x": 1}, make_asset_bytes("a1"))
     a2 = asset_factory("mf_none_null.safetensors", t, {"maybe": None}, make_asset_bytes("a2"))
     a3 = asset_factory("mf_none_value.safetensors", t, {"maybe": "x"}, make_asset_bytes("a3"))
@@ -166,7 +166,7 @@ def test_meta_none_semantics_missing_or_null_and_any_of_with_none(
 
 def test_meta_nested_json_object_equality(http, api_base, asset_factory, make_asset_bytes):
     name = "mf_nested_json.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-nested"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-nested"]
     cfg = {"optimizer": "adam", "lr": 0.001, "schedule": {"type": "cosine", "warmup": 100}}
     asset_factory(name, tags, {"config": cfg}, make_asset_bytes(name, 2200))
 
@@ -197,7 +197,7 @@ def test_meta_nested_json_object_equality(http, api_base, asset_factory, make_as
 
 def test_meta_list_of_objects_any_of(http, api_base, asset_factory, make_asset_bytes):
     name = "mf_list_objects.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-objlist"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-objlist"]
     transforms = [{"type": "crop", "size": 128}, {"type": "flip", "p": 0.5}]
     asset_factory(name, tags, {"transforms": transforms}, make_asset_bytes(name, 2048))
 
@@ -228,7 +228,7 @@ def test_meta_list_of_objects_any_of(http, api_base, asset_factory, make_asset_b
 
 def test_meta_with_special_and_unicode_keys(http, api_base, asset_factory, make_asset_bytes):
     name = "mf_keys_unicode.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-keys"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-keys"]
     meta = {
         "weird.key": "v1",
         "path/like": 7,
@@ -259,7 +259,7 @@ def test_meta_with_special_and_unicode_keys(http, api_base, asset_factory, make_
 
 
 def test_meta_with_zero_and_boolean_lists(http, api_base, asset_factory, make_asset_bytes):
-    t = ["models", "checkpoints", "unit-tests", "mf-zero-bool"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "mf-zero-bool"]
     a0 = asset_factory("mf_zero_count.safetensors", t, {"count": 0}, make_asset_bytes("z", 1025))
     a1 = asset_factory("mf_bool_list.safetensors", t, {"choices": [True, False]}, make_asset_bytes("b", 1026))
 
@@ -286,7 +286,7 @@ def test_meta_with_zero_and_boolean_lists(http, api_base, asset_factory, make_as
 
 def test_meta_mixed_list_types_and_strictness(http, api_base, asset_factory, make_asset_bytes):
     name = "mf_mixed_list.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "mf-mixed"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "mf-mixed"]
     meta = {"mix": ["1", 1, True, None]}
     asset_factory(name, tags, meta, make_asset_bytes(name, 1999))
 
@@ -311,7 +311,7 @@ def test_meta_mixed_list_types_and_strictness(http, api_base, asset_factory, mak
 
 def test_meta_unknown_key_and_none_behavior_with_scope_tags(http, api_base, asset_factory, make_asset_bytes):
     # Use a unique scope tag to avoid interference
-    t = ["models", "checkpoints", "unit-tests", "mf-unknown-scope"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "mf-unknown-scope"]
     x = asset_factory("mf_unknown_a.safetensors", t, {"k1": 1}, make_asset_bytes("ua"))
     y = asset_factory("mf_unknown_b.safetensors", t, {"k2": 2}, make_asset_bytes("ub"))
 
@@ -340,13 +340,13 @@ def test_meta_with_tags_include_exclude_and_name_contains(http, api_base, asset_
     # alpha matches epoch=1; beta has epoch=2
     a = asset_factory(
         "mf_tag_alpha.safetensors",
-        ["models", "checkpoints", "unit-tests", "mf-tag", "alpha"],
+        ["models", "model_type:checkpoints", "unit-tests", "mf-tag", "alpha"],
         {"epoch": 1},
         make_asset_bytes("alpha"),
     )
     b = asset_factory(
         "mf_tag_beta.safetensors",
-        ["models", "checkpoints", "unit-tests", "mf-tag", "beta"],
+        ["models", "model_type:checkpoints", "unit-tests", "mf-tag", "beta"],
         {"epoch": 2},
         make_asset_bytes("beta"),
     )
@@ -367,7 +367,7 @@ def test_meta_with_tags_include_exclude_and_name_contains(http, api_base, asset_
 
 def test_meta_sort_and_paging_under_filter(http, api_base, asset_factory, make_asset_bytes):
     # Three assets in same scope with different sizes and a common filter key
-    t = ["models", "checkpoints", "unit-tests", "mf-sort"]
+    t = ["models", "model_type:checkpoints", "unit-tests", "mf-sort"]
     n1, n2, n3 = "mf_sort_1.safetensors", "mf_sort_2.safetensors", "mf_sort_3.safetensors"
     asset_factory(n1, t, {"group": "g"}, make_asset_bytes(n1, 1024))
     asset_factory(n2, t, {"group": "g"}, make_asset_bytes(n2, 2048))

tests-unit/assets_test/test_prune_orphaned_assets.py

@@ -29,7 +29,7 @@ def create_seed_file(comfy_tmp_base_dir: Path):
 def find_asset(http: requests.Session, api_base: str):
     """Query API for assets matching scope and optional name."""
     def _find(scope: str, name: str | None = None) -> list[dict]:
-        params = {"include_tags": f"unit-tests,{scope}"}
+        params = {"limit": "500"}
         if name:
             params["name_contains"] = name
         r = http.get(f"{api_base}/api/assets", params=params, timeout=120)
@@ -91,7 +91,7 @@ def test_hashed_asset_not_pruned_when_file_missing(
     data = make_asset_bytes("test", 2048)
     a = asset_factory("test.bin", ["input", "unit-tests", scope], {}, data)
 
-    path = comfy_tmp_base_dir / "input" / "unit-tests" / scope / get_asset_filename(a["asset_hash"], ".bin")
+    path = comfy_tmp_base_dir / "input" / get_asset_filename(a["asset_hash"], ".bin")
     path.unlink()
 
     trigger_sync_seed_assets(http, api_base)
@@ -108,18 +108,20 @@ def test_prune_across_multiple_roots(
 ):
     """Prune correctly handles assets across input and output roots."""
     scope = f"multi-{uuid.uuid4().hex[:6]}"
-    input_fp = create_seed_file("input", scope, "input.bin")
-    create_seed_file("output", scope, "output.bin")
+    input_name = f"{scope}-input.bin"
+    output_name = f"{scope}-output.bin"
+    input_fp = create_seed_file("input", scope, input_name)
+    create_seed_file("output", scope, output_name)
 
     trigger_sync_seed_assets(http, api_base)
-    assert len(find_asset(scope)) == 2
+    assert find_asset(scope, input_name)
+    assert find_asset(scope, output_name)
 
     input_fp.unlink()
     trigger_sync_seed_assets(http, api_base)
 
-    remaining = find_asset(scope)
-    assert len(remaining) == 1
-    assert remaining[0]["name"] == "output.bin"
+    assert not find_asset(scope, input_name)
+    assert find_asset(scope, output_name)
 
 
 @pytest.mark.parametrize("dirname", ["100%_done", "my_folder_name", "has spaces"])

tests-unit/assets_test/test_tags_api.py

@@ -10,9 +10,9 @@ def test_tags_present(http: requests.Session, api_base: str, seeded_asset: dict)
     body1 = r1.json()
     assert r1.status_code == 200
     names = [t["name"] for t in body1["tags"]]
-    # A few system tags from migration should exist:
+    # A few selected contract tags should exist.
     assert "models" in names
-    assert "checkpoints" in names
+    assert "model_type:checkpoints" in names
 
     # Only used tags before we add anything new from this test cycle
     r2 = http.get(api_base + "/api/tags", params={"include_zero": "false"}, timeout=120)
@@ -21,7 +21,7 @@ def test_tags_present(http: requests.Session, api_base: str, seeded_asset: dict)
     # We already seeded one asset via fixture, so used tags must be non-empty
     used_names = [t["name"] for t in body2["tags"]]
     assert "models" in used_names
-    assert "checkpoints" in used_names
+    assert "model_type:checkpoints" in used_names
 
     # Prefix filter should refine the list
     r3 = http.get(api_base + "/api/tags", params={"include_zero": "false", "prefix": "uni"}, timeout=120)
@@ -45,7 +45,7 @@ def test_tags_empty_usage(http: requests.Session, api_base: str, asset_factory,
     body1 = r1.json()
     assert r1.status_code == 200
     names = [t["name"] for t in body1["tags"]]
-    assert "models" in names and "checkpoints" in names
+    assert "models" in names and "model_type:checkpoints" in names
 
     # Create a short-lived asset under input with a unique custom tag
     scope = f"tags-empty-usage-{uuid.uuid4().hex[:6]}"
@@ -89,28 +89,28 @@ def test_tags_empty_usage(http: requests.Session, api_base: str, asset_factory,
 def test_add_and_remove_tags(http: requests.Session, api_base: str, seeded_asset: dict):
     aid = seeded_asset["id"]
 
-    # Add tags with duplicates and mixed case
-    payload_add = {"tags": ["NewTag", "unit-tests", "newtag", "BETA"]}
+    # Add tags with duplicates while preserving source case.
+    payload_add = {"tags": ["NewTag", "unit-tests", "NewTag", "BETA"]}
     r1 = http.post(f"{api_base}/api/assets/{aid}/tags", json=payload_add, timeout=120)
     b1 = r1.json()
     assert r1.status_code == 200, b1
-    # normalized, deduplicated; 'unit-tests' was already present from the seed
-    assert set(b1["added"]) == {"newtag", "beta"}
+    # stripped, deduplicated; 'unit-tests' was already present from the seed
+    assert set(b1["added"]) == {"NewTag", "BETA"}
     assert set(b1["already_present"]) == {"unit-tests"}
-    assert "newtag" in b1["total_tags"] and "beta" in b1["total_tags"]
+    assert "NewTag" in b1["total_tags"] and "BETA" in b1["total_tags"]
 
     rg = http.get(f"{api_base}/api/assets/{aid}", timeout=120)
     g = rg.json()
     assert rg.status_code == 200
     tags_now = set(g["tags"])
-    assert {"newtag", "beta"}.issubset(tags_now)
+    assert {"NewTag", "BETA"}.issubset(tags_now)
 
     # Remove a tag and a non-existent tag
-    payload_del = {"tags": ["newtag", "does-not-exist"]}
+    payload_del = {"tags": ["NewTag", "does-not-exist"]}
     r2 = http.delete(f"{api_base}/api/assets/{aid}/tags", json=payload_del, timeout=120)
     b2 = r2.json()
     assert r2.status_code == 200
-    assert set(b2["removed"]) == {"newtag"}
+    assert set(b2["removed"]) == {"NewTag"}
     assert set(b2["not_present"]) == {"does-not-exist"}
 
     # Verify remaining tags after deletion
@@ -118,8 +118,44 @@ def test_add_and_remove_tags(http: requests.Session, api_base: str, seeded_asset
     g2 = rg2.json()
     assert rg2.status_code == 200
     tags_later = set(g2["tags"])
-    assert "newtag" not in tags_later
-    assert "beta" in tags_later  # still present
+    assert "NewTag" not in tags_later
+    assert "BETA" in tags_later  # still present
+
+
+def test_add_system_looking_tags_allowed_as_labels(
+    http: requests.Session, api_base: str, seeded_asset: dict
+):
+    aid = seeded_asset["id"]
+
+    response = http.post(
+        f"{api_base}/api/assets/{aid}/tags",
+        json={
+            "tags": [
+                "models",
+                "model_type:manual",
+                "model:true",
+                "models:foo",
+                "input:true",
+                "output:true",
+                "uploaded:true",
+                "temp:true",
+                "temporary",
+            ]
+        },
+        timeout=120,
+    )
+    body = response.json()
+
+    assert response.status_code == 200, body
+    assert "models" in body["total_tags"]
+    assert "model_type:manual" in body["total_tags"]
+    assert "model:true" in body["total_tags"]
+    assert "models:foo" in body["total_tags"]
+    assert "input:true" in body["total_tags"]
+    assert "output:true" in body["total_tags"]
+    assert "uploaded:true" in body["total_tags"]
+    assert "temp:true" in body["total_tags"]
+    assert "temporary" in body["total_tags"]
 
 
 def test_tags_list_order_and_prefix(http: requests.Session, api_base: str, seeded_asset: dict):

tests-unit/assets_test/test_uploads.py

@@ -1,11 +1,14 @@
 import json
 import uuid
 from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
 
 import requests
 import pytest
 
+from app.assets.api.schemas_in import UploadAssetSpec
 from app.assets.api.schemas_out import Asset, AssetCreated
+from helpers import get_asset_filename
 
 
 def test_asset_created_inherits_hash_field():
@@ -20,9 +23,18 @@ def test_asset_created_inherits_hash_field():
     assert AssetCreated.model_fields["hash"].annotation == Asset.model_fields["hash"].annotation
 
 
+def test_upload_asset_spec_ignores_subfolder_field():
+    spec = UploadAssetSpec.model_validate(
+        {"tags": ["input"], "subfolder": "pasted", "name": "image.png"}
+    )
+
+    assert "subfolder" not in UploadAssetSpec.model_fields
+    assert not hasattr(spec, "subfolder")
+
+
 def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, make_asset_bytes):
     name = "dup_a.safetensors"
-    tags = ["models", "checkpoints", "unit-tests", "alpha"]
+    tags = ["models", "model_type:checkpoints", "unit-tests", "alpha"]
     meta = {"purpose": "dup"}
     data = make_asset_bytes(name)
     files = {"file": (name, data, "application/octet-stream")}
@@ -43,6 +55,8 @@ def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, ma
     assert a2["asset_hash"] == a1["asset_hash"]
     assert a2["hash"] == a1["hash"]
     assert a2["id"] != a1["id"]  # new reference with same content
+    assert a2.get("loader_path") is None
+    assert a2.get("display_name") is None
 
     # Third upload with the same data but different name also creates new AssetReference
     files = {"file": (name, data, "application/octet-stream")}
@@ -53,12 +67,14 @@ def test_upload_ok_duplicate_reference(http: requests.Session, api_base: str, ma
     assert a3["asset_hash"] == a1["asset_hash"]
     assert a3["id"] != a1["id"]
     assert a3["id"] != a2["id"]
+    assert a3.get("loader_path") is None
+    assert a3.get("display_name") is None
 
 
 def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_base: str):
     # Seed a small file first
     name = "fastpath_seed.safetensors"
-    tags = ["models", "checkpoints", "unit-tests"]
+    tags = ["input", "unit-tests"]
     meta = {}
     files = {"file": (name, b"B" * 1024, "application/octet-stream")}
     form = {"tags": json.dumps(tags), "name": name, "user_metadata": json.dumps(meta)}
@@ -69,9 +85,10 @@ def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_
     assert b1["hash"] == h
 
     # Now POST /api/assets with only hash and no file
+    hash_only_tags = ["models", "checkpoints", "unit-tests", "hash-labels"]
     files = [
         ("hash", (None, h)),
-        ("tags", (None, json.dumps(tags))),
+        ("tags", (None, json.dumps(hash_only_tags))),
         ("name", (None, "fastpath_copy.safetensors")),
         ("user_metadata", (None, json.dumps({"purpose": "copy"}))),
     ]
@@ -81,6 +98,53 @@ def test_upload_fastpath_from_existing_hash_no_file(http: requests.Session, api_
     assert b2["created_new"] is False
     assert b2["asset_hash"] == h
     assert b2["hash"] == h
+    assert "models" in b2["tags"]
+    assert "checkpoints" in b2["tags"]
+    assert "uploaded" not in b2["tags"]
+    assert not any(tag.startswith("model_type:") for tag in b2["tags"])
+    assert b2.get("loader_path") is None
+    assert b2.get("display_name") is None
+
+    rg = http.get(f"{api_base}/api/assets/{b2['id']}", timeout=120)
+    detail = rg.json()
+    assert rg.status_code == 200, detail
+    assert detail.get("loader_path") is None
+    assert detail.get("display_name") is None
+
+
+def test_create_from_hash_with_model_tags_does_not_synthesize_loader_path(
+    http: requests.Session, api_base: str
+):
+    seed_name = "from_hash_seed.safetensors"
+    seed_tags = ["models", "model_type:checkpoints", "unit-tests"]
+    files = {"file": (seed_name, b"D" * 1024, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(seed_tags),
+        "name": seed_name,
+        "user_metadata": json.dumps({}),
+    }
+    seed_r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    seed = seed_r.json()
+    assert seed_r.status_code == 201, seed
+
+    payload = {
+        "hash": seed["asset_hash"],
+        "name": "from_hash_copy.safetensors",
+        "tags": ["models", "model_type:checkpoints", "unit-tests", "spoofed"],
+    }
+    created_r = http.post(api_base + "/api/assets/from-hash", json=payload, timeout=120)
+    created = created_r.json()
+    assert created_r.status_code == 201, created
+    assert created["created_new"] is False
+    assert created["asset_hash"] == seed["asset_hash"]
+    assert created.get("loader_path") is None
+    assert created.get("display_name") is None
+
+    detail_r = http.get(f"{api_base}/api/assets/{created['id']}", timeout=120)
+    detail = detail_r.json()
+    assert detail_r.status_code == 200, detail
+    assert detail.get("loader_path") is None
+    assert detail.get("display_name") is None
 
 
 def test_upload_fastpath_with_known_hash_and_file(
@@ -88,7 +152,7 @@ def test_upload_fastpath_with_known_hash_and_file(
 ):
     # Seed
     files = {"file": ("seed.safetensors", b"C" * 128, "application/octet-stream")}
-    form = {"tags": json.dumps(["models", "checkpoints", "unit-tests", "fp"]), "name": "seed.safetensors", "user_metadata": json.dumps({})}
+    form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "fp"]), "name": "seed.safetensors", "user_metadata": json.dumps({})}
     r1 = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
     b1 = r1.json()
     assert r1.status_code == 201, b1
@@ -104,11 +168,49 @@ def test_upload_fastpath_with_known_hash_and_file(
     assert b2["created_new"] is False
     assert b2["asset_hash"] == h
     assert b2["hash"] == h
+    assert "checkpoints" in b2["tags"]
+    assert "uploaded" not in b2["tags"]
+    assert not any(tag == "model_type:checkpoints" for tag in b2["tags"])
+
+
+def test_duplicate_byte_upload_is_reference_only_and_does_not_need_destination(
+    http: requests.Session, api_base: str
+):
+    data = b"duplicate-reference-only" * 64
+    seed_files = {"file": ("duplicate-seed.bin", data, "application/octet-stream")}
+    seed_form = {
+        "tags": json.dumps(["input", "unit-tests", "duplicate-seed"]),
+        "name": "duplicate-seed.bin",
+        "user_metadata": json.dumps({}),
+    }
+    seed_response = http.post(api_base + "/api/assets", data=seed_form, files=seed_files, timeout=120)
+    seed = seed_response.json()
+    assert seed_response.status_code == 201, seed
+
+    duplicate_files = {"file": ("duplicate-copy.bin", data, "application/octet-stream")}
+    duplicate_form = {
+        "tags": json.dumps(["not-a-destination", "unit-tests", "duplicate-copy"]),
+        "name": "duplicate-copy.bin",
+        "user_metadata": json.dumps({}),
+    }
+    duplicate_response = http.post(
+        api_base + "/api/assets", data=duplicate_form, files=duplicate_files, timeout=120
+    )
+    duplicate = duplicate_response.json()
+
+    assert duplicate_response.status_code == 200, duplicate
+    assert duplicate["created_new"] is False
+    assert duplicate["asset_hash"] == seed["asset_hash"]
+    assert "not-a-destination" in duplicate["tags"]
+    assert "uploaded" not in duplicate["tags"]
+    assert "input" not in duplicate["tags"]
+    assert duplicate.get("loader_path") is None
+    assert duplicate.get("display_name") is None
 
 
 def test_upload_multiple_tags_fields_are_merged(http: requests.Session, api_base: str):
     data = [
-        ("tags", "models,checkpoints"),
+        ("tags", "models,model_type:checkpoints"),
         ("tags", json.dumps(["unit-tests", "alpha"])),
         ("name", "merge.safetensors"),
         ("user_metadata", json.dumps({"u": 1})),
@@ -124,7 +226,71 @@ def test_upload_multiple_tags_fields_are_merged(http: requests.Session, api_base
     detail = rg.json()
     assert rg.status_code == 200, detail
     tags = set(detail["tags"])
-    assert {"models", "checkpoints", "unit-tests", "alpha"}.issubset(tags)
+    assert {"models", "model_type:checkpoints", "unit-tests", "alpha"}.issubset(tags)
+
+
+@pytest.mark.parametrize(
+    (
+        "tags",
+        "extension",
+        "expected_display_prefix",
+    ),
+    [
+        (["input", "unit-tests"], ".png", ""),
+        (
+            ["models", "model_type:checkpoints", "unit-tests"],
+            ".safetensors",
+            "checkpoints/",
+        ),
+    ],
+)
+def test_upload_response_includes_loader_path_and_display_name(
+    tags: list[str],
+    extension: str,
+    expected_display_prefix: str,
+    http: requests.Session,
+    api_base: str,
+    make_asset_bytes,
+):
+    scope = f"response-paths-{uuid.uuid4().hex[:6]}"
+    scoped_tags = [*tags, scope]
+    name = f"asset_response_path{extension}"
+
+    files = {"file": (name, make_asset_bytes(name, 1024), "application/octet-stream")}
+    form = {
+        "tags": json.dumps(scoped_tags),
+        "name": name,
+        "user_metadata": json.dumps({}),
+    }
+    created_r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    created = created_r.json()
+    assert created_r.status_code in (200, 201), created
+    stored_filename = get_asset_filename(created["asset_hash"], extension)
+    expected_suffix = stored_filename
+    expected_display_name = f"{expected_display_prefix}{expected_suffix}"
+    # In-root loader path: model category dropped, no subfolders here -> just the filename.
+    expected_loader_path = expected_suffix
+
+    assert created["loader_path"] == expected_loader_path
+    assert created["display_name"] == expected_display_name
+    assert "logical_path" not in created
+
+    detail_r = http.get(f"{api_base}/api/assets/{created['id']}", timeout=120)
+    detail = detail_r.json()
+    assert detail_r.status_code == 200, detail
+    assert detail["loader_path"] == expected_loader_path
+    assert detail["display_name"] == expected_display_name
+
+    list_r = http.get(
+        api_base + "/api/assets",
+        params={"include_tags": f"unit-tests,{scope}", "limit": "50"},
+        timeout=120,
+    )
+    listed = list_r.json()
+    assert list_r.status_code == 200, listed
+    match = next(a for a in listed["assets"] if a["id"] == created["id"])
+    assert match["loader_path"] == expected_loader_path
+    assert match["display_name"] == expected_display_name
 
 
 @pytest.mark.parametrize("root", ["input", "output"])
@@ -192,16 +358,55 @@ def test_create_from_hash_endpoint_404(http: requests.Session, api_base: str):
     assert body["error"]["code"] == "ASSET_NOT_FOUND"
 
 
+def test_create_from_hash_accepts_arbitrary_system_looking_tags(
+    http: requests.Session, api_base: str
+):
+    files = {"file": ("hash-seed.bin", b"hash-seed" * 64, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(["input", "unit-tests", "hash-seed"]),
+        "name": "hash-seed.bin",
+        "user_metadata": json.dumps({}),
+    }
+    seed_response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    seed = seed_response.json()
+    assert seed_response.status_code == 201, seed
+
+    response = http.post(
+        api_base + "/api/assets/from-hash",
+        json={
+            "hash": seed["asset_hash"],
+            "name": "hash-copy.bin",
+            "tags": [
+                "models",
+                "model:true",
+                "models:foo",
+                "temporary:true",
+                "unit-tests",
+                "hash-copy",
+            ],
+        },
+        timeout=120,
+    )
+    body = response.json()
+
+    assert response.status_code == 201, body
+    assert "models" in body["tags"]
+    assert "model:true" in body["tags"]
+    assert "models:foo" in body["tags"]
+    assert "temporary:true" in body["tags"]
+    assert "uploaded" not in body["tags"]
+
+
 def test_upload_zero_byte_rejected(http: requests.Session, api_base: str):
     files = {"file": ("empty.safetensors", b"", "application/octet-stream")}
-    form = {"tags": json.dumps(["models", "checkpoints", "unit-tests", "edge"]), "name": "empty.safetensors", "user_metadata": json.dumps({})}
+    form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "edge"]), "name": "empty.safetensors", "user_metadata": json.dumps({})}
     r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
     body = r.json()
     assert r.status_code == 400
     assert body["error"]["code"] == "EMPTY_UPLOAD"
 
 
-def test_upload_invalid_root_tag_rejected(http: requests.Session, api_base: str):
+def test_upload_rejects_arbitrary_labels_without_required_destination_role(http: requests.Session, api_base: str):
     files = {"file": ("badroot.bin", b"A" * 64, "application/octet-stream")}
     form = {"tags": json.dumps(["not-a-root", "whatever"]), "name": "badroot.bin", "user_metadata": json.dumps({})}
     r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
@@ -212,7 +417,7 @@ def test_upload_invalid_root_tag_rejected(http: requests.Session, api_base: str)
 
 def test_upload_user_metadata_must_be_json(http: requests.Session, api_base: str):
     files = {"file": ("badmeta.bin", b"A" * 128, "application/octet-stream")}
-    form = {"tags": json.dumps(["models", "checkpoints", "unit-tests", "edge"]), "name": "badmeta.bin", "user_metadata": "{not json}"}
+    form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "edge"]), "name": "badmeta.bin", "user_metadata": "{not json}"}
     r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
     body = r.json()
     assert r.status_code == 400
@@ -228,7 +433,7 @@ def test_upload_requires_multipart(http: requests.Session, api_base: str):
 
 def test_upload_missing_file_and_hash(http: requests.Session, api_base: str):
     files = [
-        ("tags", (None, json.dumps(["models", "checkpoints", "unit-tests"]))),
+        ("tags", (None, json.dumps(["models", "model_type:checkpoints", "unit-tests"]))),
         ("name", (None, "x.safetensors")),
     ]
     r = http.post(api_base + "/api/assets", files=files, timeout=120)
@@ -237,17 +442,33 @@ def test_upload_missing_file_and_hash(http: requests.Session, api_base: str):
     assert body["error"]["code"] == "MISSING_FILE"
 
 
-def test_upload_models_unknown_category(http: requests.Session, api_base: str):
+def test_upload_models_unknown_model_type(http: requests.Session, api_base: str):
     files = {"file": ("m.safetensors", b"A" * 128, "application/octet-stream")}
-    form = {"tags": json.dumps(["models", "no_such_category", "unit-tests"]), "name": "m.safetensors"}
+    form = {"tags": json.dumps(["models", "model_type:no_such_category", "unit-tests"]), "name": "m.safetensors"}
     r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
     body = r.json()
-    assert r.status_code == 400
+    assert r.status_code == 400, body
     assert body["error"]["code"] == "INVALID_BODY"
-    assert body["error"]["message"].startswith("unknown models category")
 
 
-def test_upload_models_requires_category(http: requests.Session, api_base: str):
+@pytest.mark.parametrize("model_type", ["configs", "custom_nodes"])
+def test_upload_models_rejects_non_model_registered_folder(
+    model_type: str, http: requests.Session, api_base: str
+):
+    files = {"file": ("not-a-model.py", b"A" * 128, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(["models", f"model_type:{model_type}", "unit-tests"]),
+        "name": "not-a-model.py",
+    }
+
+    response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    body = response.json()
+
+    assert response.status_code == 400, body
+    assert body["error"]["code"] == "INVALID_BODY"
+
+
+def test_upload_models_requires_model_type(http: requests.Session, api_base: str):
     files = {"file": ("nocat.safetensors", b"A" * 64, "application/octet-stream")}
     form = {"tags": json.dumps(["models"]), "name": "nocat.safetensors", "user_metadata": json.dumps({})}
     r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
@@ -256,13 +477,152 @@ def test_upload_models_requires_category(http: requests.Session, api_base: str):
     assert body["error"]["code"] == "INVALID_BODY"
 
 
-def test_upload_tags_traversal_guard(http: requests.Session, api_base: str):
+def test_upload_extra_tags_are_labels_not_path_components(http: requests.Session, api_base: str):
     files = {"file": ("evil.safetensors", b"A" * 256, "application/octet-stream")}
-    form = {"tags": json.dumps(["models", "checkpoints", "unit-tests", "..", "zzz"]), "name": "evil.safetensors"}
+    form = {"tags": json.dumps(["models", "model_type:checkpoints", "unit-tests", "..", "zzz"]), "name": "evil.safetensors"}
     r = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
     body = r.json()
-    assert r.status_code == 400
-    assert body["error"]["code"] in ("BAD_REQUEST", "INVALID_BODY")
+    assert r.status_code == 201, body
+    assert ".." in body["tags"]
+    assert "zzz" in body["tags"]
+    assert "models" in body["tags"]
+    assert "model_type:checkpoints" in body["tags"]
+
+
+@pytest.mark.parametrize(
+    ("subfolder", "expected_tag", "unexpected_tags"),
+    [
+        ("custom/session", None, {"custom", "session"}),
+        ("pasted", "pasted", set()),
+    ],
+)
+def test_upload_image_accepts_arbitrary_subfolder_but_only_known_values_become_tags(
+    http: requests.Session,
+    api_base: str,
+    comfy_tmp_base_dir: Path,
+    subfolder: str,
+    expected_tag: str | None,
+    unexpected_tags: set[str],
+):
+    name = f"upload-image-{uuid.uuid4().hex}.png"
+    files = {"image": (name, b"image-upload" * 64, "image/png")}
+    form = {"type": "input", "subfolder": subfolder}
+
+    response = http.post(api_base + "/upload/image", data=form, files=files, timeout=120)
+    body = response.json()
+
+    assert response.status_code == 200, body
+    assert body["subfolder"] == subfolder
+    assert (comfy_tmp_base_dir / "input" / subfolder / body["name"]).exists()
+
+    asset = body["asset"]
+    tags = set(asset["tags"])
+    assert "input" in tags
+    assert "uploaded" in tags
+    if expected_tag:
+        assert expected_tag in tags
+    assert tags.isdisjoint(unexpected_tags)
+
+
+def test_multipart_upload_accepts_system_looking_extra_labels(
+    http: requests.Session, api_base: str
+):
+    files = {"file": ("relaxed-labels.bin", b"relaxed" * 64, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(
+            [
+                "input",
+                "unit-tests",
+                "model:true",
+                "models:foo",
+                "temporary",
+                "uploaded:true",
+            ]
+        ),
+        "name": "relaxed-labels.bin",
+        "user_metadata": json.dumps({}),
+    }
+    response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    body = response.json()
+
+    assert response.status_code == 201, body
+    assert "input" in body["tags"]
+    assert "model:true" in body["tags"]
+    assert "models:foo" in body["tags"]
+    assert "temporary" in body["tags"]
+    assert "uploaded:true" in body["tags"]
+
+
+def test_multipart_upload_rejects_ambiguous_destination_roles(
+    http: requests.Session, api_base: str
+):
+    files = {"file": ("ambiguous.bin", b"ambiguous" * 64, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(["input", "output", "unit-tests"]),
+        "name": "ambiguous.bin",
+        "user_metadata": json.dumps({}),
+    }
+    response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    body = response.json()
+
+    assert response.status_code == 400, body
+    assert body["error"]["code"] == "INVALID_BODY"
+
+
+def test_multipart_upload_rejects_multiple_model_types_for_models_destination(
+    http: requests.Session, api_base: str
+):
+    files = {"file": ("ambiguous-model.safetensors", b"ambiguous-model" * 64, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(
+            ["models", "model_type:checkpoints", "model_type:loras", "unit-tests"]
+        ),
+        "name": "ambiguous-model.safetensors",
+        "user_metadata": json.dumps({}),
+    }
+    response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    body = response.json()
+
+    assert response.status_code == 400, body
+    assert body["error"]["code"] == "INVALID_BODY"
+
+
+@pytest.mark.parametrize(
+    ("tags", "expected_root", "extension"),
+    [
+        (["input", "unit-tests", "upload-location-input"], "input", ".bin"),
+        (["output", "unit-tests", "upload-location-output"], "output", ".bin"),
+        (
+            ["models", "model_type:checkpoints", "unit-tests", "upload-location-model"],
+            "models/checkpoints",
+            ".safetensors",
+        ),
+    ],
+)
+def test_multipart_upload_role_selects_write_location(
+    http: requests.Session,
+    api_base: str,
+    comfy_tmp_base_dir: Path,
+    tags: list[str],
+    expected_root: str,
+    extension: str,
+):
+    role = next(tag for tag in tags if tag in {"input", "models", "output"})
+    name = f"{role}-role-upload{extension}"
+    files = {"file": (name, f"{role}-role-bytes".encode() * 64, "application/octet-stream")}
+    form = {
+        "tags": json.dumps(tags),
+        "name": name,
+        "user_metadata": json.dumps({}),
+    }
+
+    response = http.post(api_base + "/api/assets", data=form, files=files, timeout=120)
+    body = response.json()
+
+    assert response.status_code == 201, body
+    stored_name = get_asset_filename(body["asset_hash"], extension)
+    expected_disk_path = comfy_tmp_base_dir / expected_root / stored_name
+    assert expected_disk_path.exists()
 
 
 def test_upload_empty_tags_rejected(http: requests.Session, api_base: str):

tests-unit/comfy_test/folder_path_test.py

@@ -163,3 +163,20 @@ def test_base_path_change_clears_old(set_base_dir):
 
     for name in ["controlnet", "diffusion_models", "text_encoders"]:
         assert len(folder_paths.get_folder_paths(name)) == 2
+
+
+def test_models_directory_cli_and_getters(temp_dir):
+    try:
+        with patch.object(sys, 'argv', ["main.py", "--models-directory", temp_dir]):
+            reload(comfy.cli_args)
+            reload(folder_paths)
+
+        assert folder_paths.models_dir == os.path.abspath(temp_dir)
+
+        with pytest.raises(Exception):
+            comfy.cli_args.is_valid_directory(os.path.join(temp_dir, "non_existent_folder_path"))
+    finally:
+        with patch.object(sys, 'argv', ["main.py"]):
+            reload(comfy.cli_args)
+            reload(folder_paths)
+

tests-unit/feature_flags_test.py

@@ -29,6 +29,8 @@ class TestFeatureFlags:
         features = get_server_features()
         assert "supports_preview_metadata" in features
         assert features["supports_preview_metadata"] is True
+        assert "supports_model_type_tags" in features
+        assert features["supports_model_type_tags"] is True
         assert "max_upload_size" in features
         assert isinstance(features["max_upload_size"], (int, float))
 

tests-unit/websocket_feature_flags_test.py

@@ -12,6 +12,8 @@ class TestWebSocketFeatureFlags:
         # Check expected server features
         assert "supports_preview_metadata" in features
         assert features["supports_preview_metadata"] is True
+        assert "supports_model_type_tags" in features
+        assert features["supports_model_type_tags"] is True
         assert "max_upload_size" in features
         assert isinstance(features["max_upload_size"], (int, float))
 
@@ -75,3 +77,5 @@ class TestWebSocketFeatureFlags:
         assert server_message["type"] == "feature_flags"
         assert "supports_preview_metadata" in server_message["data"]
         assert server_message["data"]["supports_preview_metadata"] is True
+        assert "supports_model_type_tags" in server_message["data"]
+        assert server_message["data"]["supports_model_type_tags"] is True