Merge ca5adea2e3 into 51bf508a0b

* fix(Video): don't crash on videos with undecodable audio streams

Signed-off-by: bigcat88 <bigcat88@icloud.com>

* Update comfy_api_nodes/util/upload_helpers.py

---------

Signed-off-by: bigcat88 <bigcat88@icloud.com>
Co-authored-by: Alexis Rolland <alexisrolland@hotmail.com>

.github/workflows/cla.yml

@@ -0,0 +1,91 @@
+name: CLA Assistant
+
+on:
+  issue_comment:
+    types: [created]
+  pull_request_target:
+    types: [opened, synchronize, closed]
+
+permissions:
+  actions: write
+  contents: read # 'read' is enough because signatures live in a REMOTE repo
+  pull-requests: write
+  statuses: write
+
+jobs:
+  cla-assistant:
+    runs-on: ubuntu-latest
+    steps:
+      # The CLA action normally requires every commit author in a PR to sign.
+      # We only want the PR author to sign, so we allowlist all other committers
+      # by computing them from the PR's commits and excluding the PR author.
+      - name: Build author-only allowlist
+        id: allowlist
+        if: >
+          github.event_name == 'pull_request_target' ||
+          (github.event_name == 'issue_comment' && github.event.issue.pull_request && (
+            github.event.comment.body == 'recheck' ||
+            github.event.comment.body == 'I have read and agree to the Contributor License Agreement'
+          ))
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number }}
+          PR_AUTHOR: ${{ github.event.pull_request.user.login || github.event.issue.user.login }}
+          BASE_ALLOWLIST: action@github.com,actions-user,ampagent,claude,comfy-pr-bot,GitHub Action,github-actions,github-actions[bot],Glary Bot,Glary-Bot,*[bot]
+        run: |
+          others=$(gh api "repos/${{ github.repository }}/pulls/${PR_NUMBER}/commits" --paginate \
+            --jq '.[] | (.author.login // empty), (.committer.login // empty)' \
+            | sort -u | grep -vix "${PR_AUTHOR}" | paste -sd, -)
+          if [ -n "$others" ]; then
+            echo "allowlist=${BASE_ALLOWLIST},${others}" >> "$GITHUB_OUTPUT"
+          else
+            echo "allowlist=${BASE_ALLOWLIST}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: CLA Assistant
+        # Run on PR events, on "recheck" comment, or when someone posts the exact signing phrase.
+        # IMPORTANT: this phrase must match `custom-pr-sign-comment` below.
+        if: >
+          github.event_name == 'pull_request_target' ||
+          (github.event_name == 'issue_comment' && github.event.issue.pull_request && (
+            github.event.comment.body == 'recheck' ||
+            github.event.comment.body == 'I have read and agree to the Contributor License Agreement'
+          ))
+        uses: contributor-assistant/github-action@ca4a40a7d1004f18d9960b404b97e5f30a505a08 # v2.6.1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          # PAT required to write to the centralized signatures repo.
+          PERSONAL_ACCESS_TOKEN: ${{ secrets.PERSONAL_ACCESS_TOKEN }}
+        with:
+          # Where the CLA document lives (shown to contributors)
+          path-to-document: https://github.com/Comfy-Org/comfy-cla/blob/main/comfyui_icla.md
+
+          # Centralized signature storage
+          remote-organization-name: comfy-org
+          remote-repository-name: comfy-cla
+          path-to-signatures: signatures/cla.json
+          branch: main
+
+          # Only the PR author must sign: bots plus every non-author committer
+          # are allowlisted via the "Build author-only allowlist" step above.
+          # *[bot] is a catch-all for any GitHub App bot account.
+          allowlist: ${{ steps.allowlist.outputs.allowlist }}
+
+          # Custom PR comment messages
+          custom-notsigned-prcomment: |
+            🎉 Thank you for your contribution, we really appreciate it! 🎉
+
+            Like many open source projects, we require contributors to sign our [Contributor License Agreement (CLA)](https://github.com/Comfy-Org/comfy-cla/blob/main/comfyui_icla.md). A CLA makes the ownership of contributions explicit, so contributors and the project share a clear understanding of how the code can be used. By signing, you:
+
+            - Confirm that you own your contribution.
+            - Keep the right to reuse your own code.
+            - Grant us a copyright license to include and share it within our projects.
+
+            CLAs are standard practice across major open source projects including those under the Apache Software Foundation and the Linux Foundation. Ours is based on the Apache Software Foundation's CLA. Most importantly, it would enable us to relicense the project under a more permissive license in the future, giving the project and its community greater flexibility.
+
+            ✍ **To sign, please post a new comment on this PR with exactly the following text:** ✍
+
+          custom-pr-sign-comment: I have read and agree to the Contributor License Agreement
+
+          custom-allsigned-prcomment: |
+            ✅ All contributors have signed the CLA. Thank you! This PR is ready to be merged.

CLAUDE.md

@@ -1 +0,0 @@
-AGENTS.md

alembic_db/versions/0005_allow_case_sensitive_tags.py

@@ -0,0 +1,107 @@
+"""
+Allow case-sensitive tag names.
+
+Revision ID: 0005_allow_case_sensitive_tags
+Revises: 0004_drop_tag_type
+Create Date: 2026-06-16
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0005_allow_case_sensitive_tags"
+down_revision = "0004_drop_tag_type"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name == "sqlite":
+        # SQLite cannot ALTER/DROP CHECK constraints. Recreate the small tag
+        # vocabulary table without the lowercase constraint while preserving
+        # existing tag names.
+        op.execute("PRAGMA foreign_keys=OFF")
+        try:
+            op.execute(
+                "CREATE TABLE tags_new ("
+                "name VARCHAR(512) NOT NULL, "
+                "CONSTRAINT pk_tags PRIMARY KEY (name)"
+                ")"
+            )
+            op.execute("INSERT INTO tags_new(name) SELECT name FROM tags")
+            op.execute("DROP TABLE tags")
+            op.execute("ALTER TABLE tags_new RENAME TO tags")
+        finally:
+            op.execute("PRAGMA foreign_keys=ON")
+        return
+
+    op.drop_constraint("ck_tags_ck_tags_lowercase", "tags", type_="check")
+
+
+def downgrade() -> None:
+    # Existing mixed-case tags cannot satisfy the old constraint. Lowercase them
+    # before restoring it, merging duplicate vocabulary/link rows that collide.
+    bind = op.get_bind()
+
+    tag_names = [row[0] for row in bind.execute(sa.text("SELECT name FROM tags"))]
+    existing_names = set(tag_names)
+    lowercase_names = sorted({name.lower() for name in tag_names})
+    missing_lowercase_rows = [
+        {"name": name} for name in lowercase_names if name not in existing_names
+    ]
+    if missing_lowercase_rows:
+        bind.execute(sa.text("INSERT INTO tags(name) VALUES (:name)"), missing_lowercase_rows)
+
+    link_rows = bind.execute(
+        sa.text(
+            "SELECT asset_reference_id, tag_name, origin, added_at "
+            "FROM asset_reference_tags "
+            "ORDER BY asset_reference_id, tag_name"
+        )
+    ).mappings()
+    deduped_links = {}
+    for row in link_rows:
+        key = (row["asset_reference_id"], row["tag_name"].lower())
+        deduped_links.setdefault(
+            key,
+            {
+                "asset_reference_id": row["asset_reference_id"],
+                "tag_name": row["tag_name"].lower(),
+                "origin": row["origin"],
+                "added_at": row["added_at"],
+            },
+        )
+
+    op.execute("DELETE FROM asset_reference_tags")
+    if deduped_links:
+        bind.execute(
+            sa.text(
+                "INSERT INTO asset_reference_tags "
+                "(asset_reference_id, tag_name, origin, added_at) "
+                "VALUES (:asset_reference_id, :tag_name, :origin, :added_at)"
+            ),
+            list(deduped_links.values()),
+        )
+    op.execute("DELETE FROM tags WHERE name != lower(name)")
+
+    if bind.dialect.name == "sqlite":
+        op.execute("PRAGMA foreign_keys=OFF")
+        try:
+            op.execute(
+                "CREATE TABLE tags_new ("
+                "name VARCHAR(512) NOT NULL, "
+                "CONSTRAINT pk_tags PRIMARY KEY (name), "
+                "CONSTRAINT ck_tags_lowercase CHECK (name = lower(name))"
+                ")"
+            )
+            op.execute("INSERT INTO tags_new(name) SELECT name FROM tags")
+            op.execute("DROP TABLE tags")
+            op.execute("ALTER TABLE tags_new RENAME TO tags")
+        finally:
+            op.execute("PRAGMA foreign_keys=ON")
+        return
+
+    op.create_check_constraint(
+        "ck_tags_ck_tags_lowercase", "tags", "name = lower(name)"
+    )

app/assets/api/routes.py

@@ -10,7 +10,6 @@ from typing import Any
 from aiohttp import web
 from pydantic import ValidationError
 
-import folder_paths
 from app import user_manager
 from app.assets.api import schemas_in, schemas_out
 from app.assets.services import schemas
@@ -40,6 +39,7 @@ from app.assets.services import (
     upload_from_temp_path,
 )
 from app.assets.services.cursor import InvalidCursorError
+from app.assets.services.path_utils import compute_asset_response_paths
 from app.assets.services.tagging import list_tag_histogram
 
 ROUTES = web.RouteTableDef()
@@ -161,11 +161,18 @@ def _build_asset_response(result: schemas.AssetDetailResult | schemas.UploadResu
             preview_url = None
     else:
         preview_url = _build_preview_url_from_view(result.tags, result.ref.user_metadata)
+    if result.ref.file_path:
+        paths = compute_asset_response_paths(result.ref.file_path)
+        file_path, display_name = paths if paths else (None, None)
+    else:
+        file_path, display_name = None, None
     asset_content_hash = result.asset.hash if result.asset else None
     return schemas_out.Asset(
         id=result.ref.id,
         name=result.ref.name,
         hash=asset_content_hash,
+        file_path=file_path,
+        display_name=display_name,
         asset_hash=asset_content_hash,
         size=int(result.asset.size_bytes) if result.asset else None,
         mime_type=result.asset.mime_type if result.asset else None,
@@ -419,17 +426,6 @@ async def upload_asset(request: web.Request) -> web.Response:
             400, "INVALID_BODY", f"Validation failed: {ve.json()}"
         )
 
-    if spec.tags and spec.tags[0] == "models":
-        if (
-            len(spec.tags) < 2
-            or spec.tags[1] not in folder_paths.folder_names_and_paths
-        ):
-            delete_temp_file_if_exists(parsed.tmp_path)
-            category = spec.tags[1] if len(spec.tags) >= 2 else ""
-            return _build_error_response(
-                400, "INVALID_BODY", f"unknown models category '{category}'"
-            )
-
     try:
         # Fast path: hash exists, create AssetReference without writing anything
         if spec.hash and parsed.provided_hash_exists is True:
@@ -473,7 +469,7 @@ async def upload_asset(request: web.Request) -> web.Response:
         return _build_error_response(400, e.code, str(e))
     except ValueError as e:
         delete_temp_file_if_exists(parsed.tmp_path)
-        return _build_error_response(400, "BAD_REQUEST", str(e))
+        return _build_error_response(400, "INVALID_BODY", str(e))
     except HashMismatchError as e:
         delete_temp_file_if_exists(parsed.tmp_path)
         return _build_error_response(400, "HASH_MISMATCH", str(e))

app/assets/api/schemas_in.py

@@ -140,7 +140,7 @@ class CreateFromHashBody(BaseModel):
         if v is None:
             return []
         if isinstance(v, list):
-            out = [str(t).strip().lower() for t in v if str(t).strip()]
+            out = [str(t).strip() for t in v if str(t).strip()]
             seen = set()
             dedup = []
             for t in out:
@@ -149,7 +149,7 @@ class CreateFromHashBody(BaseModel):
                     dedup.append(t)
             return dedup
         if isinstance(v, str):
-            return [t.strip().lower() for t in v.split(",") if t.strip()]
+            return list(dict.fromkeys(t.strip() for t in v.split(",") if t.strip()))
         return []
 
 
@@ -206,7 +206,7 @@ class TagsListQuery(BaseModel):
         if v is None:
             return v
         v = v.strip()
-        return v.lower() or None
+        return v or None
 
 
 class TagsAdd(BaseModel):
@@ -220,7 +220,7 @@ class TagsAdd(BaseModel):
         for t in v:
             if not isinstance(t, str):
                 raise TypeError("tags must be strings")
-            tnorm = t.strip().lower()
+            tnorm = t.strip()
             if tnorm:
                 out.append(tnorm)
         seen = set()
@@ -239,8 +239,8 @@ class TagsRemove(TagsAdd):
 class UploadAssetSpec(BaseModel):
     """Upload Asset operation.
 
-    - tags: optional list; if provided, first is root ('models'|'input'|'output');
-            if root == 'models', second must be a valid category
+    - tags: labels plus one destination role ('models'|'input'|'output') for new bytes;
+            if role == 'models', exactly one model_type:<folder_name> tag is required
     - name: display name
     - user_metadata: arbitrary JSON object (optional)
     - hash: optional canonical 'blake3:<hex>' for validation / fast-path
@@ -309,7 +309,7 @@ class UploadAssetSpec(BaseModel):
         norm = []
         seen = set()
         for t in items:
-            tnorm = str(t).strip().lower()
+            tnorm = str(t).strip()
             if tnorm and tnorm not in seen:
                 seen.add(tnorm)
                 norm.append(tnorm)
@@ -335,14 +335,4 @@ class UploadAssetSpec(BaseModel):
 
     @model_validator(mode="after")
     def _validate_order(self):
-        if not self.tags:
-            raise ValueError("at least one tag is required for uploads")
-        root = self.tags[0]
-        if root not in {"models", "input", "output"}:
-            raise ValueError("first tag must be one of: models, input, output")
-        if root == "models":
-            if len(self.tags) < 2:
-                raise ValueError(
-                    "models uploads require a category tag as the second tag"
-                )
         return self

app/assets/api/schemas_out.py

@@ -9,8 +9,20 @@ class Asset(BaseModel):
     ``id`` here is the AssetReference id, not the content-addressed Asset id."""
 
     id: str
-    name: str
+    name: str = Field(
+        ...,
+        deprecated=True,
+        description="Reference label, often caller-provided or derived from the filename. Deprecated for storage path/display semantics; use `file_path` and `display_name` when present.",
+    )
     hash: str | None = None
+    file_path: str | None = Field(
+        default=None,
+        description="Runtime storage locator for filesystem-backed assets, using Comfy storage namespaces such as `input/`, `output/`, `temp/`, or `models/`. Not an absolute filesystem path, unique identity, or model loader path.",
+    )
+    display_name: str | None = Field(
+        default=None,
+        description="Human-facing label derived from `file_path`, usually the path below the top-level storage namespace. Not unique.",
+    )
     asset_hash: str | None = None
     size: int | None = None
     mime_type: str | None = None

app/assets/api/upload.py

@@ -140,7 +140,6 @@ async def parse_multipart_upload(
             provided_mime_type = ((await field.text()) or "").strip() or None
         elif fname == "preview_id":
             provided_preview_id = ((await field.text()) or "").strip() or None
-
     if not file_present and not (provided_hash and provided_hash_exists):
         raise UploadError(
             400, "MISSING_FILE", "Form must include a 'file' part or a known 'hash'."

app/assets/database/queries/tags.py

@@ -265,6 +265,8 @@ def list_tags_with_usage(
     order: str = "count_desc",
     owner_id: str = "",
 ) -> tuple[list[tuple[str, str, int]], int]:
+    prefix_filter = prefix.strip() if prefix else ""
+
     counts_sq = (
         select(
             AssetReferenceTag.tag_name.label("tag_name"),
@@ -293,9 +295,8 @@ def list_tags_with_usage(
         .join(counts_sq, counts_sq.c.tag_name == Tag.name, isouter=True)
     )
 
-    if prefix:
-        escaped, esc = escape_sql_like_string(prefix.strip().lower())
-        q = q.where(Tag.name.like(escaped + "%", escape=esc))
+    if prefix_filter:
+        q = q.where(func.substr(Tag.name, 1, len(prefix_filter)) == prefix_filter)
 
     if not include_zero:
         q = q.where(func.coalesce(counts_sq.c.cnt, 0) > 0)
@@ -306,9 +307,8 @@ def list_tags_with_usage(
         q = q.order_by(func.coalesce(counts_sq.c.cnt, 0).desc(), Tag.name.asc())
 
     total_q = select(func.count()).select_from(Tag)
-    if prefix:
-        escaped, esc = escape_sql_like_string(prefix.strip().lower())
-        total_q = total_q.where(Tag.name.like(escaped + "%", escape=esc))
+    if prefix_filter:
+        total_q = total_q.where(func.substr(Tag.name, 1, len(prefix_filter)) == prefix_filter)
     if not include_zero:
         visible_tags_sq = (
             select(AssetReferenceTag.tag_name)

app/assets/helpers.py

@@ -41,10 +41,10 @@ def get_utc_now() -> datetime:
 def normalize_tags(tags: list[str] | None) -> list[str]:
     """
     Normalize a list of tags by:
-      - Stripping whitespace and converting to lowercase.
-      - Removing duplicates.
+      - Stripping whitespace.
+      - Removing exact duplicates while preserving order and case.
     """
-    return list(dict.fromkeys(t.strip().lower() for t in (tags or []) if (t or "").strip()))
+    return list(dict.fromkeys(t.strip() for t in (tags or []) if (t or "").strip()))
 
 
 def validate_blake3_hash(s: str) -> str:

app/assets/services/bulk_ingest.py

@@ -134,6 +134,14 @@ def batch_insert_seed_assets(
 
     for spec in specs:
         absolute_path = os.path.abspath(spec["abs_path"])
+        existing_asset_id = path_to_asset_id.get(absolute_path)
+        if existing_asset_id is not None:
+            existing_tags = asset_id_to_ref_data[existing_asset_id]["tags"]
+            asset_id_to_ref_data[existing_asset_id]["tags"] = list(
+                dict.fromkeys([*existing_tags, *spec["tags"]])
+            )
+            continue
+
         asset_id = str(uuid.uuid4())
         reference_id = str(uuid.uuid4())
         absolute_path_list.append(absolute_path)

app/assets/services/ingest.py

@@ -35,6 +35,7 @@ from app.assets.services.image_dimensions import extract_image_dimensions
 from app.assets.services.path_utils import (
     compute_relative_filename,
     get_name_and_tags_from_asset_path,
+    get_path_derived_tags_from_path,
     resolve_destination_from_tags,
     validate_path_within_base,
 )
@@ -101,17 +102,32 @@ def _ingest_file_from_path(
             if preview_id and ref.preview_id != preview_id:
                 ref.preview_id = preview_id
 
-            norm = normalize_tags(list(tags))
-            if norm:
+            try:
+                backend_tags = get_path_derived_tags_from_path(locator)
+            except ValueError:
+                backend_tags = []
+            caller_tags = normalize_tags(tags)
+            backend_tags = normalize_tags(backend_tags)
+            all_tags = normalize_tags([*caller_tags, *backend_tags])
+            if all_tags:
                 if require_existing_tags:
-                    validate_tags_exist(session, norm)
-                add_tags_to_reference(
-                    session,
-                    reference_id=reference_id,
-                    tags=norm,
-                    origin=tag_origin,
-                    create_if_missing=not require_existing_tags,
-                )
+                    validate_tags_exist(session, all_tags)
+                if backend_tags:
+                    add_tags_to_reference(
+                        session,
+                        reference_id=reference_id,
+                        tags=backend_tags,
+                        origin="automatic",
+                        create_if_missing=not require_existing_tags,
+                    )
+                if caller_tags:
+                    add_tags_to_reference(
+                        session,
+                        reference_id=reference_id,
+                        tags=caller_tags,
+                        origin=tag_origin,
+                        create_if_missing=not require_existing_tags,
+                    )
 
             _update_metadata_with_filename(
                 session,
@@ -474,6 +490,10 @@ def upload_from_temp_path(
         existing = get_asset_by_hash(session, asset_hash=asset_hash)
 
     if existing is not None:
+        # Once content is already known, duplicate byte uploads are treated as
+        # reference-only creation. Request tags are labels only here: do not
+        # require upload destination tags, do not move bytes, and do not
+        # synthesize path-derived classification or uploaded provenance.
         with contextlib.suppress(Exception):
             if temp_path and os.path.exists(temp_path):
                 os.remove(temp_path)
@@ -535,7 +555,7 @@ def upload_from_temp_path(
         owner_id=owner_id,
         preview_id=preview_id,
         user_metadata=user_metadata or {},
-        tags=tags,
+        tags=[*(tags or []), "uploaded"],
         tag_origin="manual",
         require_existing_tags=False,
     )
@@ -569,15 +589,19 @@ def register_file_in_place(
 ) -> UploadResult:
     """Register an already-saved file in the asset database without moving it.
 
-    Tags are derived from the filesystem path (root category + subfolder names),
-    merged with any caller-provided tags, matching the behavior of the scanner.
+    This helper is used by upload paths that have already written bytes before
+    registering the file, so it records the same ``uploaded`` tag as the
+    multipart byte-upload path.
+
+    Tags are derived from trusted filesystem classification and merged with any
+    caller-provided tags, matching the behavior of the scanner.
     If the path is not under a known root, only the caller-provided tags are used.
     """
     try:
         _, path_tags = get_name_and_tags_from_asset_path(abs_path)
     except ValueError:
         path_tags = []
-    merged_tags = normalize_tags([*path_tags, *tags])
+    merged_tags = normalize_tags([*path_tags, *tags, "uploaded"])
 
     try:
         digest, _ = hashing.compute_blake3_hash(abs_path)

app/assets/services/path_utils.py

@@ -3,10 +3,10 @@ from pathlib import Path
 from typing import Literal
 
 import folder_paths
-from app.assets.helpers import normalize_tags
 
 
-_NON_MODEL_FOLDER_NAMES = frozenset({"custom_nodes"})
+_NON_MODEL_FOLDER_NAMES = frozenset({"configs", "custom_nodes"})
+_KNOWN_SUBFOLDER_TAGS = frozenset({"3d", "pasted", "painter", "threed", "webcam"})
 
 
 def get_comfy_models_folders() -> list[tuple[str, list[str]]]:
@@ -14,7 +14,7 @@ def get_comfy_models_folders() -> list[tuple[str, list[str]]]:
 
     Includes every category registered in folder_names_and_paths,
     regardless of whether its paths are under the main models_dir,
-    but excludes non-model entries like custom_nodes.
+    but excludes non-model entries like configs and custom_nodes.
     """
     targets: list[tuple[str, list[str]]] = []
     for name, values in folder_paths.folder_names_and_paths.items():
@@ -27,35 +27,37 @@ def get_comfy_models_folders() -> list[tuple[str, list[str]]]:
 
 
 def resolve_destination_from_tags(tags: list[str]) -> tuple[str, list[str]]:
-    """Validates and maps tags -> (base_dir, subdirs_for_fs)"""
-    if not tags:
-        raise ValueError("tags must not be empty")
-    root = tags[0].lower()
+    """Validates and maps upload routing tags -> (base_dir, subdirs_for_fs).
+
+    The request tags are only used to choose the write destination. Extra tags
+    remain labels; they do not become path components or trusted classification.
+    """
+    destination_roles = [t for t in tags if t in {"input", "models", "output"}]
+    if len(destination_roles) != 1:
+        raise ValueError("uploads require exactly one destination role: input, models, or output")
+
+    root = destination_roles[0]
     if root == "models":
-        if len(tags) < 2:
-            raise ValueError("at least two tags required for model asset")
+        model_type_tags = [t for t in tags if t.startswith("model_type:")]
+        if len(model_type_tags) != 1:
+            raise ValueError("models uploads require exactly one model_type:<folder_name> tag")
+        folder_name = model_type_tags[0].split(":", 1)[1]
+        if not folder_name:
+            raise ValueError("models uploads require exactly one model_type:<folder_name> tag")
+        model_folder_paths = dict(get_comfy_models_folders())
         try:
-            bases = folder_paths.folder_names_and_paths[tags[1]][0]
+            bases = model_folder_paths[folder_name]
         except KeyError:
-            raise ValueError(f"unknown model category '{tags[1]}'")
+            raise ValueError(f"unknown model category '{folder_name}'")
         if not bases:
-            raise ValueError(f"no base path configured for category '{tags[1]}'")
+            raise ValueError(f"no base path configured for category '{folder_name}'")
         base_dir = os.path.abspath(bases[0])
-        raw_subdirs = tags[2:]
     elif root == "input":
         base_dir = os.path.abspath(folder_paths.get_input_directory())
-        raw_subdirs = tags[1:]
-    elif root == "output":
-        base_dir = os.path.abspath(folder_paths.get_output_directory())
-        raw_subdirs = tags[1:]
     else:
-        raise ValueError(f"unknown root tag '{tags[0]}'; expected 'models', 'input', or 'output'")
-    _sep_chars = frozenset(("/", "\\", os.sep))
-    for i in raw_subdirs:
-        if i in (".", "..") or _sep_chars & set(i):
-            raise ValueError("invalid path component in tags")
+        base_dir = os.path.abspath(folder_paths.get_output_directory())
 
-    return base_dir, raw_subdirs if raw_subdirs else []
+    return base_dir, []
 
 
 def validate_path_within_base(candidate: str, base: str) -> None:
@@ -65,6 +67,62 @@ def validate_path_within_base(candidate: str, base: str) -> None:
         raise ValueError("destination escapes base directory")
 
 
+def _compute_relative_path(child: str, parent: str) -> str:
+    rel = os.path.relpath(os.path.abspath(child), os.path.abspath(parent))
+    if rel == ".":
+        return ""
+    return rel.replace(os.sep, "/")
+
+
+def _is_relative_to(child: str, parent: str) -> bool:
+    return Path(os.path.abspath(child)).is_relative_to(os.path.abspath(parent))
+
+
+def compute_asset_response_paths(file_path: str) -> tuple[str, str | None] | None:
+    """Return public (file_path, display_name) response fields for a file path.
+
+    These fields are storage locators, not model-loader namespaces. Registered
+    model-folder membership is represented by backend tags such as
+    ``model_type:<folder_name>``; response paths only use known storage roots.
+    """
+    fp_abs = os.path.abspath(file_path)
+    candidates: list[tuple[int, int, str, str]] = []
+
+    for order, (namespace, base) in enumerate(
+        (
+            ("input", folder_paths.get_input_directory()),
+            ("output", folder_paths.get_output_directory()),
+            ("temp", folder_paths.get_temp_directory()),
+            ("models", getattr(folder_paths, "models_dir", "")),
+        )
+    ):
+        if not base:
+            continue
+        base_abs = os.path.abspath(base)
+        if _is_relative_to(fp_abs, base_abs):
+            candidates.append((len(base_abs), -order, namespace, base_abs))
+
+    if not candidates:
+        return None
+
+    _base_len, _order, namespace, base = max(candidates)
+    rel = _compute_relative_path(fp_abs, base)
+    public_path = f"{namespace}/{rel}" if rel else namespace
+    return public_path, rel or None
+
+
+def compute_display_name(file_path: str) -> str | None:
+    """Return the asset's `display_name`, or None for unknown paths."""
+    result = compute_asset_response_paths(file_path)
+    return result[1] if result else None
+
+
+def compute_file_path(file_path: str) -> str | None:
+    """Return the asset's logical storage `file_path`, or None for unknown paths."""
+    result = compute_asset_response_paths(file_path)
+    return result[0] if result else None
+
+
 def compute_relative_filename(file_path: str) -> str | None:
     """
     Return the model's path relative to the last well-known folder (the model category),
@@ -72,6 +130,9 @@ def compute_relative_filename(file_path: str) -> str | None:
       /.../models/checkpoints/flux/123/flux.safetensors -> "flux/123/flux.safetensors"
       /.../models/text_encoders/clip_g.safetensors -> "clip_g.safetensors"
 
+    This is legacy metadata/view filename logic, not the public Asset response
+    `display_name`. Response fields should use compute_asset_response_paths().
+
     For non-model paths, returns None.
     """
     try:
@@ -116,9 +177,10 @@ def get_asset_category_and_relative_path(
     def _compute_relative(child: str, parent: str) -> str:
         # Normalize relative path, stripping any leading ".." components
         # by anchoring to root (os.sep) then computing relpath back from it.
-        return os.path.relpath(
+        rel = os.path.relpath(
             os.path.join(os.sep, os.path.relpath(child, parent)), os.sep
         )
+        return "" if rel == "." else rel.replace(os.sep, "/")
 
     # 1) input
     input_base = os.path.abspath(folder_paths.get_input_directory())
@@ -149,25 +211,99 @@ def get_asset_category_and_relative_path(
     if best is not None:
         _, bucket, rel_inside = best
         combined = os.path.join(bucket, rel_inside)
-        return "models", os.path.relpath(os.path.join(os.sep, combined), os.sep)
+        normalized = os.path.relpath(os.path.join(os.sep, combined), os.sep)
+        return "models", normalized.replace(os.sep, "/")
 
     raise ValueError(
         f"Path is not within input, output, temp, or configured model bases: {file_path}"
     )
 
 
+def get_backend_system_tags_from_path(path: str) -> list[str]:
+    """Return trusted backend tags derived from current filesystem facts.
+
+    The returned tags are only the backend-generated system tags: ``models``,
+    ``model_type:<folder_name>``, ``input``, ``output``, and ``temp``. Model
+    type tags are based on registered folder names, not path components.
+    """
+    fp_abs = os.path.abspath(path)
+    fp_path = Path(fp_abs)
+    tags: list[str] = []
+
+    def _add(tag: str) -> None:
+        if tag not in tags:
+            tags.append(tag)
+
+    for role, base in (
+        ("input", folder_paths.get_input_directory()),
+        ("output", folder_paths.get_output_directory()),
+        ("temp", folder_paths.get_temp_directory()),
+    ):
+        if fp_path.is_relative_to(os.path.abspath(base)):
+            _add(role)
+
+    model_types: list[str] = []
+    for folder_name, bases in get_comfy_models_folders():
+        for base in bases:
+            if fp_path.is_relative_to(os.path.abspath(base)):
+                model_types.append(folder_name)
+                break
+
+    if model_types:
+        _add("models")
+        for folder_name in model_types:
+            _add(f"model_type:{folder_name}")
+
+    if not tags:
+        raise ValueError(
+            f"Path is not within input, output, temp, or configured model bases: {path}"
+        )
+    return tags
+
+
+def get_known_subfolder_tags(subfolder: str | None) -> list[str]:
+    """Return tags for known UI/input subfolder names."""
+    if subfolder in _KNOWN_SUBFOLDER_TAGS:
+        return [subfolder]
+    return []
+
+
+def get_known_input_subfolder_tags_from_path(path: str) -> list[str]:
+    """Return known input-layout tags for files in canonical input subfolders.
+
+    These are compatibility tags for current UI-origin input directories such as
+    ``pasted`` and ``webcam``. They are intentionally narrow: only files directly
+    inside a known top-level input directory receive the matching tag.
+    """
+    fp_abs = os.path.abspath(path)
+    input_base = os.path.abspath(folder_paths.get_input_directory())
+    if not Path(fp_abs).is_relative_to(input_base):
+        return []
+
+    rel = os.path.relpath(fp_abs, input_base)
+    parts = Path(rel).parts
+    if len(parts) == 2:
+        return get_known_subfolder_tags(parts[0])
+    return []
+
+
+def get_path_derived_tags_from_path(path: str) -> list[str]:
+    """Return all backend-derived tags for an asset path."""
+    tags = get_backend_system_tags_from_path(path)
+    for tag in get_known_input_subfolder_tags_from_path(path):
+        if tag not in tags:
+            tags.append(tag)
+    return tags
+
+
 def get_name_and_tags_from_asset_path(file_path: str) -> tuple[str, list[str]]:
     """Return (name, tags) derived from a filesystem path.
 
     - name: base filename with extension
-    - tags: [root_category] + parent folder names in order
+    - tags: backend-derived tags from root/model classification and known input
+      subfolder layout conventions
 
     Raises:
         ValueError: path does not belong to any known root.
     """
-    root_category, some_path = get_asset_category_and_relative_path(file_path)
-    p = Path(some_path)
-    parent_parts = [
-        part for part in p.parent.parts if part not in (".", "..", p.anchor)
-    ]
-    return p.name, list(dict.fromkeys(normalize_tags([root_category, *parent_parts])))
+    return Path(file_path).name, get_path_derived_tags_from_path(file_path)

comfy/sd.py

@@ -468,6 +468,9 @@ class CLIP:
     def decode(self, token_ids, skip_special_tokens=True):
         return self.tokenizer.decode(token_ids, skip_special_tokens=skip_special_tokens)
 
+    def is_dynamic(self):
+        return self.patcher.is_dynamic()
+
 class VAE:
     def __init__(self, sd=None, device=None, config=None, dtype=None, metadata=None):
         if 'decoder.up_blocks.0.resnets.0.norm1.weight' in sd.keys(): #diffusers format
@@ -1251,6 +1254,8 @@ class VAE:
         except:
             return None
 
+    def is_dynamic(self):
+        return self.patcher.is_dynamic()
 
 class StyleModel:
     def __init__(self, model, device="cpu"):

comfy_api/feature_flags.py

@@ -100,6 +100,7 @@ def _parse_cli_feature_flags() -> dict[str, Any]:
 # Default server capabilities
 _CORE_FEATURE_FLAGS: dict[str, Any] = {
     "supports_preview_metadata": True,
+    "supports_model_type_tags": True,
     "max_upload_size": args.max_upload_size * 1024 * 1024, # Convert MB to bytes
     "extension": {"manager": {"supports_v4": True}},
     "node_replacements": True,

comfy_api/latest/_input_impl/video_types.py

@@ -281,11 +281,18 @@ class VideoFromFile(VideoInput):
         video_done = False
         audio_done = True
 
-        if len(container.streams.audio):
-            audio_stream = container.streams.audio[-1]
+        # Use the last decodable audio stream. Streams FFmpeg has no decoder for have no codec context,
+        # and decoding their packets crashes the process. (e.g. APAC spatial-audio track in iPhone)
+        audio_stream = next(
+            (s for s in reversed(container.streams.audio) if s.codec_context is not None),
+            None,
+        )
+        if audio_stream is not None:
             streams += [audio_stream]
             resampler = av.audio.resampler.AudioResampler(format='fltp')
             audio_done = False
+        elif len(container.streams.audio):
+            logging.warning("No decodable audio stream found in video; ignoring audio.")
 
         for packet in container.demux(*streams):
             if video_done and audio_done:
@@ -457,10 +464,13 @@ class VideoFromFile(VideoInput):
                         else:
                             output_container.metadata[key] = json.dumps(value)
 
-                # Add streams to the new container
+                # Add streams to the new container. Streams with no codec context cannot be used as an output template.
                 stream_map = {}
                 for stream in streams:
                     if isinstance(stream, (av.VideoStream, av.AudioStream, SubtitleStream)):
+                        if stream.codec_context is None:
+                            logging.warning("Skipping %s stream %d with unsupported codec", stream.type, stream.index)
+                            continue
                         out_stream = output_container.add_stream_from_template(template=stream, opaque=True)
                         stream_map[stream] = out_stream
 

comfy_api_nodes/util/upload_helpers.py

@@ -158,7 +158,14 @@ async def upload_video_to_comfyapi(
 
     # Convert VideoInput to BytesIO using specified container/codec
     video_bytes_io = BytesIO()
-    video.save_to(video_bytes_io, format=container, codec=codec)
+    try:
+        video.save_to(video_bytes_io, format=container, codec=codec)
+    except Exception as e:
+        raise ValueError(
+            f"Could not convert the input video to {container.value.upper()} for upload; "
+            f"the file may be corrupted or use an unsupported codec. "
+            f"Try re-exporting it as MP4 (H.264). Original error: {e}"
+        ) from e
     video_bytes_io.seek(0)
 
     return await upload_file_to_comfyapi(cls, video_bytes_io, filename, upload_mime_type, wait_label)

comfy_execution/caching.py

@@ -503,6 +503,21 @@ RAM_CACHE_DEFAULT_RAM_USAGE = 0.05
 
 RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER = 1.3
 
+
+def all_outputs_dynamic(outputs):
+    if outputs is None:
+        return False
+
+    for output in outputs:
+        if isinstance(output, (list, tuple)):
+            if not all_outputs_dynamic(output):
+                return False
+        elif not hasattr(output, "is_dynamic") or not output.is_dynamic():
+            return False
+
+    return True
+
+
 class RAMPressureCache(LRUCache):
 
     def __init__(self, key_class, enable_providers=False):
@@ -533,7 +548,11 @@ class RAMPressureCache(LRUCache):
         for key, cache_entry in self.cache.items():
             if not free_active and self.used_generation[key] == self.generation:
                 continue
-            oom_score =  RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER ** (self.generation - self.used_generation[key])
+
+            if all_outputs_dynamic(cache_entry.outputs) and self.used_generation[key] == self.generation:
+                continue
+
+            oom_score = RAM_CACHE_OLD_WORKFLOW_OOM_MULTIPLIER ** (self.generation - self.used_generation[key])
 
             ram_usage = RAM_CACHE_DEFAULT_RAM_USAGE
             def scan_list_for_ram_usage(outputs):

comfy_extras/nodes_text_overlay.py

@@ -0,0 +1,150 @@
+import numpy as np
+import torch
+from PIL import Image as PILImage, ImageColor, ImageDraw, ImageFont
+from typing_extensions import override
+
+from comfy_api.latest import ComfyExtension, IO
+
+
+class TextOverlay(IO.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return IO.Schema(
+            node_id="TextOverlay",
+            display_name="Draw Text Overlay",
+            category="text",
+            description="Draw text overlay on an image or batch of images.",
+            search_aliases=["text", "label", "caption", "subtitle", "watermark", "title", "addlabel", "overlay"],
+            inputs=[
+                IO.Image.Input("images"),
+                IO.String.Input("text", multiline=True, default=""),
+                IO.Float.Input("font_size", default=5.0, min=0.5, max=50.0, step=0.5, tooltip="Font size as a percentage of the image height."),
+                IO.Color.Input("color", default="#ffffff", tooltip="Color of the text."),
+                IO.Combo.Input("position", options=["top", "bottom"], default="top"),
+                IO.Combo.Input("align", options=["left", "center", "right"], default="left"),
+                IO.Boolean.Input("outline", default=True, tooltip="Draw a black outline around the text."),
+            ],
+            outputs=[IO.Image.Output(display_name="images")],
+        )
+
+    @classmethod
+    def execute(cls, images, text, font_size, color, position, align, outline) -> IO.NodeOutput:
+        if text.strip() == "":
+            return IO.NodeOutput(images)
+
+        text = text.replace("\\n", "\n").replace("\\t", "\t")
+
+        text_rgba = cls.parse_color_to_rgba(color)
+        outline_rgba = (0, 0, 0, 255) if outline else (0, 0, 0, 0)
+
+        # Render the overlay once and composite it across all frames in the batch
+        height = images.shape[1]
+        width = images.shape[2]
+        overlay_rgb, overlay_alpha = cls.render_overlay_text(width, height, text, position, align, font_size, text_rgba, outline_rgba)
+        overlay_rgb = overlay_rgb.to(device=images.device, dtype=images.dtype)
+        overlay_alpha = overlay_alpha.to(device=images.device, dtype=images.dtype)
+
+        result = images * (1.0 - overlay_alpha) + overlay_rgb * overlay_alpha
+        return IO.NodeOutput(result)
+
+    @staticmethod
+    def parse_color_to_rgba(color_string):
+        parsed = ImageColor.getrgb(color_string)
+
+        if len(parsed) == 3:
+            return (*parsed, 255)
+
+        return parsed
+
+    @classmethod
+    def render_overlay_text(cls, width, height, text, position, align, font_size, text_rgba, outline_rgba):
+        line_spacing = 1.2
+        margin_percent = 1.0
+        min_font_percent = 2.0
+        min_font_pixels = 10
+        outline_thickness_factor = 0.04
+
+        # Draw onto a transparent layer so the result can be alpha-composited over any frame.
+        layer = PILImage.new("RGBA", (width, height), (0, 0, 0, 0))
+        draw = ImageDraw.Draw(layer)
+
+        margin = int(round(margin_percent / 100.0 * min(width, height)))
+        max_width = max(1, width - 2 * margin)
+        max_height = max(1, height - 2 * margin)
+
+        # Font scales with resolution, then shrinks to fit the height.
+        size = max(1, int(round(font_size / 100.0 * height)))
+        floor = min(size, max(min_font_pixels, int(round(min_font_percent / 100.0 * height))))
+
+        while True:
+            font = ImageFont.load_default(size=size)
+            stroke = max(1, int(round(size * outline_thickness_factor))) if outline_rgba[3] > 0 else 0
+            block = "\n".join(cls.wrap_text(text, font, max_width))
+            # convert line spacing to pixel spacing
+            single = draw.textbbox((0, 0), "Ay", font=font, stroke_width=stroke)
+            double = draw.multiline_textbbox((0, 0), "Ay\nAy", font=font, spacing=0, stroke_width=stroke)
+            natural_advance = (double[3] - double[1]) - (single[3] - single[1])
+            pixel_spacing = int(round(size * line_spacing - natural_advance))
+            box = draw.multiline_textbbox((0, 0), block, font=font, spacing=pixel_spacing, stroke_width=stroke)
+            block_height = box[3] - box[1]
+
+            if block_height <= max_height or size <= floor:
+                break
+
+            size = max(floor, int(size * 0.9))
+
+        anchor_h, x = {"left": ("l", margin), "center": ("m", width / 2), "right": ("r", width - margin)}[align]
+
+        # Offset y so the rendered text sits flush against the margin
+        if position == "bottom":
+            y = height - margin - box[3]
+        else:
+            y = margin - box[1]
+
+        draw.multiline_text((x, y), block, font=font, fill=text_rgba, anchor=anchor_h + "a",
+                            align=align, spacing=pixel_spacing, stroke_width=stroke, stroke_fill=outline_rgba)
+
+        overlay = np.array(layer).astype(np.float32) / 255.0
+        overlay_rgb = torch.from_numpy(overlay[:, :, :3])
+        overlay_alpha = torch.from_numpy(overlay[:, :, 3:4])
+        return overlay_rgb, overlay_alpha
+
+    @staticmethod
+    def wrap_text(text, font, max_width):
+        lines = []
+        for raw_line in text.split("\n"):
+            words = raw_line.split()
+            if not words:
+                lines.append("")
+                continue
+            current = ""
+            # Break the line into words and split words that are too long
+            for word in words:
+                while font.getlength(word) > max_width and len(word) > 1:
+                    cut = 1
+                    while cut < len(word) and font.getlength(word[:cut + 1]) <= max_width:
+                        cut += 1
+                    if current:
+                        lines.append(current)
+                        current = ""
+                    lines.append(word[:cut])
+                    word = word[cut:]
+                candidate = word if not current else current + " " + word
+                if not current or font.getlength(candidate) <= max_width:
+                    current = candidate
+                else:
+                    lines.append(current)
+                    current = word
+            if current:
+                lines.append(current)
+        return lines
+
+
+class TextOverlayExtension(ComfyExtension):
+    @override
+    async def get_node_list(self) -> list[type[IO.ComfyNode]]:
+        return [TextOverlay]
+
+
+async def comfy_entrypoint() -> TextOverlayExtension:
+    return TextOverlayExtension()

nodes.py

@@ -2478,6 +2478,7 @@ async def init_builtin_extra_nodes():
         "nodes_glsl.py",
         "nodes_lora_debug.py",
         "nodes_textgen.py",
+        "nodes_text_overlay.py",
         "nodes_color.py",
         "nodes_toolkit.py",
         "nodes_replacements.py",

openapi.yaml

@@ -7,18 +7,18 @@ components:
                     description: Timestamp when the asset was created
                     format: date-time
                     type: string
-                display_name:
-                    description: Display name of the asset. Mirrors name for backwards compatibility.
-                    nullable: true
-                    type: string
-                file_path:
-                    description: Relative path in global-namespace-root form (e.g. "models/checkpoints/flux.safetensors")
-                    nullable: true
-                    type: string
                 hash:
                     description: Blake3 hash of the asset content.
                     pattern: ^blake3:[a-f0-9]{64}$
                     type: string
+                file_path:
+                    description: Runtime storage locator for filesystem-backed assets, using Comfy storage namespaces such as `input/`, `output/`, `temp/`, or `models/`. Not an absolute filesystem path, unique identity, or model loader path.
+                    nullable: true
+                    type: string
+                display_name:
+                    description: Human-facing label derived from `file_path`, usually the path below the top-level storage namespace. Not unique.
+                    nullable: true
+                    type: string
                 id:
                     description: Unique identifier for the asset
                     format: uuid
@@ -144,14 +144,6 @@ components:
         AssetUpdated:
             description: Response returned when an existing asset is successfully updated.
             properties:
-                display_name:
-                    description: Display name of the asset. Mirrors name for backwards compatibility.
-                    nullable: true
-                    type: string
-                file_path:
-                    description: Relative path in global-namespace-root form (e.g. "models/checkpoints/flux.safetensors")
-                    nullable: true
-                    type: string
                 hash:
                     description: Blake3 hash of the asset content.
                     pattern: ^blake3:[a-f0-9]{64}$
@@ -1644,7 +1636,7 @@ paths:
                                     format: uuid
                                     type: string
                                 tags:
-                                    description: JSON-encoded array of freeform tag strings, e.g. '["models","checkpoint"]'. Common types include "models", "input", "output", and "temp", but any tag can be used in any order.
+                                    description: JSON-encoded array of tag strings. For new byte uploads, include exactly one destination role (`input`, `output`, or `models`); `models` uploads also require exactly one `model_type:<folder_name>` tag. Extra tags are stored as labels and do not create path components.
                                     type: string
                                 user_metadata:
                                     description: Custom JSON metadata as a string
@@ -1829,7 +1821,7 @@ paths:
                     content:
                         application/json:
                             schema:
-                                $ref: '#/components/schemas/AssetUpdated'
+                                $ref: '#/components/schemas/Asset'
                     description: Asset updated successfully
                 "400":
                     content:
@@ -2470,6 +2462,9 @@ paths:
                                     supports_preview_metadata:
                                         description: Whether the server supports preview metadata
                                         type: boolean
+                                    supports_model_type_tags:
+                                        description: Whether the server supports namespaced model type asset tags
+                                        type: boolean
                                 type: object
                     description: Success
                     headers:

server.py

@@ -46,6 +46,7 @@ from comfy_api.internal import _ComfyNodeInternal
 from app.assets.seeder import asset_seeder
 from app.assets.api.routes import register_assets_routes
 from app.assets.services.ingest import register_file_in_place
+from app.assets.services.path_utils import get_known_subfolder_tags
 from app.assets.services.asset_management import resolve_hash_to_path
 
 from app.user_manager import UserManager
@@ -441,7 +442,9 @@ class PromptServer():
                 if args.enable_assets:
                     try:
                         tag = image_upload_type if image_upload_type in ("input", "output") else "input"
-                        result = register_file_in_place(abs_path=filepath, name=filename, tags=[tag])
+                        tags = [tag]
+                        tags.extend(get_known_subfolder_tags(subfolder))
+                        result = register_file_in_place(abs_path=filepath, name=filename, tags=tags)
                         resp["asset"] = {
                             "id": result.ref.id,
                             "name": result.ref.name,

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_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_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,152 @@ 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
+        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)]),
+                    ("diffusion_models", [str(shared_root)]),
+                ],
+            ),
+        ):
+            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",
+        }
+
 
 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")

tests-unit/assets_test/services/test_path_utils.py

@@ -6,7 +6,15 @@ 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_file_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 +25,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,6 +34,7 @@ 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",
@@ -34,6 +44,7 @@ def fake_dirs():
                     "input": input_dir,
                     "output": output_dir,
                     "temp": temp_dir,
+                    "models_root": models_root,
                     "models": models_dir,
                 }
 
@@ -76,6 +87,376 @@ 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)])],
+        ):
+            _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)]),
+                ("loras", [str(shared_root)]),
+            ],
+        ):
+            _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_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)])],
+        ):
+            _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_file_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_file_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_file_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_file_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_file_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_file_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)])],
+        ):
+            assert compute_file_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)])],
+        ):
+            assert (
+                compute_file_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)])],
+        ):
+            assert compute_file_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)])],
+        ):
+            assert compute_file_path(str(foo_file)) is None
+            assert compute_file_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)])],
+        ):
+            assert compute_file_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)])],
+        ):
+            assert compute_file_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_file_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_file_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_file_path("/some/random/path.png") is None
+        assert compute_display_name("/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

@@ -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("file_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("file_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("file_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("file_path") is None
+    assert detail.get("display_name") is None
+
+
+def test_create_from_hash_with_model_tags_does_not_synthesize_file_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("file_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("file_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("file_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,72 @@ 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_prefix",
+        "expected_display_prefix",
+    ),
+    [
+        (["input", "unit-tests"], ".png", "input", ""),
+        (
+            ["models", "model_type:checkpoints", "unit-tests"],
+            ".safetensors",
+            "models/checkpoints",
+            "checkpoints/",
+        ),
+    ],
+)
+def test_upload_response_includes_file_path_and_display_name(
+    tags: list[str],
+    extension: str,
+    expected_prefix: 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_file_path = f"{expected_prefix}/{expected_suffix}"
+    expected_display_name = f"{expected_display_prefix}{expected_suffix}"
+
+    assert created["file_path"] == expected_file_path
+    assert created["display_name"] == expected_display_name
+
+    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["file_path"] == expected_file_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["file_path"] == expected_file_path
+    assert match["display_name"] == expected_display_name
 
 
 @pytest.mark.parametrize("root", ["input", "output"])
@@ -192,16 +359,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 +418,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 +434,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 +443,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 +478,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/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