fixed metadata[filename] feature + new tests for this

app/database/services/__init__.py

@@ -32,6 +32,7 @@ from .queries import (
     get_asset_info_by_id,
     get_cache_state_by_asset_id,
     list_cache_states_by_asset_id,
+    pick_best_live_path,
 )
 
 __all__ = [
@@ -39,6 +40,7 @@ __all__ = [
     "asset_exists_by_hash", "get_asset_by_hash", "get_asset_info_by_id", "asset_info_exists_for_asset_id",
     "get_cache_state_by_asset_id",
     "list_cache_states_by_asset_id",
+    "pick_best_live_path",
     # info
     "list_asset_infos_page", "create_asset_info_for_existing_asset", "set_asset_info_tags",
     "update_asset_info_full", "replace_asset_info_metadata_projection",

app/database/services/content.py

@@ -12,7 +12,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import noload
 
-from ..._assets_helpers import compute_model_relative_filename, normalize_tags
+from ..._assets_helpers import compute_relative_filename, normalize_tags
 from ...storage import hashing as hashing_mod
 from ..helpers import (
     ensure_tags_exist,
@@ -21,6 +21,7 @@ from ..helpers import (
 from ..models import Asset, AssetCacheState, AssetInfo, AssetInfoTag, Tag
 from ..timeutil import utcnow
 from .info import replace_asset_info_metadata_projection
+from .queries import list_cache_states_by_asset_id, pick_best_live_path
 
 
 async def check_fs_asset_exists_quick(
@@ -106,6 +107,15 @@ async def ensure_seed_for_path(
     session.add(info)
     await session.flush()
 
+    with contextlib.suppress(Exception):
+        computed = compute_relative_filename(locator)
+        if computed:
+            await replace_asset_info_metadata_projection(
+                session,
+                asset_info_id=info.id,
+                user_metadata={"filename": computed},
+            )
+
     want = normalize_tags(tags)
     if want:
         await ensure_tags_exist(session, want, tag_type="user")
@@ -265,6 +275,8 @@ async def compute_hash_and_dedup_for_cache_state(
                 if int(remaining or 0) == 0:
                     await session.delete(asset)
                     await session.flush()
+                else:
+                    await _recompute_and_apply_filename_for_asset(session, asset_id=asset.id)
             return None
 
         digest = await hashing_mod.blake3_hash(path)
@@ -316,6 +328,7 @@ async def compute_hash_and_dedup_for_cache_state(
                     state.needs_verify = False
                     with contextlib.suppress(Exception):
                         await remove_missing_tag_for_asset_id(session, asset_id=canonical.id)
+                    await _recompute_and_apply_filename_for_asset(session, asset_id=canonical.id)
                     await session.flush()
                 return canonical.id
 
@@ -343,6 +356,7 @@ async def compute_hash_and_dedup_for_cache_state(
                         state.needs_verify = False
                         with contextlib.suppress(Exception):
                             await remove_missing_tag_for_asset_id(session, asset_id=canonical.id)
+                        await _recompute_and_apply_filename_for_asset(session, asset_id=canonical.id)
                         await session.flush()
                     return canonical.id
                 # If we got here, the integrity error was not about hash uniqueness
@@ -353,6 +367,7 @@ async def compute_hash_and_dedup_for_cache_state(
             state.needs_verify = False
             with contextlib.suppress(Exception):
                 await remove_missing_tag_for_asset_id(session, asset_id=this_asset.id)
+            await _recompute_and_apply_filename_for_asset(session, asset_id=this_asset.id)
             await session.flush()
             return this_asset.id
 
@@ -364,6 +379,7 @@ async def compute_hash_and_dedup_for_cache_state(
             state.needs_verify = False
             with contextlib.suppress(Exception):
                 await remove_missing_tag_for_asset_id(session, asset_id=this_asset.id)
+            await _recompute_and_apply_filename_for_asset(session, asset_id=this_asset.id)
             await session.flush()
             return this_asset.id
 
@@ -385,11 +401,10 @@ async def compute_hash_and_dedup_for_cache_state(
         state.needs_verify = False
         with contextlib.suppress(Exception):
             await remove_missing_tag_for_asset_id(session, asset_id=target_id)
+        await _recompute_and_apply_filename_for_asset(session, asset_id=target_id)
         await session.flush()
         return target_id
-
     except Exception:
-        # Propagate; caller records the error and continues the worker.
         raise
 
 
@@ -663,15 +678,8 @@ async def ingest_fs_asset(
 
         # metadata["filename"] hack
         if out["asset_info_id"] is not None:
-            primary_path = (
-                await session.execute(
-                    select(AssetCacheState.file_path)
-                    .where(AssetCacheState.asset_id == asset.id)
-                    .order_by(AssetCacheState.id.asc())
-                    .limit(1)
-                )
-            ).scalars().first()
-            computed_filename = compute_model_relative_filename(primary_path) if primary_path else None
+            primary_path = pick_best_live_path(await list_cache_states_by_asset_id(session, asset_id=asset.id))
+            computed_filename = compute_relative_filename(primary_path) if primary_path else None
 
             current_meta = existing_info.user_metadata or {}
             new_meta = dict(current_meta)
@@ -760,3 +768,26 @@ async def list_cache_states_with_asset_under_prefixes(
         )
     ).all()
     return [(r[0], r[1], int(r[2] or 0)) for r in rows]
+
+
+async def _recompute_and_apply_filename_for_asset(session: AsyncSession, *, asset_id: str) -> None:
+    """Compute filename from the first *existing* cache state path and apply it to all AssetInfo (if changed)."""
+    try:
+        primary_path = pick_best_live_path(await list_cache_states_by_asset_id(session, asset_id=asset_id))
+        if not primary_path:
+            return
+        new_filename = compute_relative_filename(primary_path)
+        if not new_filename:
+            return
+        infos = (
+            await session.execute(select(AssetInfo).where(AssetInfo.asset_id == asset_id))
+        ).scalars().all()
+        for info in infos:
+            current_meta = info.user_metadata or {}
+            if current_meta.get("filename") == new_filename:
+                continue
+            updated = dict(current_meta)
+            updated["filename"] = new_filename
+            await replace_asset_info_metadata_projection(session, asset_info_id=info.id, user_metadata=updated)
+    except Exception:
+        logging.exception("Failed to recompute filename metadata for asset %s", asset_id)

app/database/services/info.py

@@ -8,7 +8,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import contains_eager, noload
 
-from ..._assets_helpers import compute_model_relative_filename, normalize_tags
+from ..._assets_helpers import compute_relative_filename, normalize_tags
 from ..helpers import (
     apply_metadata_filter,
     apply_tag_filters,
@@ -18,7 +18,11 @@ from ..helpers import (
 )
 from ..models import Asset, AssetInfo, AssetInfoMeta, AssetInfoTag, Tag
 from ..timeutil import utcnow
-from .queries import get_asset_by_hash, get_cache_state_by_asset_id
+from .queries import (
+    get_asset_by_hash,
+    list_cache_states_by_asset_id,
+    pick_best_live_path,
+)
 
 
 async def list_asset_infos_page(
@@ -196,9 +200,9 @@ async def create_asset_info_for_existing_asset(
     new_meta = dict(user_metadata or {})
     computed_filename = None
     try:
-        state = await get_cache_state_by_asset_id(session, asset_id=asset.id)
-        if state and state.file_path:
-            computed_filename = compute_model_relative_filename(state.file_path)
+        p = pick_best_live_path(await list_cache_states_by_asset_id(session, asset_id=asset.id))
+        if p:
+            computed_filename = compute_relative_filename(p)
     except Exception:
         computed_filename = None
     if computed_filename:
@@ -280,9 +284,9 @@ async def update_asset_info_full(
 
     computed_filename = None
     try:
-        state = await get_cache_state_by_asset_id(session, asset_id=info.asset_id)
-        if state and state.file_path:
-            computed_filename = compute_model_relative_filename(state.file_path)
+        p = pick_best_live_path(await list_cache_states_by_asset_id(session, asset_id=info.asset_id))
+        if p:
+            computed_filename = compute_relative_filename(p)
     except Exception:
         computed_filename = None
 

app/database/services/queries.py

@@ -1,3 +1,4 @@
+import os
 from typing import Optional, Sequence, Union
 
 import sqlalchemy as sa
@@ -57,3 +58,19 @@ async def list_cache_states_by_asset_id(
             .order_by(AssetCacheState.id.asc())
         )
     ).scalars().all()
+
+
+def pick_best_live_path(states: Union[list[AssetCacheState], Sequence[AssetCacheState]]) -> str:
+    """
+    Return the best on-disk path among cache states:
+      1) Prefer a path that exists with needs_verify == False (already verified).
+      2) Otherwise, pick the first path that exists.
+      3) Otherwise return empty string.
+    """
+    alive = [s for s in states if getattr(s, "file_path", None) and os.path.isfile(s.file_path)]
+    if not alive:
+        return ""
+    for s in alive:
+        if not getattr(s, "needs_verify", False):
+            return s.file_path
+    return alive[0].file_path