Merge pull request 'Gallery speed + fandom editing + filters + pinned filter bar' (#61) from dev into main

Gallery now has in-view filtering, styled like the app's sticky v-tabs
chrome (pinned at top:64px under TopNav).

- GalleryFilterBar: combined tag+artist autocomplete (searches
  /api/tags + /api/artists), closable filter chips (multi-tag AND),
  media toggle (All/Images/Videos), Newest/Oldest sort, Clear. Writes all
  state to the URL via router.push.
- gallery store: filter is now { tag_ids, artist_id, media_type, sort,
  post_id }; applyFilterFromQuery makes the URL the single source of truth
  (deep-linkable, back-button works); chip labels resolved by id or
  pre-noted on pick. Replaces the standalone tag chip + setTag/PostFilter.
- GalleryView: renders the bar (hidden in post-detail), syncs route.query
  → store on mount + every query change.

Also untracks the transient .claude/scheduled_tasks.lock committed in
3f30327 and gitignores it.

Tests: store parses query → composable scroll params, post_id exclusivity,
newest-sort omitted, label pre-seed, single initial fetch.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.gitignore

@@ -61,6 +61,9 @@ Thumbs.db
 
 # Claude Code per-user local overrides (shared .claude/settings.json is OK to commit)
 .claude/settings.local.json
+# Transient scheduler lock/state (committed by accident in 3f30327)
+.claude/scheduled_tasks.lock
+.claude/scheduled_tasks*.json
 
 # Alembic / DB scratch
 alembic/versions/__pycache__/

alembic/versions/0034_artist_visit.py

@@ -0,0 +1,53 @@
+"""artist_visit: per-artist last-viewed timestamp for the "+N new" badge
+
+Revision ID: 0034
+Revises: 0033
+Create Date: 2026-06-03
+
+Powers the artists-directory "+N new since last visit" badge + ArtistView
+banner. Single row per artist (no user_id yet — rule #47 multi-user ACL
+is aspirational; widens to (user_id, artist_id) PK when User lands).
+
+Seed every existing artist with `last_viewed_at = NOW()` so the badge
+starts at 0 across the board — no noisy "you have 5000 unseen images"
+on first deploy. New artists auto-get a row via
+`ArtistService.find_or_create`.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0034"
+down_revision: Union[str, None] = "0033"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "artist_visit",
+        sa.Column(
+            "artist_id",
+            sa.Integer,
+            sa.ForeignKey("artist.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column(
+            "last_viewed_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+    )
+    # Seed: every existing artist starts "fully caught up". Without this,
+    # every operator with N artists would see N badges (worth of every
+    # image ever imported) on first deploy.
+    op.execute(
+        "INSERT INTO artist_visit (artist_id, last_viewed_at) "
+        "SELECT id, NOW() FROM artist"
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("artist_visit")

alembic/versions/0035_image_record_effective_date.py

@@ -0,0 +1,70 @@
+"""image_record.effective_date: materialized gallery sort key + index
+
+Revision ID: 0035
+Revises: 0034
+Create Date: 2026-06-04
+
+The gallery ordered/cursored on COALESCE(post.post_date,
+image_record.created_at) across the Post outer join. That expression spans
+two tables, so no index can serve it — every /scroll sorted a large slice
+of the library, and the frontend fired ten of them serially per initial
+load. Materialize the value into image_record.effective_date and index
+(effective_date DESC, id DESC) so the cursor scroll is an index range scan.
+
+Backfill = COALESCE(primary post's post_date, created_at) so existing rows
+keep their exact ordering. New rows get the created_at-equivalent server
+default; services/importer.py overrides it with the post's date when a
+primary post with a date is linked.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0035"
+down_revision: Union[str, None] = "0034"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add nullable first so the backfill can populate before NOT NULL.
+    op.add_column(
+        "image_record",
+        sa.Column("effective_date", sa.DateTime(timezone=True), nullable=True),
+    )
+    # Pure set-based UPDATEs (no per-row params) — immune to the 65535
+    # bind-parameter ceiling regardless of library size.
+    op.execute(
+        """
+        UPDATE image_record AS ir
+        SET effective_date = COALESCE(p.post_date, ir.created_at)
+        FROM post AS p
+        WHERE ir.primary_post_id = p.id
+        """
+    )
+    op.execute(
+        """
+        UPDATE image_record
+        SET effective_date = created_at
+        WHERE effective_date IS NULL
+        """
+    )
+    op.alter_column(
+        "image_record",
+        "effective_date",
+        nullable=False,
+        server_default=sa.text("now()"),
+    )
+    # DESC/DESC matches the gallery's ORDER BY effective_date DESC, id DESC
+    # so the scroll is a forward index scan; raw SQL because alembic's
+    # column list doesn't express per-column DESC cleanly.
+    op.execute(
+        "CREATE INDEX ix_image_record_effective_date "
+        "ON image_record (effective_date DESC, id DESC)"
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_image_record_effective_date", table_name="image_record")
+    op.drop_column("image_record", "effective_date")

backend/app/api/gallery.py

@@ -8,26 +8,42 @@ from ..services.gallery_service import GalleryService
 gallery_bp = Blueprint("gallery", __name__, url_prefix="/api/gallery")
 
 
+def _parse_filters():
+    """Parse the composable gallery filters from query args. Raises
+    ValueError (→ 400) on malformed ids. `tag_id` accepts a single id or a
+    comma-separated list (AND); `media` is image|video; `sort` is
+    newest|oldest."""
+    tag_raw = request.args.get("tag_id")
+    tag_ids = (
+        [int(x) for x in tag_raw.split(",") if x.strip()] if tag_raw else None
+    ) or None
+    post_id_raw = request.args.get("post_id")
+    post_id = int(post_id_raw) if post_id_raw else None
+    artist_id_raw = request.args.get("artist_id")
+    artist_id = int(artist_id_raw) if artist_id_raw else None
+    media = request.args.get("media")
+    media_type = media if media in ("image", "video") else None
+    sort = request.args.get("sort")
+    sort = sort if sort in ("newest", "oldest") else "newest"
+    return tag_ids, post_id, artist_id, media_type, sort
+
+
 @gallery_bp.route("/scroll", methods=["GET"])
 async def scroll():
     cursor = request.args.get("cursor") or None
     try:
         limit = int(request.args.get("limit", "50"))
+        tag_ids, post_id, artist_id, media_type, sort = _parse_filters()
     except ValueError:
-        return jsonify({"error": "limit must be an integer"}), 400
+        return jsonify({"error": "invalid filter or limit parameter"}), 400
-    tag_id_raw = request.args.get("tag_id")
-    tag_id = int(tag_id_raw) if tag_id_raw else None
-    post_id_raw = request.args.get("post_id")
-    post_id = int(post_id_raw) if post_id_raw else None
-    artist_id_raw = request.args.get("artist_id")
-    artist_id = int(artist_id_raw) if artist_id_raw else None
 
     async with get_session() as session:
         svc = GalleryService(session)
         try:
             page = await svc.scroll(
-                cursor=cursor, limit=limit, tag_id=tag_id,
+                cursor=cursor, limit=limit, tag_ids=tag_ids,
                 post_id=post_id, artist_id=artist_id,
+                media_type=media_type, sort=sort,
             )
         except ValueError as exc:
             return jsonify({"error": str(exc)}), 400
@@ -58,17 +74,16 @@ async def scroll():
 
 @gallery_bp.route("/timeline", methods=["GET"])
 async def timeline():
-    tag_id_raw = request.args.get("tag_id")
+    try:
-    tag_id = int(tag_id_raw) if tag_id_raw else None
+        tag_ids, post_id, artist_id, media_type, _sort = _parse_filters()
-    post_id_raw = request.args.get("post_id")
+    except ValueError:
-    post_id = int(post_id_raw) if post_id_raw else None
+        return jsonify({"error": "invalid filter parameter"}), 400
-    artist_id_raw = request.args.get("artist_id")
-    artist_id = int(artist_id_raw) if artist_id_raw else None
     async with get_session() as session:
         svc = GalleryService(session)
         try:
             buckets = await svc.timeline(
-                tag_id=tag_id, post_id=post_id, artist_id=artist_id
+                tag_ids=tag_ids, post_id=post_id, artist_id=artist_id,
+                media_type=media_type,
             )
         except ValueError as exc:
             return jsonify({"error": str(exc)}), 400
@@ -82,20 +97,16 @@ async def jump():
     try:
         year = int(request.args["year"])
         month = int(request.args["month"])
+        tag_ids, post_id, artist_id, media_type, sort = _parse_filters()
     except (KeyError, ValueError):
         return jsonify({"error": "year and month query params required"}), 400
-    tag_id_raw = request.args.get("tag_id")
-    tag_id = int(tag_id_raw) if tag_id_raw else None
-    post_id_raw = request.args.get("post_id")
-    post_id = int(post_id_raw) if post_id_raw else None
-    artist_id_raw = request.args.get("artist_id")
-    artist_id = int(artist_id_raw) if artist_id_raw else None
     async with get_session() as session:
         svc = GalleryService(session)
         try:
             cursor = await svc.jump_cursor(
-                year=year, month=month, tag_id=tag_id,
+                year=year, month=month, tag_ids=tag_ids,
                 post_id=post_id, artist_id=artist_id,
+                media_type=media_type, sort=sort,
             )
         except ValueError as exc:
             return jsonify({"error": str(exc)}), 400

backend/app/api/tags.py

@@ -194,15 +194,46 @@ async def remove_tag_from_image(image_id: int, tag_id: int):
     return "", 204
 
 
+@tags_bp.route("/tags/<int:tag_id>", methods=["GET"])
+async def get_tag(tag_id: int):
+    """Resolve a single tag (used by the gallery to label its active
+    tag-filter chip)."""
+    async with get_session() as session:
+        tag = await session.get(Tag, tag_id)
+        if tag is None:
+            return jsonify({"error": "tag not found"}), 404
+        return jsonify(
+            {
+                "id": tag.id,
+                "name": tag.name,
+                "kind": tag.kind.value,
+                "fandom_id": tag.fandom_id,
+            }
+        )
+
+
 @tags_bp.route("/tags/<int:tag_id>", methods=["PATCH"])
-async def rename_tag(tag_id: int):
+async def update_tag(tag_id: int):
-    body = await request.get_json()
+    """Rename and/or re-fandom a tag. Body may carry `name` and/or
-    if not body or "name" not in body:
+    `fandom_id` (a fandom tag id, or null to clear — character tags only).
-        return jsonify({"error": "name required"}), 400
+    `merge: true` resolves a collision by merging into the existing tag.
+    """
+    body = await request.get_json() or {}
+    has_name = "name" in body
+    has_fandom = "fandom_id" in body
+    if not has_name and not has_fandom:
+        return jsonify({"error": "name or fandom_id required"}), 400
+    do_merge = bool(body.get("merge"))
     async with get_session() as session:
         svc = TagService(session)
         try:
+            tag = None
+            if has_name:
                 tag = await svc.rename(tag_id, body["name"])
+            if has_fandom:
+                tag = await svc.set_fandom(
+                    tag_id, body["fandom_id"], merge=do_merge
+                )
         except TagMergeConflict as exc:
             return jsonify(
                 {
@@ -219,7 +250,12 @@ async def rename_tag(tag_id: int):
             return jsonify({"error": str(exc)}), 400
         await session.commit()
         return jsonify(
-            {"id": tag.id, "name": tag.name, "kind": tag.kind.value}
+            {
+                "id": tag.id,
+                "name": tag.name,
+                "kind": tag.kind.value,
+                "fandom_id": tag.fandom_id,
+            }
         )
 
 

backend/app/models/__init__.py

@@ -2,6 +2,7 @@
 
 from .app_setting import AppSetting
 from .artist import Artist
+from .artist_visit import ArtistVisit
 from .backup_run import BackupRun
 from .base import Base
 from .credential import Credential
@@ -28,6 +29,7 @@ __all__ = [
     "Base",
     "AppSetting",
     "Artist",
+    "ArtistVisit",
     "BackupRun",
     "Source",
     "Credential",

backend/app/models/artist_visit.py

@@ -0,0 +1,36 @@
+"""ArtistVisit — per-artist 'last viewed' timestamp.
+
+Powers the "+N new since last visit" badge on the artists directory and
+the matching banner on `ArtistView`. One row per artist, single global
+operator. When the multi-user model lands, the PK widens to
+`(user_id, artist_id)` — currently aspirational only (no User model,
+no services/access.py); operator approved skipping `user_id` for now
+under rule #22 (breaking changes welcome).
+
+Seed at migration time: every existing artist gets `last_viewed_at = NOW()`
+so the badge starts at 0 across the board (no noisy "5000 unseen" on
+first deploy). New artists also auto-get a row via
+`ArtistService.find_or_create`.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+
+class ArtistVisit(Base):
+    __tablename__ = "artist_visit"
+
+    artist_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("artist.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    last_viewed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        nullable=False,
+        server_default=func.now(),
+    )

backend/app/models/image_record.py

@@ -74,6 +74,17 @@ class ImageRecord(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
+    # Denormalized gallery sort key = COALESCE(primary post's post_date,
+    # created_at) (alembic 0035). The gallery used to compute this as a
+    # COALESCE across the Post outer join on every /scroll, which can't use
+    # an index and re-sorted a large slice of the library per page (×10 with
+    # the old serial batching). Materializing it lets the cursor scroll read
+    # ix_image_record_effective_date directly. Maintained by the importer
+    # (services/importer.py _apply_sidecar) when a primary post with a date
+    # is linked; plain inserts keep the created_at-equivalent server default.
+    effective_date: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         nullable=False,

backend/app/services/artist_directory_service.py

@@ -13,10 +13,10 @@ from __future__ import annotations
 import base64
 from dataclasses import dataclass
 
-from sqlalchemy import and_, exists, func, or_, select
+from sqlalchemy import and_, case, exists, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ..models import Artist, ImageRecord, Source
+from ..models import Artist, ArtistVisit, ImageRecord, Source
 from .gallery_service import thumbnail_url
 
 _SEP = "|"
@@ -58,9 +58,27 @@ class ArtistDirectoryService:
             raise ValueError("limit must be between 1 and 200")
 
         count_col = func.count(ImageRecord.id).label("image_count")
+        # Unseen = images imported since the artist's last_viewed_at.
+        # NULL last_viewed_at (artist created before alembic 0034 seed
+        # or before find_or_create autoseed) defensively counts as
+        # "never visited" → all images unseen. Single grouped query, no
+        # N+1.
+        unseen_col = func.count(
+            case(
+                (
+                    or_(
+                        ArtistVisit.last_viewed_at.is_(None),
+                        ImageRecord.created_at > ArtistVisit.last_viewed_at,
+                    ),
+                    ImageRecord.id,
+                ),
+                else_=None,
+            )
+        ).label("unseen_count")
         stmt = (
-            select(Artist, count_col)
+            select(Artist, count_col, unseen_col)
             .outerjoin(ImageRecord, ImageRecord.artist_id == Artist.id)
+            .outerjoin(ArtistVisit, ArtistVisit.artist_id == Artist.id)
             .group_by(Artist.id)
         )
         if q:
@@ -94,7 +112,7 @@ class ArtistDirectoryService:
             next_cursor = _encode(last_artist.name, last_artist.id)
             rows = rows[:limit]
 
-        artist_ids = [a.id for a, _ in rows]
+        artist_ids = [a.id for a, _, _ in rows]
         previews = await self._previews(artist_ids)
 
         cards = [
@@ -104,9 +122,10 @@ class ArtistDirectoryService:
                 "slug": artist.slug,
                 "is_subscription": bool(artist.is_subscription),
                 "image_count": int(image_count),
+                "unseen_count": int(unseen_count),
                 "preview_thumbnails": previews.get(artist.id, []),
             }
-            for artist, image_count in rows
+            for artist, image_count, unseen_count in rows
         ]
         return DirectoryPage(cards=cards, next_cursor=next_cursor)
 

backend/app/services/artist_service.py

@@ -9,11 +9,13 @@ Dates come from Post.post_date via ImageProvenance.post_id.
 from dataclasses import dataclass
 
 from sqlalchemy import and_, case, func, or_, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import (
     Artist,
+    ArtistVisit,
     ImageProvenance,
     ImageRecord,
     Post,
@@ -122,6 +124,12 @@ class ArtistService:
             )
         ).scalar_one()
 
+        # Mark this artist as "visited now"; the returned count is what
+        # the operator should see in the banner ("N new since last
+        # visit"). Done LAST so the read aggregates above all see the
+        # pre-visit state (cosmetic — none depend on visit data).
+        unseen_at_visit = await self._mark_visited_returning_unseen(aid)
+
         return {
             "id": artist.id,
             "name": artist.name,
@@ -129,6 +137,7 @@ class ArtistService:
             "is_subscription": bool(artist.is_subscription),
             "image_count": int(image_count),
             "post_count": int(post_count),
+            "unseen_count_at_visit": unseen_at_visit,
             "date_range": {
                 "min": dmin.isoformat() if dmin else None,
                 "max": dmax.isoformat() if dmax else None,
@@ -157,6 +166,39 @@ class ArtistService:
             ],
         }
 
+    async def _mark_visited_returning_unseen(self, artist_id: int) -> int:
+        """Read pre-visit `last_viewed_at`, count images added since,
+        then upsert `last_viewed_at = NOW()`. Returns the count BEFORE
+        the upsert so the banner has data to render.
+
+        Postgres UPSERT (`ON CONFLICT DO UPDATE`) keeps the write
+        atomic — no SELECT-then-INSERT race per
+        `reference_scalar_one_or_none_duplicates`.
+        """
+        prev = (
+            await self.session.execute(
+                select(ArtistVisit.last_viewed_at).where(
+                    ArtistVisit.artist_id == artist_id
+                )
+            )
+        ).scalar_one_or_none()
+
+        count_stmt = select(func.count(ImageRecord.id)).where(
+            ImageRecord.artist_id == artist_id
+        )
+        if prev is not None:
+            count_stmt = count_stmt.where(ImageRecord.created_at > prev)
+        unseen = (await self.session.execute(count_stmt)).scalar_one()
+
+        upsert = pg_insert(ArtistVisit.__table__).values(artist_id=artist_id)
+        upsert = upsert.on_conflict_do_update(
+            index_elements=["artist_id"],
+            set_={"last_viewed_at": func.now()},
+        )
+        await self.session.execute(upsert)
+        await self.session.commit()
+        return int(unseen)
+
     async def images(
         self, slug: str, cursor: str | None, limit: int = 60
     ) -> ArtistImagesPage | None:
@@ -230,6 +272,13 @@ class ArtistService:
             artist = Artist(name=cleaned, slug=slug)
             self.session.add(artist)
             await self.session.flush()
+            # New artist starts "caught up" — seed ArtistVisit so the
+            # directory's `+N new` badge stays at 0 until real new
+            # content arrives. Without this, the unseen-count query
+            # treats NULL last_viewed_at as "never visited" and would
+            # count every image imported in the same session.
+            self.session.add(ArtistVisit(artist_id=artist.id))
+            await self.session.flush()
             await sp.commit()
         except IntegrityError:
             await sp.rollback()

backend/app/services/download_service.py

@@ -91,10 +91,25 @@ class DownloadService:
             return setup["event_id"]
         ctx = setup
 
+        # Release the phase-1 DB connections before the (up to ~19.5-min in
+        # backfill) gallery-dl subprocess. Held checked-out across that idle
+        # window, the asyncpg/psycopg connections get reaped by the server,
+        # and phase 3's first query then hits a dead socket
+        # (asyncpg ConnectionDoesNotExistError) → download_source autoretry →
+        # _phase1_setup's in-flight guard no-ops the retry → the event
+        # strands empty for the recovery sweep (Anduo #40014, 2026-06-04).
+        # pool_pre_ping can't help a *held* connection — it only validates on
+        # pool checkout. Closing returns them to the pool so phase 3 re-
+        # acquires a live one (the async task engine uses NullPool, the sync
+        # engine pre_ping + pool_recycle=300). This is what makes the
+        # "Phase 2 — no DB connection" contract in the class docstring true.
+        await self.async_session.close()
+        self.sync_session.close()
+
         source_config = SourceConfig.from_dict(ctx["config_overrides"] or {})
         # alembic 0031 / plan #544: derive skip_value + timeout from the
         # source's backfill_runs_remaining counter. When > 0, walk the full
-        # post history (skip: True + 1800s); when 0, exit gallery-dl after
+        # post history (skip: True + 1170s); when 0, exit gallery-dl after
         # 20 contiguous archived items (skip: "exit:20" + the default
         # 870s). Operator sets backfill via POST /api/sources/{id}/backfill.
         backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0

backend/app/services/gallery_dl.py

@@ -59,29 +59,33 @@ TICK_SKIP_VALUE = "exit:20"
 # Source.backfill_runs_remaining > 0 selects this mode; the longer
 # timeout below absorbs creators with thousands of posts.
 #
-# 30 seconds shy of Celery's hard `time_limit=1200` on download_source
+# Sits below download_source's Celery soft_time_limit
-# (tasks/download.py:33). subprocess.run MUST raise TimeoutExpired
+# (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py) with ~180s of
-# before Celery SIGKILLs the worker — same rationale as the tick
+# headroom for phase-3 persist. subprocess.run MUST raise TimeoutExpired
-# default at line 74. The audit (2026-06-02) caught this at 1800,
+# before Celery raises SoftTimeLimitExceeded — that exception path
-# guaranteeing SIGKILL on any backfill that ran to its subprocess
+# captures partial stdout/stderr and finalizes the event; the soft-limit
-# budget: stdout/stderr lost, backfill_runs_remaining never
+# path (until the 2026-06-03 fix) did not. Audit history: 1800 guaranteed
-# decrements, recovery sweep stamps generic "stranded" 30 min later.
+# SIGKILL against the old hard limit (Knuxy #38275); 1170 was then sized
-# Recreates the exact Knuxy #38275 failure mode the tick 870s default
+# "30s shy of the hard limit (1200)" but still EXCEEDED the soft limit
-# was added to prevent. backfill_runs_remaining=3 still gives ~58
+# (900), so SoftTimeLimitExceeded preempted TimeoutExpired and every
-# minutes of cumulative walk across three runs for prolific creators.
+# backfill stranded empty (Anduo #39912). Raising the Celery soft/hard
+# limits to 1350/1500 (tasks/download.py) is what made 1170 safe.
+# backfill_runs_remaining=3 still gives ~58 minutes of cumulative walk
+# across three runs for prolific creators.
 BACKFILL_SKIP_VALUE = True
 BACKFILL_TIMEOUT_SECONDS = 1170
 
 
-# 30 seconds shy of download_source's Celery soft_time_limit (900s, see
+# Sits well below download_source's Celery soft_time_limit
-# tasks/download.py:32). subprocess.run MUST raise TimeoutExpired before
+# (DOWNLOAD_SOFT_TIME_LIMIT=1350, tasks/download.py). subprocess.run MUST
-# Celery raises SoftTimeLimitExceeded — otherwise Celery wins the race,
+# raise TimeoutExpired before Celery raises SoftTimeLimitExceeded —
-# SIGKILLs the worker, in-memory stdout/stderr is lost, and the
+# otherwise Celery wins the race, SIGKILLs the worker, in-memory
-# DownloadEvent ends up empty-logged with "stranded by recovery sweep"
+# stdout/stderr is lost, and the DownloadEvent ends up empty-logged with
-# 18 minutes later (operator-flagged 2026-05-31, Knuxy event #38275).
+# "stranded by recovery sweep" (operator-flagged 2026-05-31, Knuxy event
-# The 30s buffer absorbs scheduler jitter / GC pauses without making
+# #38275; recurred in backfill mode as Anduo #39912). Per-source bumps
-# legitimately-long-running syncs timeout-friendlier. Per-source bumps
+# still live in source.config_overrides for legitimately long syncs —
-# still live in source.config_overrides for legitimately long syncs.
+# keep any override below the soft limit, or the soft-limit salvage path
+# in tasks/download.py (_finalize_soft_limited) is the only safety net.
 _DEFAULT_GDL_TIMEOUT_SECONDS = 870
 
 

backend/app/services/gallery_service.py

@@ -43,16 +43,17 @@ def decode_cursor(cursor: str) -> tuple[datetime, int]:
 
 
 def _effective_date_col():
-    """SQL expression: COALESCE(post.post_date, image_record.created_at).
+    """The materialized gallery sort key: image_record.effective_date
+    (alembic 0035) = COALESCE(primary post's post_date, created_at),
+    maintained at write time by the importer.
 
-    Used as the canonical sort/group/filter key across the gallery so
+    Canonical sort/group/filter key across the gallery so images attached
-    images backfilled with primary_post_id (e.g. via tag_apply phase 4)
+    to a post surface at their original publish date, not their FC import
-    surface at their original publish date, not their FC import date.
+    date — and, now that it's a single indexed column rather than a
-    Images without a Post (or with Post.post_date NULL) fall back to
+    COALESCE across the Post outer join, the cursor scroll is an index
-    image_record.created_at and still order coherently against
+    range scan instead of a full re-sort per page.
-    post-attached ones.
     """
-    return func.coalesce(Post.post_date, ImageRecord.created_at)
+    return ImageRecord.effective_date
 
 
 def _outer_join_primary_post(stmt: Select) -> Select:
@@ -117,13 +118,43 @@ def thumbnail_url(thumbnail_path: str | None, sha256_hex: str, mime: str) -> str
     return f"/images/thumbs/{bucket}/{sha256_hex}{ext}"
 
 
-def _require_single_filter(tag_id, post_id, artist_id) -> None:
+def _require_single_filter(tag_ids, post_id, artist_id) -> None:
-    if sum(x is not None for x in (tag_id, post_id, artist_id)) > 1:
+    """post_id is the post-detail view — it can't combine with the
+    composable filters. tag_ids + artist_id (+ media_type) compose freely
+    (AND)."""
+    if post_id is not None and (tag_ids or artist_id is not None):
         raise ValueError(
-            "tag_id, post_id, artist_id are mutually exclusive"
+            "post_id cannot be combined with tag or artist filters"
         )
 
 
+def _apply_scope(stmt, *, tag_ids, post_id, artist_id, media_type):
+    """Apply the composable gallery filters to a statement already joined
+    to Post via _outer_join_primary_post.
+
+    - tag_ids: image must carry ALL of them — one correlated EXISTS per tag
+      (AND), which avoids the row-multiplication a multi-join would cause.
+    - post_id / artist_id: provenance EXISTS (post_id is exclusive, guarded
+      by _require_single_filter).
+    - media_type: 'image' | 'video' narrows by mime prefix.
+    """
+    for tid in tag_ids or []:
+        stmt = stmt.where(
+            exists().where(
+                image_tag.c.image_record_id == ImageRecord.id,
+                image_tag.c.tag_id == tid,
+            )
+        )
+    prov = _provenance_clause(post_id, artist_id)
+    if prov is not None:
+        stmt = stmt.where(prov)
+    if media_type == "image":
+        stmt = stmt.where(ImageRecord.mime.like("image/%"))
+    elif media_type == "video":
+        stmt = stmt.where(ImageRecord.mime.like("video/%"))
+    return stmt
+
+
 def _provenance_clause(post_id, artist_id):
     """Correlated EXISTS clause (NOT a join) so an image with multiple
     matching provenance rows is returned exactly once and the
@@ -179,35 +210,43 @@ class GalleryService:
         self,
         cursor: str | None,
         limit: int = 50,
-        tag_id: int | None = None,
+        tag_ids: list[int] | None = None,
         post_id: int | None = None,
         artist_id: int | None = None,
+        media_type: str | None = None,
+        sort: str = "newest",
     ) -> GalleryPage:
         if limit < 1 or limit > 200:
             raise ValueError("limit must be between 1 and 200")
-        _require_single_filter(tag_id, post_id, artist_id)
+        _require_single_filter(tag_ids, post_id, artist_id)
 
         eff = _effective_date_col()
         stmt = select(ImageRecord, Post.post_date, eff.label("eff"))
         stmt = _outer_join_primary_post(stmt)
-        if tag_id is not None:
+        stmt = _apply_scope(
-            stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
+            stmt, tag_ids=tag_ids, post_id=post_id,
-                image_tag.c.tag_id == tag_id
+            artist_id=artist_id, media_type=media_type,
         )
-        prov = _provenance_clause(post_id, artist_id)
-        if prov is not None:
-            stmt = stmt.where(prov)
 
+        descending = sort != "oldest"
         if cursor:
             cur_ts, cur_id = decode_cursor(cursor)
+            # The cursor is just (last eff, last id); the request's sort
+            # decides which side of it the next page lies on.
+            if descending:
                 stmt = stmt.where(
-                or_(
+                    or_(eff < cur_ts, and_(eff == cur_ts, ImageRecord.id < cur_id))
-                    eff < cur_ts,
-                    and_(eff == cur_ts, ImageRecord.id < cur_id),
                 )
+            else:
+                stmt = stmt.where(
+                    or_(eff > cur_ts, and_(eff == cur_ts, ImageRecord.id > cur_id))
                 )
 
-        stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(limit + 1)
+        if descending:
+            stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc())
+        else:
+            stmt = stmt.order_by(eff.asc(), ImageRecord.id.asc())
+        stmt = stmt.limit(limit + 1)
         rows = (await self.session.execute(stmt)).all()
 
         next_cursor = None
@@ -243,9 +282,10 @@ class GalleryService:
 
     async def timeline(
         self,
-        tag_id: int | None = None,
+        tag_ids: list[int] | None = None,
         post_id: int | None = None,
         artist_id: int | None = None,
+        media_type: str | None = None,
     ) -> list[TimelineBucket]:
         eff = _effective_date_col()
         year_col = func.date_part("year", eff).label("yr")
@@ -254,25 +294,23 @@ class GalleryService:
             year_col, month_col, func.count(ImageRecord.id).label("cnt")
         )
         stmt = _outer_join_primary_post(stmt)
-        _require_single_filter(tag_id, post_id, artist_id)
+        _require_single_filter(tag_ids, post_id, artist_id)
-        if tag_id is not None:
+        stmt = _apply_scope(
-            stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
+            stmt, tag_ids=tag_ids, post_id=post_id,
-                image_tag.c.tag_id == tag_id
+            artist_id=artist_id, media_type=media_type,
         )
-        prov = _provenance_clause(post_id, artist_id)
-        if prov is not None:
-            stmt = stmt.where(prov)
         stmt = stmt.group_by(year_col, month_col).order_by(year_col.desc(), month_col.desc())
         rows = (await self.session.execute(stmt)).all()
         return [TimelineBucket(year=int(r.yr), month=int(r.mo), count=int(r.cnt)) for r in rows]
 
     async def jump_cursor(
-        self, year: int, month: int, tag_id: int | None = None,
+        self, year: int, month: int, tag_ids: list[int] | None = None,
         post_id: int | None = None, artist_id: int | None = None,
+        media_type: str | None = None, sort: str = "newest",
     ) -> str | None:
-        """Returns a cursor that, when passed to scroll(), positions at the
+        """Returns a cursor that, when passed to scroll() with the same sort,
-        first image of the given year-month (by effective_date, not
+        positions at the first image of the given year-month. None if the
-        created_at). None if the bucket is empty.
+        bucket is empty.
         """
         from sqlalchemy import extract
 
@@ -282,22 +320,24 @@ class GalleryService:
             extract("month", eff) == month,
         )
         stmt = _outer_join_primary_post(stmt)
-        _require_single_filter(tag_id, post_id, artist_id)
+        _require_single_filter(tag_ids, post_id, artist_id)
-        if tag_id is not None:
+        stmt = _apply_scope(
-            stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
+            stmt, tag_ids=tag_ids, post_id=post_id,
-                image_tag.c.tag_id == tag_id
+            artist_id=artist_id, media_type=media_type,
         )
-        prov = _provenance_clause(post_id, artist_id)
+        descending = sort != "oldest"
-        if prov is not None:
+        if descending:
-            stmt = stmt.where(prov)
+            stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc())
-        stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(1)
+        else:
-        first = (await self.session.execute(stmt)).first()
+            stmt = stmt.order_by(eff.asc(), ImageRecord.id.asc())
+        first = (await self.session.execute(stmt.limit(1))).first()
         if first is None:
             return None
         record, eff_date = first
-        # Cursor is exclusive; we encode a cursor with id+1 so the row itself
+        # Cursor is exclusive; nudge the id one past the boundary row (in the
-        # is the first result in the next scroll().
+        # scan direction) so the row itself is the first result of scroll().
-        return encode_cursor(eff_date, record.id + 1)
+        boundary = record.id + 1 if descending else record.id - 1
+        return encode_cursor(eff_date, boundary)
 
     async def get_image_with_tags(self, image_id: int) -> dict | None:
         record = await self.session.get(ImageRecord, image_id)
@@ -357,17 +397,10 @@ class GalleryService:
         }
 
     async def _neighbors(self, record: ImageRecord) -> dict:
-        # Compute the boundary image's effective_date in Python (one query
+        # The boundary image's sort key is materialized on the row now
-        # below + the SELECT we already have on `record`) and use it for
+        # (alembic 0035) — read it directly instead of re-deriving COALESCE
-        # the neighbor comparison. Cheaper than re-deriving in SQL via
+        # via an extra Post lookup.
-        # correlated subquery.
+        boundary_eff = record.effective_date
-        boundary_eff = record.created_at
-        if record.primary_post_id is not None:
-            post_date = (await self.session.execute(
-                select(Post.post_date).where(Post.id == record.primary_post_id)
-            )).scalar_one_or_none()
-            if post_date is not None:
-                boundary_eff = post_date
 
         eff = _effective_date_col()
         prev_stmt = _outer_join_primary_post(

backend/app/services/importer.py

@@ -960,6 +960,14 @@ class Importer:
                 sp.rollback()
         if record.primary_post_id is None:
             record.primary_post_id = post.id
+        # Keep the denormalized gallery sort key (alembic 0035) aligned with
+        # the primary post's publish date so /scroll orders off
+        # ix_image_record_effective_date instead of COALESCE-ing across the
+        # post join. Only override when THIS post is the primary AND carries
+        # a date; otherwise the column keeps its created_at-equivalent server
+        # default (matches the old COALESCE(post_date, created_at) fallback).
+        if record.primary_post_id == post.id and post.post_date is not None:
+            record.effective_date = post.post_date
         self.session.flush()
 
     def _copy_to_library(

backend/app/services/tag_service.py

@@ -280,6 +280,69 @@ class TagService:
         await self.session.flush()
         return tag
 
+    async def set_fandom(
+        self, tag_id: int, fandom_id: int | None, *, merge: bool = False
+    ) -> Tag:
+        """Set / change / clear a character tag's fandom.
+
+        Raises TagValidationError unless the tag is a character and fandom_id
+        (when given) references a fandom tag. If the change would collide with
+        an existing character of the same name in the TARGET fandom, raises
+        TagMergeConflict (the API turns that into a 409 merge hint) — unless
+        merge=True, in which case this tag is merged INTO that existing
+        character (a deliberate cross-fandom merge) and the surviving target
+        is returned. Passing fandom_id=None clears the fandom.
+        """
+        tag = await self.session.get(Tag, tag_id)
+        if tag is None:
+            raise TagValidationError(f"Tag {tag_id} not found")
+        if tag.kind != TagKind.character:
+            raise TagValidationError("Only character tags can have a fandom")
+        if fandom_id is not None:
+            fandom = await self.session.get(Tag, fandom_id)
+            if fandom is None or fandom.kind != TagKind.fandom:
+                raise TagValidationError(
+                    f"fandom_id {fandom_id} does not reference a fandom tag"
+                )
+        if fandom_id == tag.fandom_id:
+            return tag
+
+        # Collision: another character with the same name already lives in the
+        # target fandom. Mirrors rename's (name, kind, fandom_id) uniqueness.
+        clash_stmt = (
+            select(Tag)
+            .where(Tag.name == tag.name)
+            .where(Tag.kind == TagKind.character)
+            .where(
+                Tag.fandom_id.is_(None)
+                if fandom_id is None
+                else Tag.fandom_id == fandom_id
+            )
+            .where(Tag.id != tag_id)
+        )
+        clash = (await self.session.execute(clash_stmt)).scalar_one_or_none()
+        if clash is not None:
+            if not merge:
+                source_image_count = await self.session.scalar(
+                    select(func.count())
+                    .select_from(image_tag)
+                    .where(image_tag.c.tag_id == tag_id)
+                )
+                will_alias = await self._keep_as_alias(tag_id)
+                raise TagMergeConflict(
+                    f"A character named {tag.name!r} already exists in that fandom",
+                    target_id=clash.id,
+                    target_name=clash.name,
+                    source_image_count=int(source_image_count or 0),
+                    will_alias=will_alias,
+                )
+            await self._do_merge(tag, clash)
+            return clash
+
+        tag.fandom_id = fandom_id
+        await self.session.flush()
+        return tag
+
     async def merge(self, source_id: int, target_id: int) -> MergeResult:
         """Transactionally repoint every FK from source→target, optionally
         keep source's name as a tagger alias, delete source. Atomic: any
@@ -298,7 +361,14 @@ class TagService:
             raise TagValidationError(
                 "Tags must be the same kind and fandom to merge"
             )
+        return await self._do_merge(source, target)
 
+    async def _do_merge(self, source: Tag, target: Tag) -> MergeResult:
+        """Repoint every FK source→target, optionally keep source's name as a
+        tagger alias, delete source. NO kind/fandom validation — callers that
+        need it (public merge()) validate first; set_fandom's collision
+        resolution calls this directly for a deliberate CROSS-fandom merge."""
+        source_id, target_id = source.id, target.id
         keep_as_alias = await self._keep_as_alias(source_id)
         source_name = source.name
         source_kind = source.kind

backend/app/tasks/_async_session.py

@@ -10,6 +10,7 @@ disposes it (``await engine.dispose()``) when its loop ends.
 """
 
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
 
 from ..config import get_config
 
@@ -17,5 +18,13 @@ from ..config import get_config
 def async_session_factory():
     """Return ``(sessionmaker, engine)`` bound to a fresh async engine."""
     cfg = get_config()
-    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
+    # NullPool: this engine lives for ONE task (created + disposed per
+    # asyncio.run loop), so intra-task connection pooling buys nothing and
+    # actively bit us — download_source releases its phase-1 connection
+    # before a multi-minute gallery-dl subprocess, and a *pooled* idle
+    # connection would be reaped by the server and handed back dead to
+    # phase 3 (asyncpg ConnectionDoesNotExistError, Anduo #40014). NullPool
+    # opens a fresh real connection on each checkout, so phase 3 always
+    # reconnects clean; pre_ping is then redundant.
+    engine = create_async_engine(cfg.database_url, future=True, poolclass=NullPool)
     return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine

backend/app/tasks/download.py

@@ -1,12 +1,17 @@
 """download_source Celery task — runs DownloadService for one source."""
 
 import asyncio
+import logging
+from datetime import UTC, datetime
 from pathlib import Path
 
+from celery.exceptions import SoftTimeLimitExceeded
+from sqlalchemy import select
 from sqlalchemy.exc import DBAPIError, OperationalError
+from sqlalchemy.orm import Session as SyncSession
 
 from ..celery_app import celery
-from ..models import ImportSettings
+from ..models import DownloadEvent, ImportSettings, Source
 from ..services.credential_crypto import CredentialCrypto
 from ..services.credential_service import CredentialService
 from ..services.download_service import DownloadService
@@ -16,9 +21,79 @@ from ..services.thumbnailer import Thumbnailer
 from ._async_session import async_session_factory
 from .import_file import _sync_session_factory
 
+log = logging.getLogger(__name__)
+
 IMAGES_ROOT = Path("/images")
 _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
 
+# Celery time budget for one download_source run. The ceiling that
+# governs *clean* teardown is the SOFT limit: it raises a catchable
+# SoftTimeLimitExceeded in-process, whereas the HARD limit SIGKILLs the
+# worker (no chance to finalize). Both gallery-dl subprocess budgets
+# (gallery_dl.py: _DEFAULT_GDL_TIMEOUT_SECONDS=870 tick,
+# BACKFILL_TIMEOUT_SECONDS=1170 backfill) MUST sit below the soft limit
+# so subprocess.run raises its own TimeoutExpired first — that path
+# captures partial stdout/stderr and finalizes the DownloadEvent. soft is
+# max-subprocess (1170) + ~180s phase-3 persist headroom; hard is soft +
+# 150s SIGKILL backstop. Audit 2026-06-03 (Anduo #39912): the old
+# soft=900 sat BELOW the 1170 backfill budget, so SoftTimeLimitExceeded
+# preempted TimeoutExpired and the event stranded empty. The recovery
+# sweep's DOWNLOAD_STALL_THRESHOLD_MINUTES (30 min) still trails the new
+# 25-min hard kill by 5 min, so it stays a true backstop. Invariant
+# guarded by test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit.
+DOWNLOAD_SOFT_TIME_LIMIT = 1350
+DOWNLOAD_HARD_TIME_LIMIT = 1500
+
+
+def _finalize_soft_limited(session: SyncSession, source_id: int) -> None:
+    """Defense in depth for the soft-time-limit kill path.
+
+    A SoftTimeLimitExceeded unwinds download_source before phase 3 can
+    finalize the DownloadEvent, leaving it 'running' until the recovery
+    sweep stamps a context-free "stranded" error 30 min later — AND
+    leaving backfill_runs_remaining undecremented so the source re-runs
+    and re-strands every tick (Anduo #39912, 2026-06-03). Flip the
+    in-flight event to error with a real reason, mirror phase 3's
+    source-health write, and decrement any backfill budget so a
+    chronically-slow source self-heals back to tick mode.
+
+    The caller owns the commit. All mutations are gated on actually
+    finding a running event, so a benign late soft-limit (phase 3 already
+    committed) is a no-op.
+    """
+    now = datetime.now(UTC)
+    ev = session.execute(
+        select(DownloadEvent)
+        .where(DownloadEvent.source_id == source_id)
+        .where(DownloadEvent.status == "running")
+        .order_by(DownloadEvent.id.desc())
+        .limit(1)
+    ).scalar_one_or_none()
+    if ev is None:
+        return
+    ev.status = "error"
+    ev.finished_at = now
+    ev.error = (
+        f"killed by Celery soft time limit ({DOWNLOAD_SOFT_TIME_LIMIT}s) "
+        "before the gallery-dl subprocess returned — the run exceeded its "
+        "budget and its stdout/stderr were lost with the worker thread. "
+        "If this recurs, the source is too large for one run; the backfill "
+        "budget was decremented so the next tick walks less."
+    )
+    ev.metadata_ = {
+        **(ev.metadata_ or {}),
+        "error_type": "timeout",
+        "soft_time_limited": True,
+    }
+    src = session.get(Source, source_id)
+    if src is not None:
+        src.consecutive_failures = (src.consecutive_failures or 0) + 1
+        src.last_error = "soft time limit exceeded"
+        src.error_type = "timeout"
+        src.last_checked_at = now
+        if (src.backfill_runs_remaining or 0) > 0:
+            src.backfill_runs_remaining = max(0, src.backfill_runs_remaining - 1)
+
 
 @celery.task(
     name="backend.app.tasks.download.download_source",
@@ -29,8 +104,8 @@ _KEY_PATH = IMAGES_ROOT / "secrets" / "credential_key.b64"
     retry_backoff_max=120,
     retry_jitter=True,
     max_retries=3,
-    soft_time_limit=900,
+    soft_time_limit=DOWNLOAD_SOFT_TIME_LIMIT,
-    time_limit=1200,
+    time_limit=DOWNLOAD_HARD_TIME_LIMIT,
 )
 def download_source(self, source_id: int) -> int:
     """Returns the DownloadEvent.id."""
@@ -73,4 +148,19 @@ def download_source(self, source_id: int) -> int:
         finally:
             await async_engine.dispose()
 
+    try:
         return asyncio.run(_run())
+    except SoftTimeLimitExceeded:
+        # phase 3 never ran — salvage the in-flight event so the operator
+        # sees a real reason instead of the recovery sweep's generic
+        # "stranded" 30 min later (Anduo #39912). Best-effort: a failure
+        # here must not mask the timeout. Re-raise so Celery + the
+        # task_run signal handler still record the kill.
+        try:
+            SyncFactory = _sync_session_factory()
+            with SyncFactory() as session:
+                _finalize_soft_limited(session, source_id)
+                session.commit()
+        except Exception:  # noqa: BLE001 — cleanup must not swallow the kill
+            log.exception("soft-limit finalize failed for source %s", source_id)
+        raise

backend/app/tasks/maintenance.py

@@ -46,9 +46,12 @@ MAX_RECOVERY_ATTEMPTS = 3
 ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 
 # DownloadEvent (pending|running) recovery threshold. download_source has
-# time_limit=1200s (20 min); 30 min is 10 min past that, so a legitimately-
+# time_limit=1500s (25 min, DOWNLOAD_HARD_TIME_LIMIT); 30 min is 5 min past
-# running task is never killed by the sweep. Operator-confirmed 2026-05-29
+# that, so a legitimately-running task is hard-killed before the sweep ever
-# after 43 sources stranded at "last check never" by the in-flight guard.
+# touches it — the sweep only catches events whose worker died without
+# finalizing. Operator-confirmed 2026-05-29 after 43 sources stranded at
+# "last check never" by the in-flight guard; budget bumped 2026-06-03 with
+# the soft/hard limit raise (Anduo #39912).
 DOWNLOAD_STALL_THRESHOLD_MINUTES = 30
 
 OLD_TASK_DAYS = 7
@@ -535,7 +538,7 @@ def recover_stalled_download_events() -> int:
     tasks.scan._tick_due_sources_async) inserts DownloadEvent(status='pending')
     and fires download_source.delay(). If that task dies before finalizing the
     event — worker OOM/SIGKILL, lost task, or a gallery-dl that didn't unwind
-    on the 1200s hard time_limit — the event stays in-flight forever. The next
+    on the 1500s hard time_limit — the event stays in-flight forever. The next
     tick then skips that source because of the in-flight guard (scan.py:168)
     and Source.last_checked_at never updates; the operator sees "last check
     never" in the Subscriptions health column, permanently.

frontend/src/components/discovery/ArtistCard.vue

@@ -8,6 +8,15 @@
       <div v-if="card.preview_thumbnails.length === 0" class="fc-artistcard__noimg">
         No preview
       </div>
+      <!-- Accent pill in the corner when this artist has content imported
+           since the operator last opened their detail view. Caps at 99+
+           to keep the layout compact; the actual count appears in the
+           banner inside ArtistView. -->
+      <span
+        v-if="(card.unseen_count || 0) > 0"
+        class="fc-artistcard__unseen"
+        :aria-label="`${card.unseen_count} new since last visit`"
+      >+{{ card.unseen_count > 99 ? '99+' : card.unseen_count }}</span>
     </div>
     <v-card-text class="fc-artistcard__body">
       <div class="fc-artistcard__name">{{ card.name }}</div>
@@ -37,6 +46,7 @@ function onCardClick() {
 <style scoped>
 .fc-artistcard { cursor: pointer; }
 .fc-artistcard__previews {
+  position: relative;
   display: grid; grid-template-columns: repeat(3, 1fr);
   gap: 2px; aspect-ratio: 3 / 1;
   /* Explicit floor + ceiling so tall source images can't escape the
@@ -45,6 +55,19 @@ function onCardClick() {
   overflow: hidden;
   background: rgb(var(--v-theme-surface-light));
 }
+.fc-artistcard__unseen {
+  position: absolute;
+  top: 6px; right: 6px;
+  display: inline-flex; align-items: center;
+  padding: 2px 8px;
+  font-size: 11px; font-weight: 700; letter-spacing: 0.02em;
+  font-variant-numeric: tabular-nums;
+  color: rgb(var(--v-theme-on-accent, 0, 0, 0));
+  background: rgb(var(--v-theme-accent));
+  border-radius: 999px;
+  box-shadow: 0 1px 3px rgba(0, 0, 0, 0.3);
+  pointer-events: none;
+}
 .fc-artistcard__previews img {
   display: block;
   width: 100%; height: 100%;

frontend/src/components/discovery/TagCard.vue

@@ -55,6 +55,12 @@
               />
             </template>
             <v-list density="compact">
+              <v-list-item
+                v-if="card.kind === 'character'"
+                title="Set fandom…"
+                prepend-icon="mdi-book-open-page-variant"
+                @click="$emit('set-fandom', card)"
+              />
               <v-list-item
                 title="Merge with…"
                 prepend-icon="mdi-call-merge"
@@ -78,7 +84,9 @@
 import { ref } from 'vue'
 
 const props = defineProps({ card: { type: Object, required: true } })
-const emit = defineEmits(['open', 'rename', 'manage', 'read', 'merge-with', 'delete'])
+const emit = defineEmits([
+  'open', 'rename', 'manage', 'read', 'merge-with', 'delete', 'set-fandom',
+])
 
 const editing = ref(false)
 const draft = ref('')

frontend/src/components/gallery/GalleryFilterBar.vue

@@ -0,0 +1,202 @@
+<template>
+  <div class="fc-filterbar">
+    <v-autocomplete
+      v-model="selected"
+      :items="searchItems"
+      :loading="searchLoading"
+      item-title="name" item-value="value"
+      no-filter clearable hide-details density="compact" variant="outlined"
+      placeholder="Filter by tag or artist…"
+      prepend-inner-icon="mdi-filter-variant"
+      class="fc-filterbar__search"
+      @update:search="onSearch"
+      @update:model-value="onPick"
+    >
+      <template #item="{ props: itemProps, item }">
+        <v-list-item v-bind="itemProps" :title="item.raw.name">
+          <template #prepend>
+            <v-icon size="small">{{ iconFor(item.raw) }}</v-icon>
+          </template>
+          <template #subtitle>
+            {{ item.raw.kind === 'artist' ? 'artist'
+              : (item.raw.fandom_name ? `character · ${item.raw.fandom_name}` : item.raw.kind) }}
+          </template>
+        </v-list-item>
+      </template>
+    </v-autocomplete>
+
+    <div class="fc-filterbar__chips">
+      <v-chip
+        v-for="id in store.filter.tag_ids" :key="`t${id}`"
+        size="small" closable :color="chipColor(id)" variant="tonal"
+        @click:close="removeTag(id)"
+      >{{ store.tagLabels[id] || `#${id}` }}</v-chip>
+      <v-chip
+        v-if="store.filter.artist_id"
+        size="small" closable color="accent" variant="tonal"
+        prepend-icon="mdi-account"
+        @click:close="clearArtist"
+      >{{ store.artistLabel || `Artist #${store.filter.artist_id}` }}</v-chip>
+    </div>
+
+    <v-spacer />
+
+    <v-btn-toggle
+      :model-value="store.filter.media_type ?? 'all'"
+      density="compact" mandatory variant="outlined" divided
+      @update:model-value="(v) => setMedia(v === 'all' ? null : v)"
+    >
+      <v-btn value="all" size="small">All</v-btn>
+      <v-btn value="image" size="small">Images</v-btn>
+      <v-btn value="video" size="small">Videos</v-btn>
+    </v-btn-toggle>
+
+    <v-select
+      :model-value="store.filter.sort"
+      :items="SORTS"
+      density="compact" hide-details variant="outlined"
+      class="fc-filterbar__sort"
+      @update:model-value="setSort"
+    />
+
+    <v-btn
+      v-if="hasActiveFilters" variant="text" size="small"
+      prepend-icon="mdi-close" @click="clearAll"
+    >Clear</v-btn>
+  </div>
+</template>
+
+<script setup>
+import { computed, ref } from 'vue'
+import { useRouter } from 'vue-router'
+import { useApi } from '../../composables/useApi.js'
+import { useGalleryStore } from '../../stores/gallery.js'
+import { useTagStore } from '../../stores/tags.js'
+
+const store = useGalleryStore()
+const tagStore = useTagStore()
+const api = useApi()
+const router = useRouter()
+
+const SORTS = [
+  { title: 'Newest first', value: 'newest' },
+  { title: 'Oldest first', value: 'oldest' },
+]
+
+const selected = ref(null)
+const searchItems = ref([])
+const searchLoading = ref(false)
+let debounce = null
+
+const hasActiveFilters = computed(() =>
+  store.filter.tag_ids.length > 0 ||
+  store.filter.artist_id != null ||
+  store.filter.media_type != null ||
+  store.filter.sort !== 'newest'
+)
+
+function iconFor(raw) {
+  if (raw.kind === 'artist') return 'mdi-account'
+  return { character: 'mdi-account-circle', fandom: 'mdi-book-open-page-variant',
+           series: 'mdi-bookshelf' }[raw.kind] || 'mdi-tag'
+}
+function chipColor(id) {
+  // Tag chips use the same per-kind palette as the rest of the app; we only
+  // know the kind from the autocomplete pick, so fall back to a neutral tone.
+  return tagStore.colorFor(pickedKind.value[id] || 'general')
+}
+const pickedKind = ref({})
+
+function onSearch(q) {
+  if (debounce) clearTimeout(debounce)
+  if (!q || !q.trim()) { searchItems.value = []; return }
+  debounce = setTimeout(async () => {
+    searchLoading.value = true
+    try {
+      const [tags, artists] = await Promise.all([
+        api.get('/api/tags/autocomplete', { params: { q, limit: 10 } }),
+        api.get('/api/artists/autocomplete', { params: { q, limit: 10 } }),
+      ])
+      searchItems.value = [
+        ...(artists || []).map((a) => ({
+          kind: 'artist', id: a.id, name: a.name, value: `artist:${a.id}`,
+        })),
+        ...(tags || []).map((t) => ({
+          kind: t.kind, id: t.id, name: t.name, value: `tag:${t.id}`,
+          fandom_name: t.fandom_name,
+        })),
+      ]
+    } catch {
+      searchItems.value = []
+    } finally {
+      searchLoading.value = false
+    }
+  }, 250)
+}
+
+function onPick(value) {
+  const item = searchItems.value.find((i) => i.value === value)
+  selected.value = null
+  searchItems.value = []
+  if (!item) return
+  if (item.kind === 'artist') {
+    store.noteArtistLabel(item.name)
+    pushFilter((n) => { n.artist_id = item.id })
+  } else {
+    store.noteTagLabel(item.id, item.name)
+    pickedKind.value = { ...pickedKind.value, [item.id]: item.kind }
+    pushFilter((n) => { if (!n.tag_ids.includes(item.id)) n.tag_ids.push(item.id) })
+  }
+}
+
+function removeTag(id) {
+  pushFilter((n) => { n.tag_ids = n.tag_ids.filter((t) => t !== id) })
+}
+function clearArtist() {
+  store.noteArtistLabel(null)
+  pushFilter((n) => { n.artist_id = null })
+}
+function setMedia(m) { pushFilter((n) => { n.media_type = m }) }
+function setSort(s) { pushFilter((n) => { n.sort = s }) }
+function clearAll() { router.push({ name: 'gallery', query: {} }) }
+
+// Single write path: clone the current filter, mutate, serialize to the URL.
+// The route watcher in GalleryView applies it to the store and reloads.
+function pushFilter(mutate) {
+  const f = store.filter
+  const n = {
+    tag_ids: [...f.tag_ids],
+    artist_id: f.artist_id,
+    media_type: f.media_type,
+    sort: f.sort,
+  }
+  mutate(n)
+  const q = {}
+  if (n.tag_ids.length) q.tag_id = n.tag_ids.join(',')
+  if (n.artist_id) q.artist_id = String(n.artist_id)
+  if (n.media_type) q.media = n.media_type
+  if (n.sort && n.sort !== 'newest') q.sort = n.sort
+  router.push({ name: 'gallery', query: q })
+}
+</script>
+
+<style scoped>
+/* Pinned under the 64px TopNav, matching the app's sticky v-tabs chrome
+   (SettingsView / ArtistHeader / SubscriptionsView). */
+.fc-filterbar {
+  position: sticky;
+  top: 64px;
+  z-index: 4;
+  display: flex;
+  align-items: center;
+  gap: 10px;
+  flex-wrap: wrap;
+  padding: 8px 4px;
+  margin-bottom: 12px;
+  background: rgb(var(--v-theme-surface));
+  border-bottom: 1px solid rgb(var(--v-theme-surface-light));
+}
+.fc-filterbar__search { max-width: 320px; min-width: 200px; }
+.fc-filterbar__chips { display: flex; align-items: center; gap: 6px; flex-wrap: wrap; }
+.fc-filterbar__sort { max-width: 150px; }
+</style>

frontend/src/components/gallery/GalleryItem.vue

@@ -9,11 +9,16 @@
   >
     <div class="fc-gallery-item__media">
       <img
-        v-if="!isVideo" :src="image.thumbnail_url" :alt="`Image ${image.id}`"
+        v-if="!isVideo" ref="imgEl" :src="image.thumbnail_url"
-        loading="lazy" @error="onThumbError"
+        :alt="`Image ${image.id}`" loading="lazy"
+        :class="{ 'is-loaded': loaded }"
+        @load="loaded = true" @error="onThumbError"
       >
       <div v-else class="fc-gallery-item__video-thumb">
-        <img :src="image.thumbnail_url" :alt="`Video ${image.id}`" loading="lazy">
+        <img
+          ref="imgEl" :src="image.thumbnail_url" :alt="`Video ${image.id}`"
+          loading="lazy" :class="{ 'is-loaded': loaded }" @load="loaded = true"
+        >
         <v-icon class="fc-gallery-item__video-badge" icon="mdi-play-circle" />
       </div>
       <div v-if="thumbError" class="fc-gallery-item__placeholder">
@@ -38,7 +43,7 @@
 </template>
 
 <script setup>
-import { computed, ref } from 'vue'
+import { computed, onMounted, ref } from 'vue'
 import { useGallerySelectionStore } from '../../stores/gallerySelection.js'
 
 const props = defineProps({ image: { type: Object, required: true } })
@@ -46,6 +51,18 @@ const emit = defineEmits(['open'])
 
 const sel = useGallerySelectionStore()
 const thumbError = ref(false)
+
+// Reveal each tile when ITS OWN thumbnail finishes loading (`@load`), not
+// when its metadata batch lands — so tiles fade/flip in individually in
+// load order instead of popping in together (operator-flagged 2026-06-04).
+const loaded = ref(false)
+const imgEl = ref(null)
+onMounted(() => {
+  // A cached thumbnail can already be complete before @load binds; reflect
+  // that so the tile reveals instead of sitting invisible at opacity 0.
+  const el = imgEl.value
+  if (el && el.complete && el.naturalWidth > 0) loaded.value = true
+})
 const isVideo = computed(
   () => props.image.mime && props.image.mime.startsWith('video/')
 )
@@ -82,6 +99,30 @@ function onThumbError() { thumbError.value = true }
 }
 .fc-gallery-item__media img {
   width: 100%; height: 100%; object-fit: cover; display: block;
+  opacity: 0;
+}
+/* Entrance: each thumbnail flips up out of a slight backward tilt and
+   settles into place, played WHEN ITS OWN image finishes loading (the
+   `is-loaded` class) rather than on a batch/timer — so tiles cascade in
+   natural load order instead of popping in together. Mirrors the showcase
+   MasonryGrid entrance styling (operator-flagged 2026-06-04). */
+.fc-gallery-item__media img.is-loaded {
+  animation: fc-gallery-item-in 0.5s cubic-bezier(0.34, 1.45, 0.64, 1) both;
+}
+@keyframes fc-gallery-item-in {
+  0% {
+    opacity: 0;
+    transform: perspective(1000px) rotateX(-22deg) translateY(20px) scale(0.96);
+  }
+  55% { opacity: 1; }
+  100% {
+    opacity: 1;
+    transform: perspective(1000px) rotateX(0) translateY(0) scale(1);
+  }
+}
+@media (prefers-reduced-motion: reduce) {
+  .fc-gallery-item__media img { opacity: 1; }
+  .fc-gallery-item__media img.is-loaded { animation: none; }
 }
 .fc-gallery-item__video-thumb { position: relative; height: 100%; }
 .fc-gallery-item__video-badge {

frontend/src/components/modal/FandomSetDialog.vue

@@ -0,0 +1,128 @@
+<template>
+  <v-card>
+    <v-card-title class="text-body-1">Fandom for “{{ tag.name }}”</v-card-title>
+    <v-card-text>
+      <template v-if="!collision">
+        <v-autocomplete
+          v-model="selectedId"
+          :items="store.fandomCache"
+          :item-title="(f) => f.name" :item-value="(f) => f.id"
+          label="Fandom" clearable density="compact"
+          :hint="selectedId == null
+            ? 'No fandom — the character will be unassigned.' : ''"
+          persistent-hint
+        />
+        <v-divider class="my-3" />
+        <p class="text-caption mb-2">Or create a new fandom:</p>
+        <div class="d-flex" style="gap: 8px;">
+          <v-text-field
+            v-model="newName" placeholder="New fandom name"
+            density="compact" hide-details
+            @keydown.enter.prevent="onCreate"
+          />
+          <v-btn
+            :disabled="!newName.trim() || busy" rounded="pill"
+            @click="onCreate"
+          >Create</v-btn>
+        </div>
+        <v-alert
+          v-if="error" type="error" variant="tonal" density="compact"
+          class="mt-3"
+        >{{ error }}</v-alert>
+      </template>
+
+      <template v-else>
+        <v-alert type="warning" variant="tonal" density="compact" class="mb-3">
+          A character named “{{ tag.name }}” already exists in that fandom.
+        </v-alert>
+        <p class="text-body-2">
+          Merge this tag into “{{ collision.target.name }}”?
+          {{ collision.source_image_count }} image
+          association{{ collision.source_image_count === 1 ? '' : 's' }}
+          will move over and this tag will be deleted{{
+            collision.will_alias ? ' (its name kept as a tagger alias)' : '' }}.
+        </p>
+      </template>
+    </v-card-text>
+    <v-card-actions>
+      <v-spacer />
+      <template v-if="!collision">
+        <v-btn variant="text" :disabled="busy" @click="$emit('cancel')">
+          Cancel
+        </v-btn>
+        <v-btn
+          color="primary" rounded="pill" :loading="busy"
+          :disabled="selectedId === (tag.fandom_id ?? null)"
+          @click="onSave"
+        >Save</v-btn>
+      </template>
+      <template v-else>
+        <v-btn variant="text" :disabled="busy" @click="collision = null">
+          Back
+        </v-btn>
+        <v-btn
+          color="warning" variant="flat" rounded="pill" :loading="busy"
+          @click="onConfirmMerge"
+        >Merge</v-btn>
+      </template>
+    </v-card-actions>
+  </v-card>
+</template>
+
+<script setup>
+import { onMounted, ref } from 'vue'
+import { useTagStore } from '../../stores/tags.js'
+
+const props = defineProps({ tag: { type: Object, required: true } })
+const emit = defineEmits(['updated', 'cancel'])
+
+const store = useTagStore()
+
+const selectedId = ref(props.tag.fandom_id ?? null)
+const newName = ref('')
+const busy = ref(false)
+const error = ref(null)
+const collision = ref(null)
+
+onMounted(() => { if (store.fandomCache.length === 0) store.loadFandoms() })
+
+async function onCreate() {
+  const name = newName.value.trim()
+  if (!name) return
+  busy.value = true
+  error.value = null
+  try {
+    const f = await store.createFandom(name)
+    selectedId.value = f.id
+    newName.value = ''
+  } catch (e) {
+    error.value = e.message || String(e)
+  } finally {
+    busy.value = false
+  }
+}
+
+async function save(merge) {
+  busy.value = true
+  error.value = null
+  try {
+    const body = await store.setFandom(props.tag.id, selectedId.value, { merge })
+    emit('updated', body)
+  } catch (e) {
+    // 409 on first attempt → surface the merge confirmation; cross-fandom
+    // collisions can't go through the regular /merge endpoint, so the
+    // resolution is a second setFandom with merge: true.
+    if (!merge && e.status === 409 && e.body && e.body.target) {
+      collision.value = e.body
+    } else {
+      error.value = e.message || String(e)
+      collision.value = null
+    }
+  } finally {
+    busy.value = false
+  }
+}
+
+function onSave() { save(false) }
+function onConfirmMerge() { save(true) }
+</script>

frontend/src/components/modal/TagPanel.vue

@@ -28,6 +28,11 @@
               <v-list-item @click="openRename(tag)">
                 <v-list-item-title>Rename…</v-list-item-title>
               </v-list-item>
+              <v-list-item
+                v-if="tag.kind === 'character'" @click="openSetFandom(tag)"
+              >
+                <v-list-item-title>Set fandom…</v-list-item-title>
+              </v-list-item>
             </v-list>
           </v-menu>
         </span>
@@ -57,6 +62,13 @@
         @renamed="onRenamed" @cancel="renameDialog = false"
       />
     </v-dialog>
+
+    <v-dialog v-model="fandomDialog" max-width="460">
+      <FandomSetDialog
+        v-if="fandomTarget" :tag="fandomTarget"
+        @updated="onFandomUpdated" @cancel="fandomDialog = false"
+      />
+    </v-dialog>
   </aside>
 </template>
 
@@ -67,6 +79,7 @@ import { useTagStore } from '../../stores/tags.js'
 import TagAutocomplete from './TagAutocomplete.vue'
 import SuggestionsPanel from './SuggestionsPanel.vue'
 import TagRenameDialog from './TagRenameDialog.vue'
+import FandomSetDialog from './FandomSetDialog.vue'
 
 const modal = useModalStore()
 const store = useTagStore()
@@ -107,6 +120,18 @@ async function onRenamed() {
   // Reflect the new name in the modal's current tag list without a full reload.
   await modal.reloadTags()
 }
+
+const fandomDialog = ref(false)
+const fandomTarget = ref(null)
+function openSetFandom(tag) {
+  fandomTarget.value = tag
+  fandomDialog.value = true
+}
+async function onFandomUpdated() {
+  fandomDialog.value = false
+  // A fandom change can merge the tag away; reload to reflect the new state.
+  await modal.reloadTags()
+}
 </script>
 
 <style scoped>

frontend/src/stores/gallery.js

@@ -3,12 +3,15 @@ import { ref, computed } from 'vue'
 import { useApi } from '../composables/useApi.js'
 import { useInflightToken } from '../composables/useInflightToken.js'
 
-// Operator-confirmed 2026-05-30: fetch PAGE-sized chunks instead of one
+// Initial paint is a SINGLE request (INITIAL_LIMIT). The old 10×serial-
-// 50-item request so items render as each batch lands. Total initial
+// batch loop (2026-05-30) only staggered METADATA, which isn't the visual
-// count is unchanged (PAGE * INITIAL_BATCHES = 50). Infinite-scroll also
+// bottleneck — thumbnails load as independent `<img>` requests, and
-// pulls PAGE per trigger to keep appends progressive.
+// GalleryItem now reveals each tile on its own image `@load`. One fetch is
-const PAGE = 5
+// far fewer round-trips and faster to first paint; the reveal-on-load is
-const INITIAL_BATCHES = 10
+// what makes appearance progressive. Infinite scroll pulls PAGE per trigger.
+// Reworked 2026-06-04.
+const PAGE = 25
+const INITIAL_LIMIT = 50
 
 export const useGalleryStore = defineStore('gallery', () => {
   const api = useApi()
@@ -18,7 +21,14 @@ export const useGalleryStore = defineStore('gallery', () => {
   const nextCursor = ref(null)
   const loading = ref(false)
   const error = ref(null)
-  const filter = ref({ tag_id: null, post_id: null })
+  const filter = ref({
+    tag_ids: [], artist_id: null, media_type: null,
+    sort: 'newest', post_id: null,
+  })
+  // Display names for the active filter chips — resolved by id on deep-link
+  // and pre-noted by the filter bar when a user picks from autocomplete.
+  const tagLabels = ref({})    // tagId -> name
+  const artistLabel = ref(null)
 
   const timelineBuckets = ref([])
   const timelineLoading = ref(false)
@@ -32,22 +42,16 @@ export const useGalleryStore = defineStore('gallery', () => {
     images.value = []
     dateGroups.value = []
     nextCursor.value = null
-    // Sequentially fetch INITIAL_BATCHES chunks so items render as each
+    await loadMore(INITIAL_LIMIT)
-    // batch lands rather than blocking on one big response. Stop early
-    // when the backend reports no more pages.
-    for (let i = 0; i < INITIAL_BATCHES; i++) {
-      if (i > 0 && nextCursor.value === null) break
-      await loadMore()
-    }
   }
 
-  async function loadMore() {
+  async function loadMore(limit = PAGE) {
     if (loading.value) return
     loading.value = true
     error.value = null
     const t = inflight.claim()
     try {
-      const params = { limit: PAGE, ...activeFilterParam() }
+      const params = { limit, ...activeFilterParam() }
       if (nextCursor.value) params.cursor = nextCursor.value
       const body = await api.get('/api/gallery/scroll', { params })
       if (!t.isCurrent()) return
@@ -89,23 +93,63 @@ export const useGalleryStore = defineStore('gallery', () => {
   }
 
   function activeFilterParam() {
-    if (filter.value.tag_id) return { tag_id: filter.value.tag_id }
+    // post_id is the exclusive post-detail view.
     if (filter.value.post_id) return { post_id: filter.value.post_id }
-    return {}
+    const p = {}
+    if (filter.value.tag_ids.length) p.tag_id = filter.value.tag_ids.join(',')
+    if (filter.value.artist_id) p.artist_id = filter.value.artist_id
+    if (filter.value.media_type) p.media = filter.value.media_type
+    if (filter.value.sort && filter.value.sort !== 'newest') p.sort = filter.value.sort
+    return p
   }
 
-  function setTagFilter(tagId) {
+  // URL is the source of truth for filters. GalleryView calls this on mount
-    filter.value.tag_id = tagId
+  // and on every route-query change; the filter bar mutates the URL
-    filter.value.post_id = null
+  // (router.push) rather than the store directly, so deep-links, the back
-    loadInitial()
+  // button, and bar actions all funnel through one path.
-    loadTimeline()
+  async function applyFilterFromQuery(q) {
+    filter.value = {
+      tag_ids: _parseIds(q.tag_id),
+      artist_id: _toId(q.artist_id),
+      media_type: ['image', 'video'].includes(q.media) ? q.media : null,
+      sort: q.sort === 'oldest' ? 'oldest' : 'newest',
+      post_id: _toId(q.post_id),
+    }
+    await loadInitial()
+    await loadTimeline()
+    _resolveLabels()
   }
 
-  function setPostFilter(postId) {
+  function _toId(v) {
-    filter.value.post_id = postId
+    const n = Number(v)
-    filter.value.tag_id = null
+    return Number.isInteger(n) && n > 0 ? n : null
-    loadInitial()
+  }
-    loadTimeline()
+  function _parseIds(raw) {
+    if (!raw) return []
+    return String(raw).split(',').map(Number).filter((n) => Number.isInteger(n) && n > 0)
+  }
+
+  // Pre-seed a label so a freshly-picked chip shows its name without a
+  // round-trip; the bar calls these before pushing the new URL.
+  function noteTagLabel(id, name) { tagLabels.value = { ...tagLabels.value, [id]: name } }
+  function noteArtistLabel(name) { artistLabel.value = name || null }
+
+  async function _resolveLabels() {
+    for (const id of filter.value.tag_ids) {
+      if (tagLabels.value[id]) continue
+      try {
+        const t = await api.get(`/api/tags/${id}`)
+        tagLabels.value = { ...tagLabels.value, [id]: t.name }
+      } catch { /* chip falls back to #id */ }
+    }
+    if (filter.value.artist_id && !artistLabel.value) {
+      // The filtered set is this artist — derive the chip label from a tile.
+      const hit = images.value.find(
+        (i) => i.artist && i.artist.id === filter.value.artist_id
+      )
+      artistLabel.value = hit?.artist?.name || null
+    }
+    if (!filter.value.artist_id) artistLabel.value = null
   }
 
   const hasMore = computed(() => nextCursor.value !== null)
@@ -113,8 +157,9 @@ export const useGalleryStore = defineStore('gallery', () => {
 
   return {
     images, dateGroups, hasMore, isEmpty, loading, error,
-    filter, timelineBuckets, timelineLoading,
+    filter, tagLabels, artistLabel, timelineBuckets, timelineLoading,
-    loadInitial, loadMore, loadTimeline, jumpTo, setTagFilter, setPostFilter
+    loadInitial, loadMore, loadTimeline, jumpTo,
+    applyFilterFromQuery, noteTagLabel, noteArtistLabel,
   }
 })
 

frontend/src/stores/tags.js

@@ -49,8 +49,21 @@ export const useTagStore = defineStore('tags', () => {
     return fandom
   }
 
+  // Set / change / clear a character tag's fandom. fandomId null clears it.
+  // Throws ApiError (status 409, body.target) on a name collision in the
+  // target fandom; pass { merge: true } to resolve it by merging this tag
+  // into the existing character. Returns the updated/surviving tag.
+  async function setFandom(tagId, fandomId, { merge = false } = {}) {
+    const body = { fandom_id: fandomId ?? null }
+    if (merge) body.merge = true
+    return await api.patch(`/api/tags/${tagId}`, { body })
+  }
+
   function kindOptions() { return KIND_OPTIONS }
   function colorFor(kind) { return KIND_COLOR[kind] || 'on-surface' }
 
-  return { fandomCache, autocomplete, loadFandoms, createFandom, kindOptions, colorFor }
+  return {
+    fandomCache, autocomplete, loadFandoms, createFandom, setFandom,
+    kindOptions, colorFor
+  }
 })

frontend/src/views/ArtistView.vue

@@ -20,6 +20,22 @@
       :last-added="store.lastAdded"
     />
     <v-container fluid class="pt-2 pb-4">
+      <!-- "N new since last visit" banner. Visible only on the initial
+           load that triggered the visit-mark; dismissable via close
+           button or by switching tabs. Re-entry only re-shows if more
+           content has arrived (overview returns 0 immediately after a
+           previous visit). -->
+      <v-alert
+        v-if="unseenBanner"
+        type="info" variant="tonal" density="compact"
+        class="mb-3" closable
+        @click:close="unseenBanner = false"
+      >
+        <span class="fc-artist__unseen-msg">
+          <strong>{{ store.overview.unseen_count_at_visit }}</strong>
+          new since last visit
+        </span>
+      </v-alert>
       <v-window v-model="tab">
         <v-window-item value="posts">
           <ArtistPostsTab
@@ -39,7 +55,7 @@
 </template>
 
 <script setup>
-import { computed, watch } from 'vue'
+import { computed, ref, watch } from 'vue'
 import { useRoute } from 'vue-router'
 
 import { useArtistStore } from '../stores/artist.js'
@@ -60,6 +76,10 @@ const { tab, resolve } = useTabQuery(
   () => ((store.postCount ?? 0) > 0 ? 'posts' : 'gallery'),
 )
 
+// One-shot banner — reset on each new artist-slug load so it re-appears
+// when navigating between artists that each have unseen content.
+const unseenBanner = ref(false)
+
 watch(slug, async (s) => {
   if (!s) return
   await store.load(s)
@@ -67,6 +87,7 @@ watch(slug, async (s) => {
     ? `${store.overview.name} — FabledCurator`
     : 'FabledCurator'
   tab.value = resolve()
+  unseenBanner.value = (store.overview?.unseen_count_at_visit || 0) > 0
 }, { immediate: true })
 </script>
 
@@ -74,4 +95,7 @@ watch(slug, async (s) => {
 .fc-artist__loading {
   display: flex; justify-content: center; padding: 64px 0;
 }
+.fc-artist__unseen-msg {
+  font-size: 14px;
+}
 </style>

frontend/src/views/GalleryView.vue

@@ -12,6 +12,7 @@
     <div class="fc-gallery-layout">
       <div class="fc-gallery-layout__main">
         <PostInfoHeader />
+        <GalleryFilterBar v-if="store.filter.post_id == null" />
         <EmptyState v-if="store.isEmpty" />
         <GalleryGrid v-else @open="openImage" />
       </div>
@@ -28,6 +29,7 @@ import { useRoute } from 'vue-router'
 import { useGalleryStore } from '../stores/gallery.js'
 import { useModalStore } from '../stores/modal.js'
 import GalleryGrid from '../components/gallery/GalleryGrid.vue'
+import GalleryFilterBar from '../components/gallery/GalleryFilterBar.vue'
 import TimelineSidebar from '../components/gallery/TimelineSidebar.vue'
 import EmptyState from '../components/gallery/EmptyState.vue'
 import PostInfoHeader from '../components/gallery/PostInfoHeader.vue'
@@ -39,25 +41,13 @@ const modal = useModalStore()
 const sel = useGallerySelectionStore()
 const route = useRoute()
 
-onMounted(async () => {
+// The URL query is the single source of truth for filters. Apply it on
-  const postId = parseInt(route.query.post_id, 10)
+// mount and on any query change (filter bar pushes, back button, deep-link).
-  const tagId = parseInt(route.query.tag_id, 10)
+onMounted(() => store.applyFilterFromQuery(route.query))
-  if (!isNaN(postId)) store.setPostFilter(postId)
-  else if (!isNaN(tagId)) store.setTagFilter(tagId)
-  await store.loadInitial()
-  await store.loadTimeline()
-})
 
-watch(() => route.query.tag_id, (q) => {
+watch(() => route.query, (q) => {
   sel.clear()  // result set changed — selected ids are no longer valid
-  const tagId = parseInt(q, 10)
+  store.applyFilterFromQuery(q)
-  store.setTagFilter(isNaN(tagId) ? null : tagId)
-})
-
-watch(() => route.query.post_id, (q) => {
-  sel.clear()  // result set changed — selected ids are no longer valid
-  const postId = parseInt(q, 10)
-  store.setPostFilter(isNaN(postId) ? null : postId)
 })
 
 function openImage(id) {

frontend/src/views/TagsView.vue

@@ -8,7 +8,7 @@
         placeholder="Search tags" clearable style="max-width: 320px;"
       />
       <v-chip-group
-        v-model="kind" selected-class="text-accent" mandatory="false"
+        v-model="kind" selected-class="text-accent" :mandatory="false"
       >
         <v-chip v-for="k in KINDS" :key="k" :value="k" filter size="small">
           {{ k }}
@@ -28,6 +28,7 @@
         v-for="c in store.cards" :key="c.id" :card="c"
         @open="openTag" @rename="onRename" @manage="onManage" @read="onRead"
         @merge-with="onMergeWith" @delete="onDeleteTag"
+        @set-fandom="onSetFandom"
       />
     </div>
 
@@ -85,6 +86,13 @@
         : ''"
       @confirm="onDeleteTagConfirm"
     />
+
+    <v-dialog v-model="fandomDialogOpen" max-width="460">
+      <FandomSetDialog
+        v-if="fandomTarget" :tag="fandomTarget"
+        @updated="onFandomUpdated" @cancel="fandomDialogOpen = false"
+      />
+    </v-dialog>
   </v-container>
 </template>
 
@@ -98,6 +106,7 @@ import { useInfiniteScroll } from '../composables/useInfiniteScroll.js'
 import TagCard from '../components/discovery/TagCard.vue'
 import MergeConfirmDialog from '../components/discovery/MergeConfirmDialog.vue'
 import DestructiveConfirmModal from '../components/modal/DestructiveConfirmModal.vue'
+import FandomSetDialog from '../components/modal/FandomSetDialog.vue'
 
 // Must stay a subset of the backend TagKind enum (character, fandom,
 // general, series, archive, post). 'fandom' is this model's
@@ -140,6 +149,18 @@ function openTag(tagId) {
   router.push({ name: 'gallery', query: { tag_id: tagId } })
 }
 
+// Character fandom editing (dots-menu → FandomSetDialog).
+const fandomDialogOpen = ref(false)
+const fandomTarget = ref(null)
+function onSetFandom(card) {
+  fandomTarget.value = card
+  fandomDialogOpen.value = true
+}
+function onFandomUpdated() {
+  fandomDialogOpen.value = false
+  store.reset()  // reload so the card reflects the new fandom (or its removal)
+}
+
 function onManage(id) {
   router.push({ name: 'series-manage', params: { tagId: id } })
 }

frontend/test/components/galleryItem.spec.js

@@ -0,0 +1,28 @@
+// @vitest-environment happy-dom
+import { describe, it, expect } from 'vitest'
+
+import GalleryItem from '../../src/components/gallery/GalleryItem.vue'
+import { freshPinia, mountComponent } from '../support/mountComponent.js'
+
+describe('GalleryItem', () => {
+  const image = {
+    id: 7, sha256: 'a'.repeat(64), mime: 'image/jpeg',
+    width: 100, height: 100,
+    thumbnail_url: '/images/thumbs/aa/abc.jpg', artist: null,
+  }
+
+  it('reveals the thumbnail only once it has loaded, so tiles fade in individually', async () => {
+    const pinia = freshPinia()
+    const w = mountComponent(GalleryItem, { props: { image }, pinia })
+    const img = w.find('img')
+    expect(img.exists()).toBe(true)
+
+    // Pre-load: the tile must NOT be revealed, so it can fade in when its
+    // own thumbnail lands rather than pop together with its API batch.
+    expect(img.classes()).not.toContain('is-loaded')
+
+    await img.trigger('load')
+
+    expect(img.classes()).toContain('is-loaded')
+  })
+})

frontend/test/gallery.spec.js

@@ -9,44 +9,82 @@ function stubFetch(handler) {
       ok: status >= 200 && status < 300,
       status,
       statusText: String(status),
-      text: async () => (body == null ? '' : JSON.stringify(body))
+      text: async () => (body == null ? '' : JSON.stringify(body)),
     }
   })
 }
 
 const EMPTY = { images: [], date_groups: [], next_cursor: null }
 
-describe('gallery store: tag/post filter exclusivity', () => {
+describe('gallery store: composable filter', () => {
   beforeEach(() => setActivePinia(createPinia()))
   afterEach(() => vi.restoreAllMocks())
 
-  it('setPostFilter sets post_id and clears tag_id', () => {
+  it('applyFilterFromQuery parses the query and loadMore sends composable params', async () => {
     const s = useGalleryStore()
-    stubFetch(() => ({ status: 200, body: EMPTY }))
+    const urls = []
-    s.setTagFilter(3)
+    stubFetch((url) => {
-    expect(s.filter.tag_id).toBe(3)
+      urls.push(url)
-    s.setPostFilter(7)
+      if (url.includes('/api/tags/')) {
-    expect(s.filter.post_id).toBe(7)
+        return { status: 200, body: { id: 1, name: 'X', kind: 'general' } }
-    expect(s.filter.tag_id).toBe(null)
+      }
+      return { status: 200, body: EMPTY }
+    })
+    await s.applyFilterFromQuery({ tag_id: '1,2', artist_id: '5', media: 'video', sort: 'oldest' })
+    expect(s.filter.tag_ids).toEqual([1, 2])
+    expect(s.filter.artist_id).toBe(5)
+    expect(s.filter.media_type).toBe('video')
+    expect(s.filter.sort).toBe('oldest')
+
+    const scroll = decodeURIComponent(urls.find((u) => u.includes('/api/gallery/scroll')))
+    expect(scroll).toContain('tag_id=1,2')
+    expect(scroll).toContain('artist_id=5')
+    expect(scroll).toContain('media=video')
+    expect(scroll).toContain('sort=oldest')
   })
 
-  it('setTagFilter clears post_id', () => {
+  it('omits sort=newest and sends no filter params when empty', async () => {
-    const s = useGalleryStore()
-    stubFetch(() => ({ status: 200, body: EMPTY }))
-    s.setPostFilter(7)
-    s.setTagFilter(3)
-    expect(s.filter.tag_id).toBe(3)
-    expect(s.filter.post_id).toBe(null)
-  })
-
-  it('loadMore sends exactly the active filter param', async () => {
     const s = useGalleryStore()
     const urls = []
     stubFetch((url) => { urls.push(url); return { status: 200, body: EMPTY } })
-    s.setPostFilter(7)
+    await s.applyFilterFromQuery({})
-    await s.loadMore()
+    const scroll = urls.find((u) => u.includes('/api/gallery/scroll'))
-    const scrollCall = urls.filter(u => u.includes('/api/gallery/scroll')).pop()
+    expect(scroll).not.toContain('sort=')
-    expect(scrollCall).toContain('post_id=7')
+    expect(scroll).not.toContain('tag_id=')
-    expect(scrollCall).not.toContain('tag_id=')
+    expect(scroll).not.toContain('media=')
+  })
+
+  it('treats post_id as the exclusive filter param', async () => {
+    const s = useGalleryStore()
+    const urls = []
+    stubFetch((url) => { urls.push(url); return { status: 200, body: EMPTY } })
+    await s.applyFilterFromQuery({ post_id: '7', tag_id: '1' })
+    expect(s.filter.post_id).toBe(7)
+    const scroll = urls.find((u) => u.includes('/api/gallery/scroll'))
+    expect(scroll).toContain('post_id=7')
+    expect(scroll).not.toContain('tag_id=')
+  })
+
+  it('noteTagLabel and noteArtistLabel pre-seed chip labels', () => {
+    const s = useGalleryStore()
+    s.noteTagLabel(9, 'Rukia')
+    expect(s.tagLabels[9]).toBe('Rukia')
+    s.noteArtistLabel('Kubo')
+    expect(s.artistLabel).toBe('Kubo')
+  })
+
+  it('loadInitial issues exactly one scroll request at the initial limit', async () => {
+    const s = useGalleryStore()
+    const urls = []
+    // Non-null cursor: the old 10×serial-batch loop would have fired ten
+    // requests here. The single-fetch path must fire exactly one.
+    stubFetch((url) => {
+      urls.push(url)
+      return { status: 200, body: { images: [], date_groups: [], next_cursor: 'c1' } }
+    })
+    await s.loadInitial()
+    const scrollCalls = urls.filter((u) => u.includes('/api/gallery/scroll'))
+    expect(scrollCalls).toHaveLength(1)
+    expect(scrollCalls[0]).toContain('limit=50')
   })
 })

frontend/test/tags.spec.js

@@ -0,0 +1,61 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
+import { setActivePinia, createPinia } from 'pinia'
+import { useTagStore } from '../src/stores/tags.js'
+
+function stubFetch(handler) {
+  globalThis.fetch = vi.fn(async (url, init) => {
+    const { status, body } = handler(url, init)
+    return {
+      ok: status >= 200 && status < 300,
+      status,
+      statusText: String(status),
+      text: async () => (body == null ? '' : JSON.stringify(body)),
+    }
+  })
+}
+
+describe('tags store: setFandom', () => {
+  beforeEach(() => setActivePinia(createPinia()))
+  afterEach(() => vi.restoreAllMocks())
+
+  it('PATCHes fandom_id, and null to clear', async () => {
+    const s = useTagStore()
+    const calls = []
+    stubFetch((url, init) => {
+      calls.push({ url, init })
+      return { status: 200, body: { id: 5, name: 'Ichigo', kind: 'character', fandom_id: 9 } }
+    })
+    const body = await s.setFandom(5, 9)
+    expect(body.fandom_id).toBe(9)
+    const last = calls.at(-1)
+    expect(last.url).toContain('/api/tags/5')
+    expect(last.init.method).toBe('PATCH')
+    expect(JSON.parse(last.init.body)).toEqual({ fandom_id: 9 })
+
+    await s.setFandom(5, null)
+    expect(JSON.parse(calls.at(-1).init.body)).toEqual({ fandom_id: null })
+  })
+
+  it('sends merge: true when requested', async () => {
+    const s = useTagStore()
+    const calls = []
+    stubFetch((url, init) => {
+      calls.push({ url, init })
+      return { status: 200, body: { id: 2 } }
+    })
+    await s.setFandom(7, 3, { merge: true })
+    expect(JSON.parse(calls.at(-1).init.body)).toEqual({ fandom_id: 3, merge: true })
+  })
+
+  it('throws ApiError carrying the 409 collision body', async () => {
+    const s = useTagStore()
+    stubFetch(() => ({
+      status: 409,
+      body: { error: 'exists', target: { id: 42, name: 'Renji' } },
+    }))
+    await expect(s.setFandom(1, 2)).rejects.toMatchObject({
+      status: 409,
+      body: { target: { id: 42 } },
+    })
+  })
+})

tests/test_api_artists_directory.py

@@ -43,7 +43,8 @@ async def test_directory_card_shape(client, seeded):
     body = await resp.get_json()
     card = next(c for c in body["cards"] if c["name"] == "alice-api")
     assert set(card.keys()) == {
-        "id", "name", "slug", "is_subscription", "image_count", "preview_thumbnails",
+        "id", "name", "slug", "is_subscription", "image_count",
+        "unseen_count", "preview_thumbnails",
     }
     assert card["is_subscription"] is True
     assert card["image_count"] == 1

tests/test_api_gallery.py

@@ -17,6 +17,7 @@ async def _seed(db, count: int = 3):
             origin="imported_filesystem", integrity_status="unknown",
         )
         r.created_at = base - timedelta(minutes=i)
+        r.effective_date = r.created_at  # no post → tracks created_at (0035)
         db.add(r)
     await db.flush()
     await db.commit()
@@ -58,3 +59,21 @@ async def test_jump_requires_year_month(client):
 async def test_image_detail_404_when_missing(client):
     resp = await client.get("/api/gallery/image/99999")
     assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_scroll_sort_param_reverses(client, db):
+    await _seed(db, 3)
+    newest = await (await client.get("/api/gallery/scroll?limit=10&sort=newest")).get_json()
+    oldest = await (await client.get("/api/gallery/scroll?limit=10&sort=oldest")).get_json()
+    ids_new = [i["id"] for i in newest["images"]]
+    ids_old = [i["id"] for i in oldest["images"]]
+    assert ids_old == list(reversed(ids_new))
+
+
+@pytest.mark.asyncio
+async def test_scroll_media_param(client, db):
+    await _seed(db, 2)  # only image/jpeg seeded
+    resp = await client.get("/api/gallery/scroll?limit=10&media=video")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["images"] == []

tests/test_api_tag_merge.py

@@ -3,11 +3,57 @@ import pytest
 pytestmark = pytest.mark.integration
 
 
-async def _mk(client, name, kind):
+async def _mk(client, name, kind, fandom_id=None):
-    r = await client.post("/api/tags", json={"name": name, "kind": kind})
+    body = {"name": name, "kind": kind}
+    if fandom_id is not None:
+        body["fandom_id"] = fandom_id
+    r = await client.post("/api/tags", json=body)
     return (await r.get_json())["id"]
 
 
+@pytest.mark.asyncio
+async def test_get_tag_returns_shape_and_404(client):
+    tid = await _mk(client, "Lookup", "general")
+    resp = await client.get(f"/api/tags/{tid}")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["id"] == tid
+    assert body["name"] == "Lookup"
+    assert body["kind"] == "general"
+    resp = await client.get("/api/tags/99999999")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_patch_sets_and_clears_character_fandom(client):
+    fandom = await _mk(client, "Bleach", "fandom")
+    char = await _mk(client, "Ichigo", "character")
+    resp = await client.patch(f"/api/tags/{char}", json={"fandom_id": fandom})
+    assert resp.status_code == 200
+    assert (await resp.get_json())["fandom_id"] == fandom
+    resp = await client.patch(f"/api/tags/{char}", json={"fandom_id": None})
+    assert resp.status_code == 200
+    assert (await resp.get_json())["fandom_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_patch_fandom_collision_then_merge(client):
+    f1 = await _mk(client, "Bleach", "fandom")
+    f2 = await _mk(client, "Naruto", "fandom")
+    c1 = await _mk(client, "Renji", "character", fandom_id=f1)
+    c2 = await _mk(client, "Renji", "character", fandom_id=f2)
+    # Moving c1 into f2 collides with c2 → rich 409 (same shape as rename).
+    resp = await client.patch(f"/api/tags/{c1}", json={"fandom_id": f2})
+    assert resp.status_code == 409
+    assert (await resp.get_json())["target"]["id"] == c2
+    # Confirm → merge c1 into c2, c2 survives.
+    resp = await client.patch(
+        f"/api/tags/{c1}", json={"fandom_id": f2, "merge": True}
+    )
+    assert resp.status_code == 200
+    assert (await resp.get_json())["id"] == c2
+
+
 @pytest.mark.asyncio
 async def test_rename_collision_returns_rich_409(client):
     tgt = await _mk(client, "Canonical", "general")

tests/test_artist_visit.py

@@ -0,0 +1,175 @@
+"""ArtistVisit unseen-count badge + ArtistService.overview banner data.
+
+Covers:
+- Directory cards include `unseen_count`
+- LEFT JOIN keeps artists without a visit row (treats NULL as "never
+  visited" → all images unseen)
+- overview() returns `unseen_count_at_visit` and stamps the visit
+- find_or_create autoseeds a visit row so freshly imported content
+  doesn't show up as unseen
+- Repeat overview() returns 0 (since the previous visit just stamped
+  last_viewed_at = NOW())
+"""
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Artist, ArtistVisit, ImageRecord
+from backend.app.services.artist_directory_service import ArtistDirectoryService
+from backend.app.services.artist_service import ArtistService
+
+pytestmark = pytest.mark.integration
+
+
+_LONG_AGO = datetime(2000, 1, 1, tzinfo=UTC)
+_RECENTLY = datetime(2099, 1, 1, tzinfo=UTC)
+
+
+async def _seed_artist(db, name: str) -> Artist:
+    a = Artist(name=name, slug=name.lower().replace(" ", "-"))
+    db.add(a)
+    await db.flush()
+    return a
+
+
+async def _seed_visit(db, artist_id: int, when: datetime) -> None:
+    db.add(ArtistVisit(artist_id=artist_id, last_viewed_at=when))
+    await db.flush()
+
+
+async def _seed_image(db, artist_id: int, suffix: str, *, created_at: datetime) -> None:
+    db.add(ImageRecord(
+        path=f"/images/visit-{suffix}.jpg",
+        sha256=f"visit{suffix}".ljust(64, "0")[:64],
+        size_bytes=10, mime="image/jpeg", width=10, height=10,
+        origin="downloaded", artist_id=artist_id,
+        created_at=created_at,
+    ))
+    await db.flush()
+
+
+# --- Directory unseen_count -----------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_directory_unseen_count_zero_when_no_images(db):
+    await _seed_artist(db, "zoey-empty-visit")
+    await db.commit()
+
+    page = await ArtistDirectoryService(db).list_artists(
+        q="zoey-empty-visit", platform=None, cursor=None, limit=60,
+    )
+    target = next(c for c in page.cards if c["name"] == "zoey-empty-visit")
+    assert target["unseen_count"] == 0
+
+
+@pytest.mark.asyncio
+async def test_directory_unseen_counts_only_images_after_visit(db):
+    a = await _seed_artist(db, "alice-visit-mix")
+    await _seed_visit(db, a.id, _LONG_AGO + timedelta(days=365))
+    # Two images BEFORE the visit (seen), three AFTER (unseen).
+    for i in range(2):
+        await _seed_image(db, a.id, f"old-{i}", created_at=_LONG_AGO)
+    for i in range(3):
+        await _seed_image(db, a.id, f"new-{i}", created_at=_RECENTLY)
+    await db.commit()
+
+    page = await ArtistDirectoryService(db).list_artists(
+        q="alice-visit-mix", platform=None, cursor=None, limit=60,
+    )
+    target = next(c for c in page.cards if c["name"] == "alice-visit-mix")
+    assert target["image_count"] == 5
+    assert target["unseen_count"] == 3
+
+
+@pytest.mark.asyncio
+async def test_directory_treats_missing_visit_as_never_visited(db):
+    """No ArtistVisit row → defensive count of all images as unseen.
+
+    Shouldn't happen in practice (migration 0034 seeds existing
+    artists, find_or_create autoseeds new ones), but the directory
+    query must not regress to "0 unseen" if a row is missing.
+    """
+    a = await _seed_artist(db, "bob-no-visit")
+    for i in range(4):
+        await _seed_image(db, a.id, f"orphan-{i}", created_at=_RECENTLY)
+    await db.commit()
+
+    page = await ArtistDirectoryService(db).list_artists(
+        q="bob-no-visit", platform=None, cursor=None, limit=60,
+    )
+    target = next(c for c in page.cards if c["name"] == "bob-no-visit")
+    assert target["unseen_count"] == 4
+
+
+# --- overview() marks visit + returns count -------------------------------
+
+
+@pytest.mark.asyncio
+async def test_overview_returns_unseen_count_at_visit_and_stamps_now(db):
+    a = await _seed_artist(db, "carol-stamp")
+    await _seed_visit(db, a.id, _LONG_AGO)
+    await _seed_image(db, a.id, "stamp-1", created_at=_RECENTLY)
+    await _seed_image(db, a.id, "stamp-2", created_at=_RECENTLY)
+    await db.commit()
+
+    data = await ArtistService(db).overview("carol-stamp")
+    assert data is not None
+    assert data["unseen_count_at_visit"] == 2
+
+    # last_viewed_at advanced to NOW() — directly check the row.
+    visit_at = (await db.execute(
+        select(ArtistVisit.last_viewed_at).where(ArtistVisit.artist_id == a.id)
+    )).scalar_one()
+    assert visit_at > _LONG_AGO
+
+
+@pytest.mark.asyncio
+async def test_overview_repeat_call_returns_zero(db):
+    a = await _seed_artist(db, "dana-repeat")
+    await _seed_visit(db, a.id, _LONG_AGO)
+    await _seed_image(db, a.id, "repeat-1", created_at=_LONG_AGO + timedelta(days=1))
+    await db.commit()
+
+    first = await ArtistService(db).overview("dana-repeat")
+    assert first is not None
+    assert first["unseen_count_at_visit"] == 1
+
+    # Second call: no new images, visit just stamped → count is 0.
+    second = await ArtistService(db).overview("dana-repeat")
+    assert second is not None
+    assert second["unseen_count_at_visit"] == 0
+
+
+# --- find_or_create autoseeds the visit row ------------------------------
+
+
+@pytest.mark.asyncio
+async def test_find_or_create_autoseeds_visit_row(db):
+    artist, created = await ArtistService(db).find_or_create("Eve-Autoseed")
+    assert created is True
+
+    row = (await db.execute(
+        select(ArtistVisit.last_viewed_at).where(ArtistVisit.artist_id == artist.id)
+    )).scalar_one_or_none()
+    assert row is not None
+
+
+@pytest.mark.asyncio
+async def test_find_or_create_existing_does_not_reset_visit(db):
+    """Calling find_or_create on an existing artist returns it as-is —
+    must NOT touch the visit row's timestamp."""
+    a = await _seed_artist(db, "Frank-Existing")
+    await _seed_visit(db, a.id, _LONG_AGO)
+    await db.commit()
+
+    artist, created = await ArtistService(db).find_or_create("Frank-Existing")
+    assert created is False
+    assert artist.id == a.id
+
+    visit_at = (await db.execute(
+        select(ArtistVisit.last_viewed_at).where(ArtistVisit.artist_id == a.id)
+    )).scalar_one()
+    assert visit_at == _LONG_AGO

tests/test_download_service.py

@@ -613,3 +613,81 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
     assert len(thumb_calls) == 2
     assert len(ml_calls) == 2
     assert sorted(thumb_calls) == sorted(ml_calls)
+
+
+@pytest.mark.asyncio
+async def test_releases_db_connections_before_subprocess(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    """Phase 1's DB connections must be released BEFORE the (up to ~19.5-min
+    in backfill) gallery-dl subprocess, so they don't idle-die and strand
+    phase 3 with ConnectionDoesNotExistError (Anduo #40014). Spy on the
+    async session close and assert it happened before gdl.download runs;
+    phase 3 must still finalize the event."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+
+    closed = {"async": False}
+    orig_close = db.close
+
+    async def spy_close():
+        closed["async"] = True
+        await orig_close()
+
+    monkeypatch.setattr(db, "close", spy_close)
+
+    seen = {}
+
+    async def fake_download(url, **k):
+        seen["closed_before_download"] = closed["async"]
+        return _make_fake_dl_result(success=True, written_paths=[], stdout="")
+
+    fake_gdl = MagicMock()
+    fake_gdl.download = fake_download
+    fake_gdl._compute_run_stats = lambda *a, **k: {
+        "exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
+        "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
+    }
+    fake_gdl._extract_errors_warnings = lambda *a, **k: ""
+    fake_gdl._truncate_log = lambda x, **k: x
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=tmp_path, import_root=tmp_path,
+        thumbnailer=Thumbnailer(images_root=tmp_path), settings=sync_settings,
+    )
+    cred_service = CredentialService(
+        db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
+    )
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    event_id = await svc.download_source(source.id)
+
+    assert seen["closed_before_download"] is True
+    ev = (await db.execute(
+        select(DownloadEvent).where(DownloadEvent.id == event_id)
+    )).scalar_one()
+    assert ev.status == "ok"
+
+
+@pytest.mark.asyncio
+async def test_async_task_engine_uses_nullpool():
+    """The per-task async engine must use NullPool so phase 3 re-acquires a
+    fresh connection instead of a pooled one the server reaped during the
+    long subprocess (Anduo #40014)."""
+    from sqlalchemy.pool import NullPool
+
+    from backend.app.tasks._async_session import async_session_factory
+
+    _factory, engine = async_session_factory()
+    try:
+        assert isinstance(engine.sync_engine.pool, NullPool)
+    finally:
+        await engine.dispose()

tests/test_download_source_task.py

@@ -1,9 +1,15 @@
-"""Smoke test that the FC-3c Celery task is registered and routed correctly.
+"""Tests for the FC-3c download_source Celery task wrapper.
 
-Mirrors test_tasks_register.py — Celery's `include=[...]` is lazy, so
+Covers registration/routing (smoke) plus the soft-time-limit salvage
-the task module must be imported explicitly to trigger registration.
+path (audit 2026-06-03, Anduo #39912): a SoftTimeLimitExceeded must not
+leave the DownloadEvent stranded empty for the recovery sweep.
 """
 
+from datetime import UTC, datetime
+
+import pytest
+from sqlalchemy import select
+
 # Side-effect import: the @celery.task decorator on download_source fires
 # at module import time and registers the task with the global instance.
 import backend.app.tasks.download  # noqa: F401
@@ -18,3 +24,164 @@ def test_download_source_routes_to_download_queue():
     routes = celery.conf.task_routes
     assert "backend.app.tasks.download.*" in routes
     assert routes["backend.app.tasks.download.*"]["queue"] == "download"
+
+
+def test_timeout_ladder_keeps_subprocess_budgets_under_soft_limit():
+    """Regression guard for Anduo #39912: every gallery-dl subprocess
+    budget MUST sit below download_source's Celery soft limit so
+    subprocess.run raises its own TimeoutExpired (which captures partial
+    logs + finalizes the event) BEFORE Celery's SoftTimeLimitExceeded
+    preempts it. soft must in turn sit below the hard SIGKILL cap."""
+    from backend.app.services.gallery_dl import (
+        _DEFAULT_GDL_TIMEOUT_SECONDS,
+        BACKFILL_TIMEOUT_SECONDS,
+    )
+    from backend.app.tasks.download import (
+        DOWNLOAD_HARD_TIME_LIMIT,
+        DOWNLOAD_SOFT_TIME_LIMIT,
+    )
+
+    assert _DEFAULT_GDL_TIMEOUT_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT
+    assert BACKFILL_TIMEOUT_SECONDS < DOWNLOAD_SOFT_TIME_LIMIT
+    assert DOWNLOAD_SOFT_TIME_LIMIT < DOWNLOAD_HARD_TIME_LIMIT
+
+
+def test_decorated_limits_match_module_constants():
+    """The @celery.task decorator must use the audited constants, not
+    drifted literals."""
+    from backend.app.tasks.download import (
+        DOWNLOAD_HARD_TIME_LIMIT,
+        DOWNLOAD_SOFT_TIME_LIMIT,
+        download_source,
+    )
+
+    assert download_source.soft_time_limit == DOWNLOAD_SOFT_TIME_LIMIT
+    assert download_source.time_limit == DOWNLOAD_HARD_TIME_LIMIT
+
+
+def _seed_running_event(db_sync, *, slug, backfill, failures=0):
+    from backend.app.models import Artist, DownloadEvent, Source
+
+    artist = Artist(name=slug, slug=slug)
+    db_sync.add(artist)
+    db_sync.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url=f"https://patreon.com/{slug}", enabled=True,
+        config_overrides={}, backfill_runs_remaining=backfill,
+        consecutive_failures=failures,
+    )
+    db_sync.add(source)
+    db_sync.flush()
+    ev = DownloadEvent(
+        source_id=source.id, status="running",
+        started_at=datetime.now(UTC),
+    )
+    db_sync.add(ev)
+    db_sync.flush()
+    return source, ev.id
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_finalize_soft_limited_flips_event_and_decrements_backfill(db_sync):
+    from backend.app.models import DownloadEvent, Source
+    from backend.app.tasks.download import _finalize_soft_limited
+
+    source, event_id = _seed_running_event(
+        db_sync, slug="anduo", backfill=2, failures=0,
+    )
+
+    _finalize_soft_limited(db_sync, source.id)
+
+    status, finished_at, error, meta = db_sync.execute(
+        select(
+            DownloadEvent.status, DownloadEvent.finished_at,
+            DownloadEvent.error, DownloadEvent.metadata_,
+        ).where(DownloadEvent.id == event_id)
+    ).one()
+    assert status == "error"
+    assert finished_at is not None
+    assert "soft time limit" in (error or "").lower()
+    assert meta.get("error_type") == "timeout"
+    assert meta.get("soft_time_limited") is True
+
+    backfill, failures, error_type = db_sync.execute(
+        select(
+            Source.backfill_runs_remaining, Source.consecutive_failures,
+            Source.error_type,
+        ).where(Source.id == source.id)
+    ).one()
+    assert backfill == 1          # 2 -> 1, source self-heals toward tick mode
+    assert failures == 1          # 0 -> 1, mirrors phase-3 source-health write
+    assert error_type == "timeout"
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_finalize_soft_limited_is_noop_without_running_event(db_sync):
+    """A benign late soft-limit (phase 3 already committed → no running
+    event) must not touch source health or backfill budget."""
+    from backend.app.models import DownloadEvent, Source
+    from backend.app.tasks.download import _finalize_soft_limited
+
+    source, event_id = _seed_running_event(
+        db_sync, slug="noevent", backfill=3, failures=0,
+    )
+    # Simulate phase 3 having already finalized the event.
+    ev = db_sync.get(DownloadEvent, event_id)
+    ev.status = "ok"
+    db_sync.flush()
+
+    _finalize_soft_limited(db_sync, source.id)
+
+    backfill, failures, error_type = db_sync.execute(
+        select(
+            Source.backfill_runs_remaining, Source.consecutive_failures,
+            Source.error_type,
+        ).where(Source.id == source.id)
+    ).one()
+    assert backfill == 3          # untouched
+    assert failures == 0          # untouched
+    assert error_type is None
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_download_source_catches_soft_limit_and_salvages_event(
+    db_sync, monkeypatch,
+):
+    """End-to-end wiring: when the inner run raises SoftTimeLimitExceeded,
+    download_source's handler must flip the in-flight event to error
+    instead of letting it strand. Uses eager mode + a stubbed asyncio.run
+    so no real gallery-dl subprocess is spawned."""
+    from celery.exceptions import SoftTimeLimitExceeded
+
+    import backend.app.tasks.download as dl
+    from backend.app.models import DownloadEvent
+
+    monkeypatch.setattr(celery.conf, "task_always_eager", True)
+    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+
+    source, event_id = _seed_running_event(
+        db_sync, slug="anduowire", backfill=1, failures=0,
+    )
+    # The task opens a fresh session via _sync_session_factory(); commit
+    # so that session can see the seeded running event.
+    db_sync.commit()
+
+    def _raise(coro=None, *a, **k):
+        # Close the un-awaited coroutine so pytest output stays pristine.
+        if coro is not None and hasattr(coro, "close"):
+            coro.close()
+        raise SoftTimeLimitExceeded("simulated soft limit")
+
+    monkeypatch.setattr(dl.asyncio, "run", _raise)
+
+    with pytest.raises(SoftTimeLimitExceeded):
+        dl.download_source.delay(source.id).get(propagate=True)
+
+    status = db_sync.execute(
+        select(DownloadEvent.status).where(DownloadEvent.id == event_id)
+    ).scalar_one()
+    assert status == "error"

tests/test_gallery_artist_payload.py

@@ -14,6 +14,7 @@ async def _img(db, n):
         origin="imported_filesystem", integrity_status="unknown",
     )
     rec.created_at = datetime.now(UTC) - timedelta(minutes=n)
+    rec.effective_date = rec.created_at  # no post → tracks created_at (0035)
     db.add(rec)
     await db.flush()
     return rec

tests/test_gallery_provenance_filter.py

@@ -22,6 +22,7 @@ async def _img(db, n):
         origin="imported_filesystem", integrity_status="unknown",
     )
     rec.created_at = base - timedelta(minutes=n)
+    rec.effective_date = rec.created_at  # no primary post → tracks created_at (0035)
     db.add(rec)
     await db.flush()
     return rec
@@ -80,10 +81,10 @@ async def test_scroll_artist_id_filter(db):
 
 
 @pytest.mark.asyncio
-async def test_mutually_exclusive_filters_raise(db):
+async def test_post_id_excludes_other_filters(db):
     svc = GalleryService(db)
     with pytest.raises(ValueError):
-        await svc.scroll(cursor=None, limit=10, tag_id=1, post_id=2)
+        await svc.scroll(cursor=None, limit=10, tag_ids=[1], post_id=2)
 
 
 @pytest.mark.asyncio

tests/test_gallery_service.py

@@ -32,6 +32,8 @@ async def _seed_images(db, count: int, sha_prefix: str = "0") -> list[ImageRecor
             integrity_status="unknown",
         )
         r.created_at = base - timedelta(minutes=i)
+        # No post → effective_date == created_at (alembic 0035 denorm).
+        r.effective_date = r.created_at
         db.add(r)
         records.append(r)
     await db.flush()
@@ -85,7 +87,7 @@ async def test_scroll_with_tag_filter(db):
     )
 
     svc = GalleryService(db)
-    page = await svc.scroll(cursor=None, limit=10, tag_id=tag.id)
+    page = await svc.scroll(cursor=None, limit=10, tag_ids=[tag.id])
     assert len(page.images) == 1
     assert page.images[0].id == images[0].id
 
@@ -166,6 +168,9 @@ async def _seed_image_with_post(
         primary_post_id=post.id,
     )
     img.created_at = image_created_at
+    # Mirror the importer's denorm (alembic 0035): a linked post with a date
+    # sets effective_date to that date, else it falls back to created_at.
+    img.effective_date = post_date or image_created_at
     db.add(img)
     await db.flush()
     return img, post
@@ -201,6 +206,7 @@ async def test_scroll_sorts_by_post_date_when_available(db):
         origin="imported_filesystem", integrity_status="unknown",
     )
     img_c.created_at = base_import - timedelta(days=5)
+    img_c.effective_date = img_c.created_at  # no post → tracks created_at
     db.add(img_c)
     await db.flush()
 
@@ -266,3 +272,53 @@ async def test_get_image_with_tags_includes_posted_at_when_present(db):
     assert payload["posted_at"] is not None
     # Image's own created_at is still surfaced separately.
     assert payload["created_at"] != payload["posted_at"]
+
+
+@pytest.mark.asyncio
+async def test_scroll_multi_tag_and(db):
+    images = await _seed_images(db, 5)
+    a = Tag(name="aa", kind=TagKind.general)
+    b = Tag(name="bb", kind=TagKind.general)
+    db.add_all([a, b])
+    await db.flush()
+    # images[0] carries BOTH tags; images[1] carries only a.
+    await db.execute(image_tag.insert().values([
+        {"image_record_id": images[0].id, "tag_id": a.id, "source": "manual"},
+        {"image_record_id": images[0].id, "tag_id": b.id, "source": "manual"},
+        {"image_record_id": images[1].id, "tag_id": a.id, "source": "manual"},
+    ]))
+    svc = GalleryService(db)
+    both = await svc.scroll(cursor=None, limit=10, tag_ids=[a.id, b.id])
+    assert [i.id for i in both.images] == [images[0].id]  # AND: only the both-tagged
+    just_a = await svc.scroll(cursor=None, limit=10, tag_ids=[a.id])
+    assert {i.id for i in just_a.images} == {images[0].id, images[1].id}
+
+
+@pytest.mark.asyncio
+async def test_scroll_media_filter(db):
+    imgs = await _seed_images(db, 2)  # image/jpeg
+    vid = ImageRecord(
+        path="/images/test/v.mp4", sha256="v" * 64, size_bytes=1,
+        mime="video/mp4", width=1, height=1,
+        origin="imported_filesystem", integrity_status="unknown",
+    )
+    vid.created_at = _now()
+    vid.effective_date = vid.created_at
+    db.add(vid)
+    await db.flush()
+    svc = GalleryService(db)
+    vids = await svc.scroll(cursor=None, limit=10, media_type="video")
+    assert [i.id for i in vids.images] == [vid.id]
+    pics = await svc.scroll(cursor=None, limit=10, media_type="image")
+    assert {i.id for i in pics.images} == {i.id for i in imgs}
+
+
+@pytest.mark.asyncio
+async def test_scroll_sort_oldest_reverses_order(db):
+    images = await _seed_images(db, 4)  # images[0] newest ... images[3] oldest
+    svc = GalleryService(db)
+    newest = await svc.scroll(cursor=None, limit=10)
+    oldest = await svc.scroll(cursor=None, limit=10, sort="oldest")
+    newest_ids = [i.id for i in newest.images]
+    assert newest_ids[0] == images[0].id
+    assert [i.id for i in oldest.images] == list(reversed(newest_ids))

tests/test_sidecar_import.py

@@ -94,6 +94,9 @@ def test_sidecar_creates_provenance(importer, import_layout):
     prov = importer.session.execute(select(ImageProvenance)).scalar_one()
     assert prov.image_record_id == rec.id and prov.post_id == post.id
     assert rec.primary_post_id == post.id
+    # Denormalized gallery sort key (alembic 0035) tracks the primary post's
+    # publish date so /scroll orders off ix_image_record_effective_date.
+    assert rec.effective_date == post.post_date
 
 
 def test_reimport_same_post_idempotent(importer, import_layout):

tests/test_tag_service.py

@@ -1,7 +1,11 @@
 import pytest
 
 from backend.app.models import TagKind
-from backend.app.services.tag_service import TagService, TagValidationError
+from backend.app.services.tag_service import (
+    TagMergeConflict,
+    TagService,
+    TagValidationError,
+)
 
 pytestmark = pytest.mark.integration
 
@@ -101,3 +105,64 @@ async def test_autocomplete_empty_query_returns_nothing(db):
     svc = TagService(db)
     await svc.find_or_create("Foo", TagKind.artist)
     assert await svc.autocomplete("") == []
+
+
+@pytest.mark.asyncio
+async def test_set_fandom_assigns_changes_and_clears(db):
+    svc = TagService(db)
+    f1 = await svc.find_or_create("Bleach", TagKind.fandom)
+    f2 = await svc.find_or_create("Naruto", TagKind.fandom)
+    char = await svc.find_or_create("Ichigo", TagKind.character)
+    assert char.fandom_id is None
+    assert (await svc.set_fandom(char.id, f1.id)).fandom_id == f1.id
+    assert (await svc.set_fandom(char.id, f2.id)).fandom_id == f2.id
+    assert (await svc.set_fandom(char.id, None)).fandom_id is None
+
+
+@pytest.mark.asyncio
+async def test_set_fandom_rejects_non_character(db):
+    svc = TagService(db)
+    fandom = await svc.find_or_create("Naruto", TagKind.fandom)
+    series = await svc.find_or_create("Arc 1", TagKind.series)
+    with pytest.raises(TagValidationError, match="character"):
+        await svc.set_fandom(series.id, fandom.id)
+
+
+@pytest.mark.asyncio
+async def test_set_fandom_rejects_non_fandom_reference(db):
+    svc = TagService(db)
+    general = await svc.find_or_create("misc", TagKind.general)
+    char = await svc.find_or_create("Rukia", TagKind.character)
+    with pytest.raises(TagValidationError, match="fandom"):
+        await svc.set_fandom(char.id, general.id)
+
+
+@pytest.mark.asyncio
+async def test_set_fandom_collision_raises_merge_conflict(db):
+    svc = TagService(db)
+    f1 = await svc.find_or_create("Bleach", TagKind.fandom)
+    f2 = await svc.find_or_create("Naruto", TagKind.fandom)
+    c1 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f1.id)
+    c2 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f2.id)
+    with pytest.raises(TagMergeConflict) as ei:
+        await svc.set_fandom(c1.id, f2.id)  # collides with c2 in f2
+    assert ei.value.target_id == c2.id
+
+
+@pytest.mark.asyncio
+async def test_set_fandom_merge_resolves_collision(db):
+    from sqlalchemy import select
+
+    from backend.app.models import Tag
+
+    svc = TagService(db)
+    f1 = await svc.find_or_create("Bleach", TagKind.fandom)
+    f2 = await svc.find_or_create("Naruto", TagKind.fandom)
+    c1 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f1.id)
+    c2 = await svc.find_or_create("Renji", TagKind.character, fandom_id=f2.id)
+    survivor = await svc.set_fandom(c1.id, f2.id, merge=True)
+    assert survivor.id == c2.id  # merged INTO the existing character
+    gone = (
+        await db.execute(select(Tag).where(Tag.id == c1.id))
+    ).scalar_one_or_none()
+    assert gone is None