Merge pull request 'v26.05.25.1: maintenance sweep + Camie v2 + corrupt-file handling + post-date gallery + clear-stuck escape hatch' (#11) from dev into main

2026-05-25 12:57:46 -04:00
parent b8dce6c483 3a359f6c5e
commit 2879ac6f2b
22 changed files with 973 additions and 149 deletions
@@ -42,6 +42,8 @@ async def scroll():
                    "width": i.width,
                    "height": i.height,
                    "created_at": i.created_at.isoformat(),
+                    "posted_at": i.posted_at.isoformat() if i.posted_at else None,
+                    "effective_date": i.effective_date.isoformat(),
                    "thumbnail_url": i.thumbnail_url,
                    "artist": i.artist,
                }
@@ -120,6 +120,83 @@ async def retry_failed():
    return jsonify({"retried": len(failed_ids)})


+@import_admin_bp.route("/clear-stuck", methods=["POST"])
+async def clear_stuck():
+    """Force any non-terminal ImportTask (status in pending/queued/
+    processing) to 'failed' AND finalize any ImportBatch that ends up
+    with no active children. Escape hatch for the operator when the
+    automatic recover_interrupted_tasks sweep keeps re-queueing the
+    same stuck row forever (e.g., underlying file is genuinely broken
+    and the import keeps OSError-looping at PIL load).
+
+    Idempotent + non-destructive: rows survive as 'failed' so the
+    Retry-Failed button can re-attempt them once whatever was broken
+    is fixed. Banked 2026-05-25 — operator hit 3 large PNGs that
+    autoretry-looped for 2 days after a corrupt-data PIL OSError.
+    """
+    async with get_session() as session:
+        stuck_ids = (
+            await session.execute(
+                select(ImportTask.id).where(
+                    ImportTask.status.in_(["pending", "queued", "processing"])
+                )
+            )
+        ).scalars().all()
+        if stuck_ids:
+            await session.execute(
+                update(ImportTask)
+                .where(ImportTask.id.in_(stuck_ids))
+                .values(
+                    status="failed",
+                    finished_at=datetime.now(UTC),
+                    error=(
+                        "manually cleared via /api/import/clear-stuck "
+                        "— stuck in non-terminal state; retry once "
+                        "underlying cause (corrupt file, missing model, "
+                        "etc.) is resolved"
+                    ),
+                )
+            )
+
+        # Finalize any 'running' ImportBatch that no longer has any
+        # active children. The "Scanning..." banner is driven by
+        # /api/import/status finding a running batch; left untouched,
+        # it would persist forever after the stuck-task clear.
+        running_batches = (
+            await session.execute(
+                select(ImportBatch.id).where(ImportBatch.status == "running")
+            )
+        ).scalars().all()
+        finalized_batches = 0
+        for batch_id in running_batches:
+            still_active = (
+                await session.execute(
+                    select(ImportTask.id)
+                    .where(ImportTask.batch_id == batch_id)
+                    .where(ImportTask.status.in_(
+                        ["pending", "queued", "processing"]
+                    ))
+                    .limit(1)
+                )
+            ).scalar_one_or_none()
+            if still_active is None:
+                await session.execute(
+                    update(ImportBatch)
+                    .where(ImportBatch.id == batch_id)
+                    .values(
+                        status="complete",
+                        finished_at=datetime.now(UTC),
+                    )
+                )
+                finalized_batches += 1
+        await session.commit()
+
+    return jsonify({
+        "tasks_failed": len(stuck_ids),
+        "batches_finalized": finalized_batches,
+    })
+
+
@import_admin_bp.route("/clear-completed", methods=["POST"])
 async def clear_completed():
    body = await request.get_json(silent=True) or {}
@@ -25,12 +25,31 @@ def _snapshot(repo_id: str, dest: Path, allow_patterns: list[str] | None) -> Non


 def ensure_camie() -> None:
+    """Fetch Camie v2 weights + metadata.
+
+    v2 layout (HuggingFace Camais03/camie-tagger-v2): the ONNX file is
+    named camie-tagger-v2.onnx (not model.onnx) and tags ship inside
+    camie-tagger-v2-metadata.json (not selected_tags.csv). Both at root.
+    The repo also contains app/, game/, training/, images/ subdirs full
+    of setup/demo files we don't need — allow_patterns scopes the fetch
+    to just the inference essentials (~790 MB instead of ~2 GB).
+    """
    dest = MODEL_ROOT / "camie"
-    if (dest / "model.onnx").is_file() and (dest / "selected_tags.csv").is_file():
+    model_file = dest / "camie-tagger-v2.onnx"
+    meta_file = dest / "camie-tagger-v2-metadata.json"
+    if model_file.is_file() and meta_file.is_file():
        print(f"[download_models] Camie present at {dest}")
        return
    print(f"[download_models] Fetching {CAMIE_REPO} -> {dest}")
-    _snapshot(CAMIE_REPO, dest, ["model.onnx", "selected_tags.csv", "*.json"])
+    _snapshot(
+        CAMIE_REPO, dest,
+        [
+            "camie-tagger-v2.onnx",
+            "camie-tagger-v2-metadata.json",
+            "config.json",
+            "config.yaml",
+        ],
+    )


 def ensure_siglip() -> None:
@@ -1,26 +1,34 @@
 """Cursor-paginated gallery queries.

-Cursor format: opaque base64-encoded "<iso8601_created_at>:<image_id>".
-Pagination key is (created_at DESC, id DESC) so we don't drift when new
-imports arrive between page loads. Decoding rejects malformed cursors with
-a ValueError; the API layer translates that to HTTP 400.
+Cursor format: opaque base64-encoded "<iso8601_effective_date>:<image_id>".
+
+Pagination key is (effective_date DESC, id DESC) where effective_date is
+COALESCE(post.post_date, image_record.created_at) so the gallery surfaces
+images by ORIGINAL publish date when known, falling back to FC's scan
+date. Important for migrated content: ~57k IR images scanned in a single
+week would otherwise all share the same created_at and pile up in one
+month bucket. The effective_date spreads them across the years they
+were originally published.
+
+Decoding rejects malformed cursors with a ValueError; the API layer
+translates that to HTTP 400.
 """

 import base64
 from dataclasses import dataclass
 from datetime import datetime

-from sqlalchemy import and_, exists, func, or_, select
+from sqlalchemy import Select, and_, exists, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession

-from ..models import Artist, ImageProvenance, ImageRecord, Source, Tag
+from ..models import Artist, ImageProvenance, ImageRecord, Post, Source, Tag
 from ..models.tag import image_tag

 CURSOR_SEPARATOR = "|"


-def encode_cursor(created_at: datetime, image_id: int) -> str:
-    raw = f"{created_at.isoformat()}{CURSOR_SEPARATOR}{image_id}"
+def encode_cursor(effective_date: datetime, image_id: int) -> str:
+    raw = f"{effective_date.isoformat()}{CURSOR_SEPARATOR}{image_id}"
    return base64.urlsafe_b64encode(raw.encode()).decode()


@@ -33,6 +41,26 @@ def decode_cursor(cursor: str) -> tuple[datetime, int]:
        raise ValueError(f"invalid cursor: {cursor!r}") from exc


+def _effective_date_col():
+    """SQL expression: COALESCE(post.post_date, image_record.created_at).
+
+    Used as the canonical sort/group/filter key across the gallery so
+    images backfilled with primary_post_id (e.g. via tag_apply phase 4)
+    surface at their original publish date, not their FC import date.
+    Images without a Post (or with Post.post_date NULL) fall back to
+    image_record.created_at and still order coherently against
+    post-attached ones.
+    """
+    return func.coalesce(Post.post_date, ImageRecord.created_at)
+
+
+def _outer_join_primary_post(stmt: Select) -> Select:
+    """LEFT JOIN Post on ImageRecord.primary_post_id so the COALESCE
+    above sees Post.post_date when available. Images without a post
+    survive the join as NULL on the Post side; COALESCE handles it."""
+    return stmt.outerjoin(Post, Post.id == ImageRecord.primary_post_id)
+
+
@dataclass(frozen=True)
 class GalleryImage:
    id: int
@@ -41,7 +69,9 @@ class GalleryImage:
    mime: str
    width: int | None
    height: int | None
-    created_at: datetime
+    created_at: datetime          # FC's row-insert time
+    effective_date: datetime      # COALESCE(post.post_date, created_at)
+    posted_at: datetime | None    # post.post_date if known, else None
    thumbnail_url: str
    artist: dict | None = None

@@ -78,7 +108,7 @@ def _require_single_filter(tag_id, post_id, artist_id) -> None:
 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
-    (created_at DESC, id DESC) cursor ordering is unaffected."""
+    (effective_date DESC, id DESC) cursor ordering is unaffected."""
    if post_id is not None:
        return exists().where(
            ImageProvenance.image_record_id == ImageRecord.id,
@@ -125,7 +155,9 @@ class GalleryService:
            raise ValueError("limit must be between 1 and 200")
        _require_single_filter(tag_id, post_id, artist_id)

-        stmt = select(ImageRecord)
+        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 = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
                image_tag.c.tag_id == tag_id
@@ -138,34 +170,38 @@ class GalleryService:
            cur_ts, cur_id = decode_cursor(cursor)
            stmt = stmt.where(
                or_(
-                    ImageRecord.created_at < cur_ts,
-                    and_(ImageRecord.created_at == cur_ts, ImageRecord.id < cur_id),
+                    eff < cur_ts,
+                    and_(eff == cur_ts, ImageRecord.id < cur_id),
                )
            )

-        stmt = stmt.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()).limit(limit + 1)
-        rows = (await self.session.execute(stmt)).scalars().all()
+        stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(limit + 1)
+        rows = (await self.session.execute(stmt)).all()

        next_cursor = None
        if len(rows) > limit:
-            last = rows[limit - 1]
-            next_cursor = encode_cursor(last.created_at, last.id)
+            last_record, _last_posted_at, last_eff = rows[limit - 1]
+            next_cursor = encode_cursor(last_eff, last_record.id)
            rows = rows[:limit]

-        artists = await _artists_for(self.session, [r.id for r in rows])
+        artists = await _artists_for(
+            self.session, [r[0].id for r in rows]
+        )
        images = [
            GalleryImage(
-                id=r.id,
-                path=r.path,
-                sha256=r.sha256,
-                mime=r.mime,
-                width=r.width,
-                height=r.height,
-                created_at=r.created_at,
-                thumbnail_url=thumbnail_url(r.sha256, r.mime),
-                artist=artists.get(r.id),
+                id=record.id,
+                path=record.path,
+                sha256=record.sha256,
+                mime=record.mime,
+                width=record.width,
+                height=record.height,
+                created_at=record.created_at,
+                effective_date=eff_date,
+                posted_at=posted_at,
+                thumbnail_url=thumbnail_url(record.sha256, record.mime),
+                artist=artists.get(record.id),
            )
-            for r in rows
+            for record, posted_at, eff_date in rows
        ]
        return GalleryPage(
            images=images,
@@ -179,11 +215,13 @@ class GalleryService:
        post_id: int | None = None,
        artist_id: int | None = None,
    ) -> list[TimelineBucket]:
-        year_col = func.date_part("year", ImageRecord.created_at).label("yr")
-        month_col = func.date_part("month", ImageRecord.created_at).label("mo")
+        eff = _effective_date_col()
+        year_col = func.date_part("year", eff).label("yr")
+        month_col = func.date_part("month", eff).label("mo")
        stmt = select(
            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)
        if tag_id is not None:
            stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
@@ -201,14 +239,17 @@ class GalleryService:
        post_id: int | None = None, artist_id: int | None = None,
    ) -> str | None:
        """Returns a cursor that, when passed to scroll(), positions at the
-        first image of the given year-month. None if the bucket is empty.
+        first image of the given year-month (by effective_date, not
+        created_at). None if the bucket is empty.
        """
        from sqlalchemy import extract

-        stmt = select(ImageRecord).where(
-            extract("year", ImageRecord.created_at) == year,
-            extract("month", ImageRecord.created_at) == month,
+        eff = _effective_date_col()
+        stmt = select(ImageRecord, eff.label("eff")).where(
+            extract("year", eff) == year,
+            extract("month", eff) == month,
        )
+        stmt = _outer_join_primary_post(stmt)
        _require_single_filter(tag_id, post_id, artist_id)
        if tag_id is not None:
            stmt = stmt.join(image_tag, image_tag.c.image_record_id == ImageRecord.id).where(
@@ -217,13 +258,14 @@ class GalleryService:
        prov = _provenance_clause(post_id, artist_id)
        if prov is not None:
            stmt = stmt.where(prov)
-        stmt = stmt.order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc()).limit(1)
-        first = (await self.session.execute(stmt)).scalar_one_or_none()
+        stmt = stmt.order_by(eff.desc(), ImageRecord.id.desc()).limit(1)
+        first = (await self.session.execute(stmt)).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
        # is the first result in the next scroll().
-        return encode_cursor(first.created_at, first.id + 1)
+        return encode_cursor(eff_date, record.id + 1)

    async def get_image_with_tags(self, image_id: int) -> dict | None:
        record = await self.session.get(ImageRecord, image_id)
@@ -236,6 +278,13 @@ class GalleryService:
            .order_by(Tag.kind.asc(), Tag.name.asc())
        )
        tags = (await self.session.execute(tag_stmt)).scalars().all()
+        # Fetch the canonical post.post_date for this image (if any) so
+        # the modal can show "Posted on <date>" alongside import date.
+        posted_at = None
+        if record.primary_post_id is not None:
+            posted_at = (await self.session.execute(
+                select(Post.post_date).where(Post.id == record.primary_post_id)
+            )).scalar_one_or_none()
        neighbors = await self._neighbors(record)
        # Direct artist FK — used by the modal's ProvenancePanel as a
        # fallback when ImageProvenance is empty (i.e., filesystem-
@@ -256,6 +305,7 @@ class GalleryService:
            "size_bytes": record.size_bytes,
            "integrity_status": record.integrity_status,
            "created_at": record.created_at.isoformat(),
+            "posted_at": posted_at.isoformat() if posted_at else None,
            "thumbnail_url": thumbnail_url(record.sha256, record.mime),
            "image_url": f"/images/{record.path.split('/images/', 1)[-1]}",
            "artist": (
@@ -275,34 +325,41 @@ class GalleryService:
        }

    async def _neighbors(self, record: ImageRecord) -> dict:
-        prev_stmt = (
-            select(ImageRecord.id)
-            .where(
+        # Compute the boundary image's effective_date in Python (one query
+        # below + the SELECT we already have on `record`) and use it for
+        # the neighbor comparison. Cheaper than re-deriving in SQL via
+        # correlated subquery.
+        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(
+            select(ImageRecord.id).where(
                or_(
-                    ImageRecord.created_at > record.created_at,
+                    eff > boundary_eff,
                    and_(
-                        ImageRecord.created_at == record.created_at,
+                        eff == boundary_eff,
                        ImageRecord.id > record.id,
                    ),
                )
            )
-            .order_by(ImageRecord.created_at.asc(), ImageRecord.id.asc())
-            .limit(1)
-        )
-        next_stmt = (
-            select(ImageRecord.id)
-            .where(
+        ).order_by(eff.asc(), ImageRecord.id.asc()).limit(1)
+        next_stmt = _outer_join_primary_post(
+            select(ImageRecord.id).where(
                or_(
-                    ImageRecord.created_at < record.created_at,
+                    eff < boundary_eff,
                    and_(
-                        ImageRecord.created_at == record.created_at,
+                        eff == boundary_eff,
                        ImageRecord.id < record.id,
                    ),
                )
            )
-            .order_by(ImageRecord.created_at.desc(), ImageRecord.id.desc())
-            .limit(1)
-        )
+        ).order_by(eff.desc(), ImageRecord.id.desc()).limit(1)
        prev_id = (await self.session.execute(prev_stmt)).scalar_one_or_none()
        next_id = (await self.session.execute(next_stmt)).scalar_one_or_none()
        return {"prev_id": prev_id, "next_id": next_id}
@@ -311,9 +368,11 @@ class GalleryService:
 def _group_by_year_month(
    images: list[GalleryImage],
 ) -> list[tuple[int, int, list[int]]]:
+    """Group by effective_date's year/month so migrated content surfaces
+    in the publish-date buckets, not the FC-scan-date bucket."""
    groups: list[tuple[int, int, list[int]]] = []
    for img in images:
-        y, m = img.created_at.year, img.created_at.month
+        y, m = img.effective_date.year, img.effective_date.month
        if groups and groups[-1][0] == y and groups[-1][1] == m:
            groups[-1][2].append(img.id)
        else:
@@ -280,7 +280,18 @@ class Importer:
                )

            if self.settings.skip_transparent and has_alpha:
-                pct = self._transparency_pct(source)
+                try:
+                    pct = self._transparency_pct(source)
+                except OSError as exc:
+                    # PIL.verify() at line 263 only validates header structure;
+                    # truncated/corrupt pixel data only surfaces when load()
+                    # actually decodes (here via getchannel('A')). Convert to
+                    # invalid_image skip so the Celery autoretry loop doesn't
+                    # bounce the same broken file forever.
+                    return ImportResult(
+                        status="skipped", skip_reason=SkipReason.invalid_image,
+                        error=f"PIL load failed during transparency check: {exc}",
+                    )
                if pct >= self.settings.transparency_threshold:
                    return ImportResult(
                        status="skipped", skip_reason=SkipReason.too_transparent,
@@ -302,8 +313,16 @@ class Importer:
        # Perceptual near-dup (images only; videos keep phash NULL).
        phash = None
        if not is_video(source):
-            with Image.open(source) as im:
-                phash = compute_phash(im)
+            try:
+                with Image.open(source) as im:
+                    phash = compute_phash(im)
+            except OSError as exc:
+                # Same rationale as the transparency-check guard above:
+                # broken-pixel-data files pass verify() but blow up here.
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.invalid_image,
+                    error=f"PIL load failed during phash compute: {exc}",
+                )
            if phash is not None:
                cand_rows = self.session.execute(
                    select(
@@ -37,13 +37,25 @@ from ...utils.slug import slugify
 from .ir_ingest import manifest_path

 # Per-platform artist-profile URL — used as Source.url when restoring
-# IR PostMetadata into FC. Keep this table in sync with
-# backend/app/services/extension_service.py:_PLATFORM_PATTERNS and
-# extension/lib/platforms.js.
+# IR PostMetadata into FC. Must cover every platform that
+# backend/app/services/extension_service.py:_PLATFORM_PATTERNS
+# recognizes; an entry missing here silently drops ALL PostMetadata for
+# that platform during phase 4 (operator hit this 2026-05-25:
+# DeviantArt + Pixiv posts in the IR migration produced empty
+# ImageProvenance because they fell through this table).
+#
+# Pixiv caveat: the real profile URL takes a numeric user_id
+# (https://www.pixiv.net/users/12345), but IR's PostMetadata.artist
+# stores the display name not the id. We use the slugified name here
+# so we preserve the artist→post→image linkage; the resulting Source.url
+# won't resolve in a browser and the operator may want to manually fix
+# it via Settings → Subscriptions once the migration lands.
 _PLATFORM_PROFILE_URL = {
    "patreon":       "https://www.patreon.com/{slug}",
    "subscribestar": "https://www.subscribestar.com/{slug}",
    "hentaifoundry": "https://www.hentai-foundry.com/user/{slug}",
+    "deviantart":    "https://www.deviantart.com/{slug}",
+    "pixiv":         "https://www.pixiv.net/users/{slug}",
 }


@@ -110,7 +122,14 @@ async def _ensure_provenance(
    db: AsyncSession, *,
    image_id: int, post_id: int, source_id: int, dry_run: bool,
 ) -> bool:
-    """Returns True if a new ImageProvenance row was inserted."""
+    """Returns True if a new ImageProvenance row was inserted.
+
+    Also sets ImageRecord.primary_post_id to this post if the image
+    doesn't already have one — preserves any primary_post_id already
+    assigned at download time by the importer (don't clobber). This is
+    the linkage gallery_service.py uses to surface Post.post_date as
+    the image's effective date for sort/group/jump/neighbor nav.
+    """
    existing = (await db.execute(
        select(ImageProvenance.id).where(
            ImageProvenance.image_record_id == image_id,
@@ -118,6 +137,18 @@ async def _ensure_provenance(
            ImageProvenance.source_id == source_id,
        )
    )).scalar_one_or_none()
+
+    # Whether-or-not the provenance row already exists, ensure the
+    # image's primary_post_id is set so the gallery date-coalesce works.
+    # Idempotent: only writes when currently NULL.
+    if not dry_run:
+        await db.execute(
+            ImageRecord.__table__.update()
+            .where(ImageRecord.id == image_id)
+            .where(ImageRecord.primary_post_id.is_(None))
+            .values(primary_post_id=post_id)
+        )
+
    if existing is not None:
        return False
    if dry_run:
@@ -4,12 +4,14 @@ CPU-only, single-image at a time. Loaded lazily inside the ml-worker
 process; NOT thread-safe — the ml queue worker must run --concurrency=1
 (set by the FC-1 entrypoint).

-Camie's selected_tags.csv columns: tag_id,name,category,count
-where category is a string: general|character|copyright|artist|meta|rating|year
-(unlike WD14's integer Danbooru category ids).
+v2 layout reference: HuggingFace Camais03/camie-tagger-v2 root has
+camie-tagger-v2.onnx (789 MB) + camie-tagger-v2-metadata.json (7.77 MB)
+ config.json. Tags ship as nested JSON, not CSV. Preprocessing and
+output handling follow the published onnx_inference.py reference:
+ImageNet normalize, NCHW layout, sigmoid on refined logits (output[1]).
 """

-import csv
+import json
 import os
 from dataclasses import dataclass
 from pathlib import Path
@@ -28,6 +30,8 @@ ImageFile.LOAD_TRUNCATED_IMAGES = True

 MODEL_NAME = os.environ.get("CAMIE_MODEL_NAME", "camie-tagger-v2")
 _MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie"
+_MODEL_FILE = f"{MODEL_NAME}.onnx"
+_METADATA_FILE = f"{MODEL_NAME}-metadata.json"

 # Below this confidence, predictions aren't stored (keeps the JSON compact).
 STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05"))
@@ -39,6 +43,12 @@ STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05"))
 # stored at STORE_FLOOR but artist never surfaces.
 SURFACED_CATEGORIES = {"character", "copyright", "general"}

+# ImageNet preprocessing constants (per Camie v2 onnx_inference.py).
+_IMAGENET_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
+_IMAGENET_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
+# Square-pad color ≈ ImageNet mean × 255 (matches reference inference).
+_PAD_COLOR = (124, 116, 104)
+

@dataclass(frozen=True)
 class TagPrediction:
@@ -51,34 +61,48 @@ class Tagger:
    def __init__(self, model_dir: Path | None = None):
        self._model_dir = model_dir or _MODEL_DIR
        self._session = None  # onnxruntime.InferenceSession once load()ed
-        self._tag_meta: list[dict] | None = None
+        self._tag_names: list[str] | None = None
+        self._tag_categories: list[str] | None = None
        self._input_name: str | None = None
-        self._output_name: str | None = None
-        self._input_size: int = 448
+        self._input_size: int = 512

    def load(self) -> None:
        if self._session is not None:
            return
-        model_path = self._model_dir / "model.onnx"
-        tags_path = self._model_dir / "selected_tags.csv"
+        model_path = self._model_dir / _MODEL_FILE
+        meta_path = self._model_dir / _METADATA_FILE
        if not model_path.is_file():
            raise RuntimeError(
-                f"Camie model.onnx missing at {model_path}. "
+                f"Camie {_MODEL_FILE} missing at {model_path}. "
                f"Populate /models via the ml-worker downloader."
            )
-        if not tags_path.is_file():
+        if not meta_path.is_file():
            raise RuntimeError(
-                f"Camie selected_tags.csv missing at {tags_path}. "
+                f"Camie {_METADATA_FILE} missing at {meta_path}. "
                f"Populate /models via the ml-worker downloader."
            )

-        tag_meta: list[dict] = []
-        with open(tags_path, newline="") as f:
-            reader = csv.DictReader(f)
-            for row in reader:
-                tag_meta.append(
-                    {"name": row["name"], "category": row["category"]}
-                )
+        with open(meta_path) as f:
+            metadata = json.load(f)
+
+        # Per Camie v2 onnx_inference.py: idx_to_tag is keyed by str(idx);
+        # tag_to_category maps tag_name -> category. Project to two parallel
+        # lists indexed by output position for O(1) lookup in the hot path.
+        ds = metadata["dataset_info"]
+        idx_to_tag = ds["tag_mapping"]["idx_to_tag"]
+        tag_to_category = ds["tag_mapping"]["tag_to_category"]
+        total = ds["total_tags"]
+        names: list[str] = []
+        cats: list[str] = []
+        for i in range(total):
+            name = idx_to_tag.get(str(i), f"unknown-{i}")
+            names.append(name)
+            cats.append(tag_to_category.get(name, "general"))
+
+        # Input size from metadata; fall back to 512 (the v2 default).
+        self._input_size = int(
+            metadata.get("model_info", {}).get("img_size", 512)
+        )

        # Lazy import — kept after the file-existence checks so the
        # missing-model RuntimeError still fires first in environments
@@ -89,51 +113,65 @@ class Tagger:
            str(model_path), providers=["CPUExecutionProvider"]
        )
        self._input_name = session.get_inputs()[0].name
-        self._output_name = session.get_outputs()[0].name
-        input_shape = session.get_inputs()[0].shape
-        for dim in input_shape:
-            if isinstance(dim, int) and dim > 1:
-                self._input_size = dim
-                break
        # Assign sentinels last so a partial load isn't observable.
-        self._tag_meta = tag_meta
+        self._tag_names = names
+        self._tag_categories = cats
        self._session = session

    def _preprocess(self, image_path: Path) -> np.ndarray:
        img = Image.open(image_path)
-        # Camie handles RGBA natively but we still composite onto white so
-        # transparency doesn't bias the model (same as IR's WD14 path).
-        if img.mode != "RGBA":
-            img = img.convert("RGBA")
-        bg = Image.new("RGBA", img.size, (255, 255, 255, 255))
-        bg.paste(img, mask=img.split()[3])
-        img = bg.convert("RGB")
+        # Composite RGBA onto neutral so transparency doesn't bias the model.
+        if img.mode == "RGBA":
+            bg = Image.new("RGBA", img.size, (255, 255, 255, 255))
+            bg.paste(img, mask=img.split()[3])
+            img = bg.convert("RGB")
+        elif img.mode != "RGB":
+            img = img.convert("RGB")

+        # Pad to square with ImageNet-mean color, then bicubic resize.
        w, h = img.size
        side = max(w, h)
-        square = Image.new("RGB", (side, side), (255, 255, 255))
+        square = Image.new("RGB", (side, side), _PAD_COLOR)
        square.paste(img, ((side - w) // 2, (side - h) // 2))
        square = square.resize(
            (self._input_size, self._input_size), Image.BICUBIC
        )
-        arr = np.array(square, dtype=np.float32)
-        return arr[np.newaxis, :, :, :]  # NHWC
+
+        arr = np.array(square, dtype=np.float32) / 255.0   # HWC, [0,1]
+        arr = (arr - _IMAGENET_MEAN) / _IMAGENET_STD        # ImageNet normalize
+        arr = arr.transpose(2, 0, 1)                        # HWC -> CHW
+        return arr[np.newaxis, :, :, :]                     # NCHW

    def infer(self, image_path: Path) -> dict[str, TagPrediction]:
-        """Run Camie on one image. Returns {name: TagPrediction}, only
-        entries with confidence >= STORE_FLOOR (across all categories —
-        the suggestion service does category filtering later)."""
+        """Run Camie v2 on one image. Returns {name: TagPrediction} with
+        confidence >= STORE_FLOOR (across all categories — the suggestion
+        service does category filtering later).
+
+        v2 emits multiple outputs; we use the refined predictions
+        (output[1] per onnx_inference.py). Sigmoid is applied to raw
+        logits to produce [0,1] confidence scores.
+        """
        self.load()
        x = self._preprocess(image_path)
-        out = self._session.run([self._output_name], {self._input_name: x})[0][0]
+        outputs = self._session.run(None, {self._input_name: x})
+        # Refined predictions if present (v2 emits initial + refined),
+        # fall back to initial for single-output forks.
+        logits = outputs[1] if len(outputs) > 1 else outputs[0]
+        # Squeeze batch dim, apply sigmoid.
+        probs = 1.0 / (1.0 + np.exp(-logits[0]))
        results: dict[str, TagPrediction] = {}
-        for idx, score in enumerate(out):
+        names = self._tag_names
+        cats = self._tag_categories
+        for idx, score in enumerate(probs):
            conf = float(score)
            if conf < STORE_FLOOR:
                continue
-            meta = self._tag_meta[idx]
-            results[meta["name"]] = TagPrediction(
-                name=meta["name"], category=meta["category"], confidence=conf
+            if idx >= len(names):
+                # Output longer than metadata declared — shouldn't happen but
+                # don't crash the import pipeline if v2 metadata desynchronizes.
+                continue
+            results[names[idx]] = TagPrediction(
+                name=names[idx], category=cats[idx], confidence=conf
            )
        return results

@@ -16,6 +16,7 @@ from ._sync_engine import sync_session_factory as _sync_session_factory
 log = logging.getLogger(__name__)

 STUCK_THRESHOLD_MINUTES = 5
+ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
 VERIFY_PAGE = 200
@@ -26,40 +27,77 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days

@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
-    """Find ImportTask rows stuck in 'processing' for >5 min and re-queue them.
+    """Recover stuck ImportTask rows. Two distinct stuck states:

-    Why 5 min: import_media_file is sub-second for the vast majority of
-    files; even a large-video transcode caps at the per-task soft_time_limit
-    (5 min) defined on the task itself. Anything still 'processing' after
-    that window is a confirmed crash (worker died, DB disconnect mid-flush,
-    OOM) and must be recycled. Was 30 min historically; tightened
-    2026-05-24 after operator hit a 2224-row zombie pile during the IR
-    migration scan.
+    1. 'processing' > 5 min — worker crash mid-import. Re-queue via
+       .delay() and let the import retry. Was 30 min historically;
+       tightened 2026-05-24 after operator hit a 2224-row zombie pile.
+       import_media_file is sub-second for the vast majority of files and
+       capped at the per-task soft_time_limit (5 min), so anything still
+       'processing' after that window is a confirmed crash.
+
+    2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
+       creates rows with status='pending' (commit), then in a second pass
+       transitions to 'queued' and calls .delay() (commit). If the scanner
+       crashes between those two commits, rows are orphaned in 'pending'
+       (never enqueued) with no recovery path — invisible to the
+       'processing' sweep above. Flagged 2026-05-25 by operator hitting a
+       5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so
+       the operator drains them via /api/import/retry-failed at their own
+       pace; bulk-re-enqueueing 5000+ rows would thundering-herd the
+       import worker.
+
+    Returns total rows touched (recovered + marked failed).
    """
    SessionLocal = _sync_session_factory()
-    cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    now = datetime.now(UTC)
+    processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
    with SessionLocal() as session:
        stuck_ids = session.execute(
            select(ImportTask.id)
            .where(ImportTask.status == "processing")
-            .where(ImportTask.started_at < cutoff)
+            .where(ImportTask.started_at < processing_cutoff)
        ).scalars().all()

-        if not stuck_ids:
+        orphan_ids = session.execute(
+            select(ImportTask.id)
+            .where(ImportTask.status.in_(["pending", "queued"]))
+            .where(ImportTask.created_at < orphan_cutoff)
+        ).scalars().all()
+
+        if not stuck_ids and not orphan_ids:
            return 0

-        session.execute(
-            update(ImportTask)
-            .where(ImportTask.id.in_(stuck_ids))
-            .values(status="queued", started_at=None, error="recovered from stuck state")
-        )
+        if stuck_ids:
+            session.execute(
+                update(ImportTask)
+                .where(ImportTask.id.in_(stuck_ids))
+                .values(status="queued", started_at=None, error="recovered from stuck state")
+            )
+
+        if orphan_ids:
+            session.execute(
+                update(ImportTask)
+                .where(ImportTask.id.in_(orphan_ids))
+                .values(
+                    status="failed",
+                    error=(
+                        "orphan pending/queued swept by recover_interrupted_tasks "
+                        "(scanner likely crashed mid-enqueue); retry via "
+                        "/api/import/retry-failed"
+                    ),
+                )
+            )
+
        session.commit()

-        from .import_file import import_media_file
-        for tid in stuck_ids:
-            import_media_file.delay(tid)
+        if stuck_ids:
+            from .import_file import import_media_file
+            for tid in stuck_ids:
+                import_media_file.delay(tid)

-        return len(stuck_ids)
+        return len(stuck_ids) + len(orphan_ids)


@celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
@@ -38,9 +38,18 @@ function onCardClick() {
 .fc-artistcard { cursor: pointer; }
 .fc-artistcard__previews {
  display: grid; grid-template-columns: repeat(3, 1fr);
-  gap: 2px; aspect-ratio: 3 / 1; background: rgb(var(--v-theme-surface-light));
+  gap: 2px; aspect-ratio: 3 / 1;
+  /* Explicit floor + ceiling so tall source images can't escape the
+     preview slot even on browsers where aspect-ratio doesn't compute. */
+  min-height: 150px; max-height: 220px;
+  overflow: hidden;
+  background: rgb(var(--v-theme-surface-light));
+}
+.fc-artistcard__previews img {
+  display: block;
+  width: 100%; height: 100%;
+  object-fit: cover; object-position: center;
 }
-.fc-artistcard__previews img { width: 100%; height: 100%; object-fit: cover; }
 .fc-artistcard__noimg {
  grid-column: 1 / -1; display: flex; align-items: center;
  justify-content: center;
@@ -106,9 +106,19 @@ function submit() {
 .fc-tagcard { cursor: pointer; }
 .fc-tagcard__previews {
  display: grid; grid-template-columns: repeat(3, 1fr);
-  gap: 2px; aspect-ratio: 3 / 1; background: rgb(var(--v-theme-surface-light));
+  gap: 2px; aspect-ratio: 3 / 1;
+  /* Explicit floor + ceiling so tall source images can't escape the
+     preview slot even on browsers where aspect-ratio doesn't compute
+     (older Safari, embedded webviews). */
+  min-height: 150px; max-height: 220px;
+  overflow: hidden;
+  background: rgb(var(--v-theme-surface-light));
+}
+.fc-tagcard__previews img {
+  display: block;
+  width: 100%; height: 100%;
+  object-fit: cover; object-position: center;
 }
-.fc-tagcard__previews img { width: 100%; height: 100%; object-fit: cover; }
 .fc-tagcard__noimg {
  grid-column: 1 / -1; display: flex; align-items: center;
  justify-content: center;
@@ -17,6 +17,12 @@
      >
        Retry failed
      </v-btn>
+      <v-btn
+        variant="text" rounded="pill" size="small" color="warning"
+        :disabled="!hasStuck" @click="onClearStuckOpen"
+      >
+        Clear stuck…
+      </v-btn>
      <v-btn
        variant="text" rounded="pill" size="small" color="error"
        @click="onClearOpen"
@@ -69,6 +75,31 @@
        </v-card-actions>
      </v-card>
    </v-dialog>
+
+    <v-dialog v-model="clearStuckDialog" max-width="480">
+      <v-card>
+        <v-card-title>Clear stuck tasks</v-card-title>
+        <v-card-text>
+          <v-alert type="warning" variant="tonal" density="compact" class="mb-3">
+            Force every <strong>pending / queued / processing</strong> task to
+            <strong>failed</strong> and finalize any active batch that
+            has no remaining work. Use this when the automatic recovery
+            sweep keeps re-queueing the same row (e.g., corrupt file in
+            an autoretry loop, or worker model missing).
+          </v-alert>
+          <p class="text-body-2">
+            Tasks remain in the database with status=<code>failed</code>;
+            click <em>Retry failed</em> once the underlying cause is
+            resolved to re-queue them.
+          </p>
+        </v-card-text>
+        <v-card-actions>
+          <v-spacer />
+          <v-btn @click="clearStuckDialog = false">Cancel</v-btn>
+          <v-btn color="warning" rounded="pill" @click="onClearStuckConfirm">Clear stuck</v-btn>
+        </v-card-actions>
+      </v-card>
+    </v-dialog>
  </v-card>
 </template>

@@ -80,6 +111,7 @@ const store = useImportStore()
 const statusFilter = ref(null)
 const clearDialog = ref(false)
 const clearAgeDays = ref(7)
+const clearStuckDialog = ref(false)

 const statusOptions = [
  { title: 'All', value: null },
@@ -100,6 +132,9 @@ const headers = [
 ]

 const hasFailed = computed(() => store.tasks.some(t => t.status === 'failed'))
+const hasStuck = computed(() => store.tasks.some(
+  t => t.status === 'pending' || t.status === 'queued' || t.status === 'processing'
+))

 function statusColor(s) {
  return {
@@ -138,4 +173,9 @@ async function onClearConfirm() {
  await store.clearCompleted(clearAgeDays.value)
  clearDialog.value = false
 }
+function onClearStuckOpen() { clearStuckDialog.value = true }
+async function onClearStuckConfirm() {
+  await store.clearStuck()
+  clearStuckDialog.value = false
+}
 </script>
@@ -2,7 +2,7 @@
  <v-card>
    <v-card-title>Trigger scan</v-card-title>
    <v-card-text>
-      <div v-if="store.activeBatch" class="d-flex align-center" style="gap: 12px;">
+      <div v-if="store.activeBatch" class="d-flex align-center mb-3" style="gap: 12px;">
        <v-progress-circular
          indeterminate color="accent" size="20"
        />
@@ -13,20 +13,40 @@
          failed {{ store.activeBatch.failed }} /
          {{ store.activeBatch.total_files }} files
        </span>
+        <v-spacer />
+        <v-btn
+          variant="text" rounded="pill" size="small" color="warning"
+          :loading="clearing" @click="onClearStuck"
+        >
+          Clear stuck
+        </v-btn>
      </div>
-      <div v-else>
-        <p class="text-body-2 mb-3">
+
+      <p class="text-body-2 mb-3">
+        <span v-if="!store.activeBatch">
          Run a quick scan of the import directory. Deep scan (pHash dedup,
          archives) lands in FC-2d.
-        </p>
-        <v-btn color="primary" rounded="pill" @click="trigger" :loading="busy">
-          <v-icon start>mdi-magnify-scan</v-icon>
-          Quick scan
-        </v-btn>
-        <v-alert v-if="store.triggerError" type="error" variant="tonal" class="mt-3" closable>
-          {{ store.triggerError }}
-        </v-alert>
-      </div>
+        </span>
+        <span v-else>
+          An active batch is in progress. Wait for it to finish, or click
+          <em>Clear stuck</em> above if it has been wedged with no
+          measurable progress.
+        </span>
+      </p>
+
+      <v-btn
+        color="primary" rounded="pill"
+        :disabled="!!store.activeBatch"
+        :loading="busy"
+        @click="trigger"
+      >
+        <v-icon start>mdi-magnify-scan</v-icon>
+        Quick scan
+      </v-btn>
+
+      <v-alert v-if="store.triggerError" type="error" variant="tonal" class="mt-3" closable>
+        {{ store.triggerError }}
+      </v-alert>
    </v-card-text>
  </v-card>
 </template>
@@ -37,9 +57,21 @@ import { useImportStore } from '../../stores/import.js'

 const store = useImportStore()
 const busy = ref(false)
+const clearing = ref(false)

 async function trigger() {
  busy.value = true
  try { await store.triggerScan() } catch {} finally { busy.value = false }
 }
+
+async function onClearStuck() {
+  clearing.value = true
+  try {
+    await store.clearStuck()
+  } catch {
+    // store surfaces error via triggerError if needed
+  } finally {
+    clearing.value = false
+  }
+}
 </script>
@@ -92,6 +92,13 @@ export const useImportStore = defineStore('import', () => {
    await loadTasks(true)
  }

+  async function clearStuck() {
+    const body = await api.post('/api/import/clear-stuck')
+    await loadTasks(true)
+    await refreshStatus()
+    return body
+  }
+
  const hasMore = computed(() => tasksNextCursor.value !== null)

  return {
@@ -101,6 +108,6 @@ export const useImportStore = defineStore('import', () => {
    triggerError,
    loadSettings, patchSettings,
    refreshStatus, triggerScan,
-    loadTasks, setStatusFilter, retryFailed, clearCompleted
+    loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck
  }
 })
@@ -84,7 +84,7 @@ onUnmounted(() => observer && observer.disconnect())
 }
 .fc-artists__grid {
  display: grid;
-  grid-template-columns: repeat(auto-fill, minmax(220px, 1fr));
+  grid-template-columns: repeat(auto-fill, minmax(440px, 1fr));
  gap: 12px;
 }
 .fc-artists__sentinel {
@@ -236,7 +236,7 @@ async function onDeleteTagConfirm() {
 }
 .fc-tags__grid {
  display: grid;
-  grid-template-columns: repeat(auto-fill, minmax(220px, 1fr));
+  grid-template-columns: repeat(auto-fill, minmax(440px, 1fr));
  gap: 12px;
 }
 .fc-tags__sentinel {
@@ -86,6 +86,61 @@ async def test_clear_completed(client, db):
    assert body["deleted"] == 1


+@pytest.mark.asyncio
+async def test_clear_stuck_fails_non_terminal_and_finalizes_orphan_batch(client, db):
+    """Operator-flagged 2026-05-25: 3 large PNGs got stuck in 'processing'
+    for 2 days, the active ImportBatch never finalized, and the UI's
+    'Scanning...' banner persisted with 0/0 files. /api/import/clear-stuck
+    is the escape hatch to break the autoretry loop manually."""
+    from sqlalchemy import select as _select
+
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    # Three stuck rows in mixed non-terminal states.
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/p1", task_type="media", status="processing",
+    ))
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/p2", task_type="media", status="queued",
+    ))
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/p3", task_type="media", status="pending",
+    ))
+    # One already-complete row should be untouched.
+    db.add(ImportTask(
+        batch_id=batch.id, source_path="/done", task_type="media",
+        status="complete", finished_at=datetime.now(UTC),
+    ))
+    await db.commit()
+
+    resp = await client.post("/api/import/clear-stuck")
+    body = await resp.get_json()
+    assert resp.status_code == 200
+    assert body["tasks_failed"] == 3
+    assert body["batches_finalized"] == 1
+
+    statuses = {
+        row.status for row in
+        (await db.execute(_select(ImportTask).where(ImportTask.batch_id == batch.id)))
+        .scalars().all()
+    }
+    assert statuses == {"failed", "complete"}
+
+    batch_status = (await db.execute(
+        _select(ImportBatch.status).where(ImportBatch.id == batch.id)
+    )).scalar_one()
+    assert batch_status == "complete"
+
+
+@pytest.mark.asyncio
+async def test_clear_stuck_no_op_when_nothing_stuck(client, db):
+    resp = await client.post("/api/import/clear-stuck")
+    body = await resp.get_json()
+    assert resp.status_code == 200
+    assert body == {"tasks_failed": 0, "batches_finalized": 0}
+
+
@pytest.mark.asyncio
 async def test_trigger_accepts_deep(client, monkeypatch):
    # Stub the task dispatch — assert the API accepts 'deep' and forwards
@@ -9,11 +9,16 @@ from backend.app.scripts import download_models as dm


 def test_ensure_camie_skips_when_present(tmp_path, monkeypatch):
+    """v2 layout (HF Camais03/camie-tagger-v2): the ONNX file is named
+    camie-tagger-v2.onnx (not model.onnx) and tags ship inside
+    camie-tagger-v2-metadata.json (not selected_tags.csv). Both at root.
+    Updated 2026-05-25 after the actual repo layout was confirmed via
+    WebFetch — the old assertion pinned the v1 filenames."""
    monkeypatch.setattr(dm, "MODEL_ROOT", tmp_path)
    camie = tmp_path / "camie"
    camie.mkdir(parents=True)
-    (camie / "model.onnx").write_bytes(b"x")
-    (camie / "selected_tags.csv").write_text("tag_id,name,category,count\n")
+    (camie / "camie-tagger-v2.onnx").write_bytes(b"x")
+    (camie / "camie-tagger-v2-metadata.json").write_text("{}")
    with patch.object(dm, "_snapshot") as snap:
        dm.ensure_camie()
        snap.assert_not_called()
@@ -133,3 +133,135 @@ async def test_get_image_with_tags_includes_integrity_status(db):
    svc = GalleryService(db)
    payload = await svc.get_image_with_tags(img.id)
    assert payload["integrity_status"] == "ok"
+
+
+async def _seed_image_with_post(
+    db, *, sha: str, image_created_at, post_date, artist_name="test-artist",
+    platform="patreon", external_post_id="42",
+):
+    """Helper: seed an Artist + Source + Post and one ImageRecord whose
+    primary_post_id points at that Post. Used for date-coalesce tests."""
+    from backend.app.models import Artist, Post, Source
+    artist = Artist(name=artist_name, slug=artist_name.lower().replace(" ", "-"))
+    db.add(artist)
+    await db.flush()
+    source = Source(
+        artist_id=artist.id, platform=platform,
+        url=f"https://www.{platform}.com/{artist.slug}",
+    )
+    db.add(source)
+    await db.flush()
+    post = Post(
+        source_id=source.id, external_post_id=external_post_id,
+        post_title="A Post", post_date=post_date,
+    )
+    db.add(post)
+    await db.flush()
+    img = ImageRecord(
+        path=f"/images/test/{sha[:8]}.jpg",
+        sha256=sha, size_bytes=1000, mime="image/jpeg",
+        width=100, height=100,
+        origin="imported_filesystem", integrity_status="unknown",
+        primary_post_id=post.id,
+    )
+    img.created_at = image_created_at
+    db.add(img)
+    await db.flush()
+    return img, post
+
+
+@pytest.mark.asyncio
+async def test_scroll_sorts_by_post_date_when_available(db):
+    """Operator-flagged 2026-05-25: ~57k IR images all imported in the
+    same week sort by image.created_at and pile up in one month bucket.
+    Once primary_post_id is wired (via tag_apply phase 4), the gallery
+    should sort by Post.post_date instead, spreading them across the
+    actual publish years."""
+    base_import = _now()
+    # Image A: imported NOW, but post was made 2 years ago.
+    img_a, _ = await _seed_image_with_post(
+        db, sha="a" * 64,
+        image_created_at=base_import,
+        post_date=base_import - timedelta(days=730),
+        artist_name="Aria", external_post_id="A-1",
+    )
+    # Image B: imported NOW (1 min later), post made YESTERDAY.
+    img_b, _ = await _seed_image_with_post(
+        db, sha="b" * 64,
+        image_created_at=base_import - timedelta(minutes=1),
+        post_date=base_import - timedelta(days=1),
+        artist_name="Bea", external_post_id="B-1",
+    )
+    # Image C: filesystem-imported, no primary_post_id, created 5 days ago.
+    img_c = ImageRecord(
+        path="/images/test/c.jpg", sha256="c" * 64,
+        size_bytes=1000, mime="image/jpeg",
+        width=100, height=100,
+        origin="imported_filesystem", integrity_status="unknown",
+    )
+    img_c.created_at = base_import - timedelta(days=5)
+    db.add(img_c)
+    await db.flush()
+
+    svc = GalleryService(db)
+    page = await svc.scroll(cursor=None, limit=10)
+    # Effective-date order: B (yesterday) > C (5 days ago) > A (2 years ago)
+    assert [i.id for i in page.images] == [img_b.id, img_c.id, img_a.id]
+
+    # API exposes both fields explicitly so the UI can show "Posted X / Imported Y".
+    a_payload = next(i for i in page.images if i.id == img_a.id)
+    assert a_payload.posted_at is not None
+    assert a_payload.posted_at < a_payload.created_at
+    c_payload = next(i for i in page.images if i.id == img_c.id)
+    assert c_payload.posted_at is None
+    assert c_payload.effective_date == c_payload.created_at
+
+
+@pytest.mark.asyncio
+async def test_timeline_buckets_use_post_date_when_available(db):
+    """Timeline group-by must follow the same effective_date rule so the
+    UI's year/month navigation surfaces publish-date buckets, not the
+    single FC-scan bucket all migrated images share."""
+    base = datetime(2026, 6, 15, 12, 0, tzinfo=UTC)
+    await _seed_image_with_post(
+        db, sha="1" * 64,
+        image_created_at=base,
+        post_date=datetime(2024, 3, 10, tzinfo=UTC),
+        artist_name="Carl", external_post_id="C-1",
+    )
+    await _seed_image_with_post(
+        db, sha="2" * 64,
+        image_created_at=base,
+        post_date=datetime(2024, 3, 11, tzinfo=UTC),
+        artist_name="Dee", external_post_id="D-1",
+    )
+    await _seed_image_with_post(
+        db, sha="3" * 64,
+        image_created_at=base,
+        post_date=datetime(2025, 9, 1, tzinfo=UTC),
+        artist_name="Eli", external_post_id="E-1",
+    )
+    svc = GalleryService(db)
+    buckets = await svc.timeline()
+    bucket_keys = {(b.year, b.month, b.count) for b in buckets}
+    # Two posts in 2024-03, one in 2025-09 — even though all imported in 2026-06.
+    assert (2024, 3, 2) in bucket_keys
+    assert (2025, 9, 1) in bucket_keys
+    # The FC-import bucket should NOT appear since all 3 images have post_date.
+    assert not any(b.year == 2026 and b.month == 6 for b in buckets)
+
+
+@pytest.mark.asyncio
+async def test_get_image_with_tags_includes_posted_at_when_present(db):
+    base = _now()
+    img, _ = await _seed_image_with_post(
+        db, sha="f" * 64,
+        image_created_at=base,
+        post_date=base - timedelta(days=365),
+        artist_name="Fred", external_post_id="F-1",
+    )
+    svc = GalleryService(db)
+    payload = await svc.get_image_with_tags(img.id)
+    assert payload["posted_at"] is not None
+    # Image's own created_at is still surfaced separately.
+    assert payload["created_at"] != payload["posted_at"]
@@ -134,3 +134,58 @@ def test_root_level_file_has_no_artist(importer, import_layout):
    importer.import_one(src)
    artists = importer.session.execute(select(Artist)).scalars().all()
    assert artists == []
+
+
+def test_pil_load_oserror_in_transparency_check_skips_not_raises(
+    importer, import_layout, monkeypatch,
+):
+    """PIL.verify() only validates header structure — broken pixel data
+    only surfaces when load() actually decodes. The importer must catch
+    the OSError and return a skipped: invalid_image result so the Celery
+    autoretry loop doesn't bounce the same broken file forever.
+    Operator hit this 2026-05-25 with a corrupt JPEG in the IR set."""
+    import_root, _ = import_layout
+    src = import_root / "Bob" / "corrupt.png"
+    # Make a real RGBA PNG so the has_alpha path engages.
+    _make_png_rgba(src, (100, 100), alpha=128)
+
+    importer.settings.skip_transparent = True
+    importer.settings.transparency_threshold = 0.5
+
+    # Force the next _transparency_pct call to raise as if PIL's load()
+    # blew up on truncated pixel data.
+    def _boom(_self, _src):
+        raise OSError("broken data stream when reading image file")
+    monkeypatch.setattr(
+        type(importer), "_transparency_pct", _boom,
+    )
+
+    result = importer.import_one(src)
+    assert result.status == "skipped"
+    assert result.skip_reason == SkipReason.invalid_image
+    assert "transparency check" in (result.error or "")
+
+
+def test_pil_load_oserror_in_phash_compute_skips_not_raises(
+    importer, import_layout, monkeypatch,
+):
+    """Same shape as the transparency-check guard, but for the phash
+    compute block — the OTHER place PIL.load() runs implicitly during
+    the dedup pipeline."""
+    import_root, _ = import_layout
+    src = import_root / "Carol" / "corrupt.jpg"
+    _make_jpeg(src)
+
+    # Disable transparency check so we reach the phash compute block.
+    importer.settings.skip_transparent = False
+
+    from backend.app.services import importer as importer_module
+
+    def _boom(_im):
+        raise OSError("broken data stream when reading image file")
+    monkeypatch.setattr(importer_module, "compute_phash", _boom)
+
+    result = importer.import_one(src)
+    assert result.status == "skipped"
+    assert result.skip_reason == SkipReason.invalid_image
+    assert "phash compute" in (result.error or "")
@@ -68,6 +68,102 @@ def test_recover_interrupted_only_old(db_sync, monkeypatch):
    assert dispatched == [stale.id]


+def test_recover_interrupted_sweeps_pending_orphans_to_failed(db_sync, monkeypatch):
+    """A scan that creates ImportTask rows but crashes before the second
+    pass (transition to 'queued' + .delay()) leaves rows orphaned at
+    status='pending'. The sweep flips them to 'failed' so the operator
+    can drain via /api/import/retry-failed without thundering-herding.
+    Banked 2026-05-25 after operator hit 5490 stuck pending rows.
+    """
+    from backend.app.tasks import import_file
+    monkeypatch.setattr(import_file.import_media_file, "delay", lambda *_: None)
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    fresh_pending = ImportTask(
+        batch_id=batch_id, source_path="/import/fresh.jpg", task_type="media",
+        status="pending",
+    )
+    db_sync.add(fresh_pending)
+    db_sync.flush()
+    # created_at defaults to now() server-side; fresh row stays untouched.
+
+    # Two stale rows simulating the orphan pile: one 'pending', one
+    # 'queued' (scanner crashed AFTER transitioning some rows but
+    # before all). Both should sweep.
+    stale_pending = ImportTask(
+        batch_id=batch_id, source_path="/import/stale1.jpg", task_type="media",
+        status="pending",
+    )
+    stale_queued = ImportTask(
+        batch_id=batch_id, source_path="/import/stale2.jpg", task_type="media",
+        status="queued",
+    )
+    db_sync.add_all([stale_pending, stale_queued])
+    db_sync.flush()
+    # Backdate created_at past the orphan cutoff (30 min).
+    from sqlalchemy import update as _upd
+    db_sync.execute(
+        _upd(ImportTask)
+        .where(ImportTask.id.in_([stale_pending.id, stale_queued.id]))
+        .values(created_at=now - timedelta(hours=2))
+    )
+    db_sync.commit()
+
+    from backend.app.tasks.maintenance import recover_interrupted_tasks
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2
+
+    db_sync.refresh(fresh_pending)
+    db_sync.refresh(stale_pending)
+    db_sync.refresh(stale_queued)
+    assert fresh_pending.status == "pending"  # fresh row untouched
+    assert stale_pending.status == "failed"
+    assert stale_queued.status == "failed"
+    assert "orphan" in (stale_pending.error or "")
+
+
+def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch):
+    """One sweep tick handles both 'processing' crashes AND
+    'pending'/'queued' orphans in a single pass."""
+    from backend.app.tasks import import_file
+    dispatched: list[int] = []
+    monkeypatch.setattr(
+        import_file.import_media_file, "delay", dispatched.append
+    )
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    stuck = ImportTask(
+        batch_id=batch_id, source_path="/import/stuck.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+    )
+    orphan = ImportTask(
+        batch_id=batch_id, source_path="/import/orphan.jpg", task_type="media",
+        status="pending",
+    )
+    db_sync.add_all([stuck, orphan])
+    db_sync.flush()
+    from sqlalchemy import update as _upd
+    db_sync.execute(
+        _upd(ImportTask).where(ImportTask.id == orphan.id)
+        .values(created_at=now - timedelta(hours=2))
+    )
+    db_sync.commit()
+
+    from backend.app.tasks.maintenance import recover_interrupted_tasks
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2
+
+    db_sync.refresh(stuck)
+    db_sync.refresh(orphan)
+    assert stuck.status == "queued"
+    assert orphan.status == "failed"
+    assert dispatched == [stuck.id]  # stuck rows re-enqueue; orphans don't
+
+
 def test_cleanup_old_deletes_finished_old(db_sync):
    batch_id = _make_batch(db_sync)
    now = datetime.now(UTC)
@@ -40,5 +40,8 @@ def test_get_tagger_singleton():

 def test_load_raises_when_model_missing(tmp_path):
    t = Tagger(model_dir=tmp_path / "nonexistent")
-    with pytest.raises(RuntimeError, match="model.onnx missing"):
+    # Match the trailing "missing at <path>" rather than the specific
+    # filename, so a future model-version bump (camie-tagger-v3.onnx, etc.)
+    # doesn't bounce this test.
+    with pytest.raises(RuntimeError, match=r"\.onnx missing at "):
        t.load()
@@ -227,6 +227,67 @@ async def test_image_posts_creates_source_post_provenance(db, tmp_path):
    )).scalar_one()
    assert prov_count == 1

+    # Phase 4 must also set ImageRecord.primary_post_id so the gallery's
+    # effective_date COALESCE can surface Post.post_date. Operator-flagged
+    # 2026-05-25: without this, IR-migrated images keep sorting by FC's
+    # scan date instead of the original publish date.
+    primary_post_id = (await db.execute(
+        select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id)
+    )).scalar_one()
+    canonical_post_id = (await db.execute(
+        select(Post.id).where(Post.external_post_id == "10001")
+    )).scalar_one()
+    assert primary_post_id == canonical_post_id
+
+
+@pytest.mark.asyncio
+async def test_image_posts_primary_post_id_not_clobbered(db, tmp_path):
+    """If the importer already set primary_post_id (e.g. a downloaded
+    image with a known provenance), phase 4 must NOT overwrite it when
+    re-running tag_apply against the IR migration. The existing
+    download-time linkage is the source of truth."""
+    sha = "9" * 64
+    await _seed_image(db, sha, suffix="9")
+    # Pre-set primary_post_id to a sentinel Post so we can detect a clobber.
+    img_id = (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.sha256 == sha)
+    )).scalar_one()
+    # Build an existing Source + Post for the sentinel.
+    art = Artist(name="Pre-existing", slug="pre-existing")
+    db.add(art)
+    await db.flush()
+    src = Source(
+        artist_id=art.id, platform="patreon",
+        url="https://www.patreon.com/pre-existing",
+    )
+    db.add(src)
+    await db.flush()
+    sentinel_post = Post(
+        source_id=src.id, external_post_id="sentinel-99",
+        post_title="Pre-existing",
+    )
+    db.add(sentinel_post)
+    await db.flush()
+    await db.execute(
+        ImageRecord.__table__.update()
+        .where(ImageRecord.id == img_id)
+        .values(primary_post_id=sentinel_post.id)
+    )
+    await db.commit()
+
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "image_sha256s": [sha]},
+    ])
+    await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+
+    # The migration created a NEW Post (external_post_id="10001") and a
+    # new ImageProvenance, but primary_post_id must still point at the
+    # original sentinel.
+    primary_post_id = (await db.execute(
+        select(ImageRecord.primary_post_id).where(ImageRecord.id == img_id)
+    )).scalar_one()
+    assert primary_post_id == sentinel_post.id
+

@pytest.mark.asyncio
 async def test_image_posts_idempotent_on_rerun(db, tmp_path):
@@ -281,6 +342,42 @@ async def test_image_posts_unknown_platform_skipped(db, tmp_path):
    assert src_count == 0


+@pytest.mark.parametrize("platform,expected_url", [
+    ("deviantart", "https://www.deviantart.com/maewix"),
+    ("pixiv",      "https://www.pixiv.net/users/maewix"),
+])
+@pytest.mark.asyncio
+async def test_image_posts_extended_platforms_create_source(
+    db, tmp_path, platform, expected_url,
+):
+    """Regression for 2026-05-25 operator-reported bug: phase 4's
+    _PLATFORM_PROFILE_URL had only patreon/subscribestar/hentaifoundry,
+    silently dropping deviantart + pixiv PostMetadata from the IR migration."""
+    sha = f"{platform[0]}" * 64
+    await _seed_image(db, sha, suffix=f"_{platform}")
+    await db.commit()
+
+    _write_manifest(tmp_path, image_posts=[
+        {**_POST_ENTRY, "platform": platform, "image_sha256s": [sha]},
+    ])
+
+    result = await tag_apply.apply_async(db, images_root=tmp_path, dry_run=False)
+    assert result["counts"]["rows_inserted"] >= 1
+
+    src_url = (await db.execute(
+        select(Source.url).where(Source.platform == platform)
+    )).scalar_one()
+    assert src_url == expected_url
+
+    # ImageProvenance row was created.
+    prov_count = (await db.execute(
+        select(func.count(ImageProvenance.id))
+        .join(Source, Source.id == ImageProvenance.source_id)
+        .where(Source.platform == platform)
+    )).scalar_one()
+    assert prov_count == 1
+
+
@pytest.mark.asyncio
 async def test_image_posts_dry_run_makes_no_writes(db, tmp_path):
    sha = "2" * 64