Derive asset preview URLs from the file path

preview_url was assembled from a /api/view link whose type was chosen by
matching the asset's tags against "input" then "output". Anything written
anywhere else - temp above all, where preview nodes put their images - fell
off the end of that chain and came back with no preview at all. Tags are
user-editable, so removing one also silently destroyed the URL.

Derive the URL from where the file actually sits instead. That covers every
root /api/view serves, temp included, and no longer depends on tags or on a
filename in user_metadata. A file outside those roots, or content no client
can render from its own bytes, gets no preview URL rather than one that
cannot work.

Nominated previews are resolved a page at a time rather than per row, so a
list costs one extra query however long it is. A preview that is soft-deleted
or not visible to the caller drops out of that lookup and is no longer
advertised.

Author: synap5e

app/assets/api/routes.py

@@ -2,6 +2,7 @@
 import functools
 import json
 import logging
+import mimetypes
 import os
 import urllib.parse
 import uuid
@@ -32,6 +33,7 @@
     create_from_hash,
     delete_asset_reference,
     get_asset_detail,
+    get_preview_file_paths,
     list_assets_page,
     list_tags,
     remove_tags,
@@ -40,7 +42,7 @@
     upload_from_temp_path,
 )
 from app.assets.services.cursor import InvalidCursorError
-from app.assets.services.path_utils import compute_display_name
+from app.assets.services.path_utils import compute_asset_response_paths
 from app.assets.services.tagging import list_tag_histogram
 
 ROUTES = web.RouteTableDef()
@@ -207,44 +209,63 @@ def _validate_sort_field(requested: str | None) -> str:
     return "created_at"
 
 
-def _build_preview_url_from_view(tags: list[str], user_metadata: dict[str, Any] | None) -> str | None:
-    """Build a /api/view preview URL from asset tags and user_metadata filename."""
-    if not user_metadata:
+# What a client can render from the bytes themselves; anything else needs a nominated preview.
+PREVIEWABLE_MIME_PREFIXES = ("image/", "video/", "audio/", "text/")
+
+# models is deliberately absent: /api/view has no directory type for it.
+VIEWABLE_NAMESPACES = frozenset({"input", "output", "temp"})
+
+
+def _has_previewable_content(asset: schemas.AssetData | None, file_path: str | None) -> bool:
+    if asset is None:
+        return False
+    # Resolved from the path, not the caller-editable name, so a rename cannot change what previews.
+    raw = asset.mime_type or mimetypes.guess_type(file_path or "")[0] or ""
+    return raw.split(";", 1)[0].strip().lower().startswith(PREVIEWABLE_MIME_PREFIXES)
+
+
+def _build_preview_url(file_path: str | None) -> str | None:
+    # /api/view is a FileResponse: byte-range seeking, no user header, no access write.
+    if not file_path:
         return None
-    filename = user_metadata.get("filename")
-    if not filename:
+    paths = compute_asset_response_paths(file_path)
+    if not paths:
         return None
-
-    if "input" in tags:
-        view_type = "input"
-    elif "output" in tags:
-        view_type = "output"
-    else:
+    logical_path, relative_path = paths
+    namespace = logical_path.split("/", 1)[0]
+    if namespace not in VIEWABLE_NAMESPACES or not relative_path:
         return None
 
-    subfolder = ""
-    if "/" in filename:
-        subfolder, filename = filename.rsplit("/", 1)
-
-    encoded_filename = urllib.parse.quote(filename, safe="")
-    url = f"/api/view?type={view_type}&filename={encoded_filename}"
+    subfolder, _, filename = relative_path.rpartition("/")
+    url = f"/api/view?type={namespace}&filename={urllib.parse.quote(filename, safe='')}"
     if subfolder:
         url += f"&subfolder={urllib.parse.quote(subfolder, safe='')}"
     return url
 
 
-def _build_asset_response(result: schemas.AssetDetailResult | schemas.UploadResult) -> schemas_out.Asset:
-    """Build an Asset response from a service result."""
+def _resolve_preview_paths(
+    results: "list[schemas.AssetDetailResult] | list[schemas.AssetSummaryData]",
+    owner_id: str,
+) -> dict[str, str]:
+    # A miss means no live, owner-visible preview - that is what keeps a soft-deleted one quiet.
+    preview_ids = {r.ref.preview_id for r in results if r.ref.preview_id}
+    return get_preview_file_paths(sorted(preview_ids), owner_id=owner_id)
+
+
+def _build_asset_response(
+    result: schemas.AssetDetailResult | schemas.UploadResult,
+    preview_paths: dict[str, str],
+) -> schemas_out.Asset:
     if result.ref.preview_id:
-        preview_detail = get_asset_detail(result.ref.preview_id)
-        if preview_detail:
-            preview_url = _build_preview_url_from_view(preview_detail.tags, preview_detail.ref.user_metadata)
-        else:
-            preview_url = None
+        # A nominated preview is one whatever it holds, so no media check here.
+        preview_url = _build_preview_url(preview_paths.get(result.ref.preview_id))
+    elif _has_previewable_content(result.asset, result.ref.file_path):
+        preview_url = _build_preview_url(result.ref.file_path)
     else:
-        preview_url = _build_preview_url_from_view(result.tags, result.ref.user_metadata)
+        preview_url = None
     if result.ref.file_path:
-        display_name = compute_display_name(result.ref.file_path)
+        paths = compute_asset_response_paths(result.ref.file_path)
+        display_name = paths[1] if paths else None
         # In-root loader path (model category dropped): what model loaders consume.
         loader_path = result.ref.loader_path
     else:
@@ -324,7 +345,10 @@ async def list_assets_route(request: web.Request) -> web.Response:
     except InvalidCursorError as e:
         return _build_error_response(400, "INVALID_CURSOR", str(e))
 
-    summaries = [_build_asset_response(item) for item in result.items]
+    preview_paths = _resolve_preview_paths(
+        result.items, USER_MANAGER.get_request_user_id(request)
+    )
+    summaries = [_build_asset_response(item, preview_paths) for item in result.items]
 
     # has_more semantics differ by mode:
     #   - cursor mode: a non-empty next_cursor means there are more results.
@@ -363,7 +387,7 @@ async def get_asset_route(request: web.Request) -> web.Response:
                 {"id": reference_id},
             )
 
-        payload = _build_asset_response(result)
+        payload = _build_asset_response(result, _resolve_preview_paths([result], USER_MANAGER.get_request_user_id(request)))
     except ValueError as e:
         return _build_error_response(
             404, "ASSET_NOT_FOUND", str(e), {"id": reference_id}
@@ -494,7 +518,7 @@ async def create_asset_from_hash_route(request: web.Request) -> web.Response:
             404, "ASSET_NOT_FOUND", f"Asset content {body.hash} does not exist"
         )
 
-    asset = _build_asset_response(result)
+    asset = _build_asset_response(result, _resolve_preview_paths([result], USER_MANAGER.get_request_user_id(request)))
     payload_out = schemas_out.AssetCreated(
         **asset.model_dump(),
         created_new=result.created_new,
@@ -585,7 +609,7 @@ async def upload_asset(request: web.Request) -> web.Response:
         logging.exception("upload_asset failed for owner_id=%s", owner_id)
         return _build_error_response(500, "INTERNAL", "Unexpected server error.")
 
-    asset = _build_asset_response(result)
+    asset = _build_asset_response(result, _resolve_preview_paths([result], owner_id))
     payload_out = schemas_out.AssetCreated(
         **asset.model_dump(),
         created_new=result.created_new,
@@ -615,7 +639,7 @@ async def update_asset_route(request: web.Request) -> web.Response:
             owner_id=USER_MANAGER.get_request_user_id(request),
             preview_id=body.preview_id,
         )
-        payload = _build_asset_response(result)
+        payload = _build_asset_response(result, _resolve_preview_paths([result], USER_MANAGER.get_request_user_id(request)))
     except PermissionError as pe:
         return _build_error_response(403, "FORBIDDEN", str(pe), {"id": reference_id})
     except ValueError as ve:

app/assets/database/queries/__init__.py

@@ -28,6 +28,7 @@
     get_reference_by_id,
     get_reference_with_owner_check,
     get_reference_ids_by_ids,
+    get_reference_paths_by_ids,
     get_references_by_paths_and_asset_ids,
     get_references_for_prefixes,
     get_unenriched_references,
@@ -101,6 +102,7 @@
     "get_reference_by_id",
     "get_reference_with_owner_check",
     "get_reference_ids_by_ids",
+    "get_reference_paths_by_ids",
     "get_reference_tags",
     "get_references_by_paths_and_asset_ids",
     "get_references_for_prefixes",

app/assets/database/queries/asset_reference.py

@@ -1064,6 +1064,29 @@ def get_references_by_paths_and_asset_ids(
     return winners
 
 
+def get_reference_paths_by_ids(
+    session: Session,
+    reference_ids: list[str],
+    owner_id: str = "",
+) -> dict[str, str]:
+    """Map reference id -> file_path for live, owner-visible, file-backed references."""
+    if not reference_ids:
+        return {}
+
+    paths: dict[str, str] = {}
+    for chunk in iter_chunks(reference_ids, MAX_BIND_PARAMS):
+        rows = session.execute(
+            select(AssetReference.id, AssetReference.file_path).where(
+                AssetReference.id.in_(chunk),
+                AssetReference.file_path.is_not(None),
+                AssetReference.deleted_at.is_(None),
+                build_visible_owner_clause(owner_id),
+            )
+        )
+        paths.update({rid: fp for rid, fp in rows})
+    return paths
+
+
 def get_reference_ids_by_ids(
     session: Session,
     reference_ids: list[str],

app/assets/services/__init__.py

@@ -4,6 +4,7 @@
     get_asset_by_hash,
     get_asset_detail,
     list_assets_page,
+    get_preview_file_paths,
     resolve_asset_for_download,
     set_asset_preview,
     update_asset_metadata,
@@ -83,6 +84,7 @@
     "list_tags",
     "cleanup_unreferenced_assets",
     "remove_tags",
+    "get_preview_file_paths",
     "resolve_asset_for_download",
     "set_asset_preview",
     "update_asset_metadata",

app/assets/services/asset_management.py

@@ -21,6 +21,7 @@
     reference_exists_for_asset_id,
     delete_reference_by_id,
     fetch_reference_and_asset,
+    get_reference_paths_by_ids,
     soft_delete_reference_by_id,
     fetch_reference_asset_and_tags,
     get_asset_by_hash as queries_get_asset_by_hash,
@@ -424,6 +425,19 @@ def resolve_hash_to_path(
     )
 
 
+def get_preview_file_paths(
+    preview_ids: list[str],
+    owner_id: str = "",
+) -> dict[str, str]:
+    """Map preview reference id -> file_path, in one query for the whole page."""
+    if not preview_ids:
+        return {}
+    with create_session() as session:
+        return get_reference_paths_by_ids(
+            session, reference_ids=preview_ids, owner_id=owner_id
+        )
+
+
 def resolve_asset_for_download(
     reference_id: str,
     owner_id: str = "",

tests-unit/assets_test/services/test_asset_response_loader_path.py

@@ -43,7 +43,7 @@ def test_uses_persisted_loader_path_without_recomputing():
         loader_path="SENTINEL/stored.safetensors",
     )
 
-    resp = _build_asset_response(result)
+    resp = _build_asset_response(result, {})
 
     assert resp.loader_path == "SENTINEL/stored.safetensors"
 
@@ -67,7 +67,7 @@ def test_null_stored_loader_path_is_served_as_null(tmp_path: Path):
         mock_fp.models_dir = str(models)
 
         result = _make_result(file_path=str(f), loader_path=None)
-        resp = _build_asset_response(result)
+        resp = _build_asset_response(result, {})
 
         assert resp.loader_path is None
         assert resp.display_name == "checkpoints/bar.safetensors"
@@ -77,7 +77,7 @@ def test_all_path_fields_null_without_file_path():
     """API-created / hash-only references (no file_path) expose no paths."""
     result = _make_result(file_path=None, loader_path=None)
 
-    resp = _build_asset_response(result)
+    resp = _build_asset_response(result, {})
 
     assert resp.loader_path is None
     assert resp.display_name is None