fix(assets): narrow create_content integrity handling; report protected tags honestly (review-12, review2-18)

app/assets/database/queries/records.py

@@ -35,6 +35,33 @@ class RecordPageSpec(NamedTuple):
     after: RecordCursorBoundary | None = None
 
 
+_LIVE_PATH_UNIQUE_INDEX = "uq_asset_contents_path_live"
+
+
+def _is_live_path_conflict(error: IntegrityError) -> bool:
+    """True only for a collision on the live-path uniqueness guard.
+
+    ``create_content`` treats exactly one integrity failure as recoverable: two
+    live rows racing for the same ``path`` under the partial unique index
+    ``uq_asset_contents_path_live`` (``asset_contents(path) WHERE is_missing = 0``).
+    That case is resolved by handing back the row that won the race. Every other
+    integrity failure — notably the ``ck_asset_contents_size_nonneg`` and
+    ``ck_asset_contents_mtime_nonneg`` CHECK constraints — MUST propagate; treating
+    it as the race destroys the real diagnostic and surfaces a misleading
+    ``NoResultFound`` from the re-query (which finds no live row for the path).
+
+    SQLite names the *column* in the message (``UNIQUE constraint failed:
+    asset_contents.path``) rather than the index, while Postgres exposes the index
+    name on ``orig.diag.constraint_name``; match either form.
+    """
+    orig = error.orig
+    diag_name = getattr(getattr(orig, "diag", None), "constraint_name", None)
+    if diag_name == _LIVE_PATH_UNIQUE_INDEX:
+        return True
+    message = str(orig)
+    return "UNIQUE constraint failed" in message and "asset_contents.path" in message
+
+
 def create_content(session: Session, path: str, hash: str | None = None, size_bytes: int = 0, mtime_ns: int | None = None) -> AssetContent:
     content = AssetContent(path=path, hash=hash, size_bytes=size_bytes, mtime_ns=mtime_ns)
     try:
@@ -42,7 +69,9 @@ def create_content(session: Session, path: str, hash: str | None = None, size_by
             session.add(content)
             session.flush()
             return content
-    except IntegrityError:
+    except IntegrityError as error:
+        if not _is_live_path_conflict(error):
+            raise
         winner = session.execute(sa.select(AssetContent).where(AssetContent.path == path, AssetContent.is_missing.is_(False))).scalar_one()
         return winner
 

app/assets/database/queries/tags.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from typing import Iterable, Sequence
 
@@ -32,6 +32,10 @@ class RemoveTagsResult:
     removed: list[str]
     not_present: list[str]
     total_tags: list[str]
+    # Tags that ARE present on the record but carry origin="automatic", so they
+    # cannot be removed via this API. Kept distinct from ``not_present`` so a
+    # caller can tell "the tag wasn't there" apart from "the tag is protected".
+    protected: list[str] = field(default_factory=list)
 
 
 def validate_tags_exist(session: Session, tags: list[str]) -> None:

app/assets/services/tagging.py

@@ -81,6 +81,19 @@ def remove_tags(
                 )
             )
         )
+        # Requested tags that ARE present but carry origin="automatic": they
+        # cannot be removed via this API, so they belong in their own bucket
+        # rather than being lumped into not_present (which would falsely claim
+        # the tag was never on the record).
+        protected_tags = set(
+            session.scalars(
+                select(AssetTag.tag_name).where(
+                    AssetTag.asset_id == reference_id,
+                    AssetTag.origin == "automatic",
+                    AssetTag.tag_name.in_(requested_tags),
+                )
+            )
+        )
         if removable_tags:
             session.execute(
                 delete(AssetTag).where(
@@ -100,8 +113,9 @@ def remove_tags(
 
     return RemoveTagsResult(
         removed=sorted(removable_tags),
-        not_present=sorted(requested_tags - removable_tags),
+        not_present=sorted(requested_tags - removable_tags - protected_tags),
         total_tags=total_tags,
+        protected=sorted(protected_tags),
     )