Merge pull request 'v26.05.28.0: downloads dashboard + task-resilience overhaul (timeouts, archive split, 3-layer poison-pill defense)' (#31) from dev into main

Layer 2 — remediate a corrupt file by re-fetching a fresh copy from its
source, bounded to a single attempt. Operator-requested 2026-05-28.

New backend/app/services/refetch_service.py:
- resolve_refetch_source: parse the failed file's sidecar → platform,
  derive the artist from the import path, find an ENABLED Source with a
  real feed URL for (artist, platform). Returns None for filesystem-only
  imports, missing sidecars, or `sidecar:<platform>:<slug>` synthetic
  anchors (not pollable).
- attempt_refetch: if not already refetched AND a Source resolves,
  delete the corrupt file (so gallery-dl's skip_existing re-fetches it),
  set ImportTask.refetched=True, and trigger ONE download_source
  re-check. Bounded by `refetched` so source-side corruption can't loop.

Wiring:
- Manual endpoint POST /api/import/tasks/<id>/refetch (only on 'failed'
  tasks). Returns refetch_queued / no_source / already_refetched /
  not_found / not_failed.
- Auto path in recover_interrupted_tasks: for each poison-pill row, if
  env FC_AUTO_REFETCH_CORRUPT=1, attempt_refetch (default OFF — the
  manual button is the primary path; auto is opt-in since re-fetch
  deletes a file + re-runs the downloader).
- Frontend: a cloud-refresh icon button on failed rows in ImportTaskList
  → stores.import.refetchTask → toast keyed on the result status.

Filesystem imports with no upstream return no_source — the operator's
only remediation there is replacing the file on disk, surfaced clearly
in the toast.

Tests: 404 unknown task, 400 non-failed task, no_source when
unresolvable, and the full resolvable-source path (file deleted,
refetched flag set, one download_source dispatched, second call is a
no-op). The resolvable test repoints the migration-seeded
import_settings(id=1) scan path rather than inserting a conflicting row.

alembic/versions/0026_import_task_recovery_count_refetched.py

@@ -0,0 +1,53 @@
+"""import_task.recovery_count + refetched — poison-pill circuit breaker
+
+Revision ID: 0026
+Revises: 0025
+Create Date: 2026-05-28
+
+Backs the import-task resilience work (operator-flagged 2026-05-28):
+
+- recovery_count: how many times recover_interrupted_tasks has
+  re-queued this row from a stuck 'processing' state. A row that
+  hard-crashes the worker (OOM / segfault on a corrupt or oversized
+  input) leaves no terminal flip, so the sweep re-queues it — and
+  without a cap it would loop forever, re-crashing the worker each
+  time. After MAX_RECOVERY_ATTEMPTS the sweep marks it 'failed' with a
+  diagnostic instead.
+
+- refetched: whether a one-shot re-download has already been attempted
+  for this task's file. Bounds the Layer-2 re-fetch remediation to a
+  single attempt so source-side corruption doesn't loop.
+
+Both default to 0 / false; additive, no backfill needed.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0026"
+down_revision: Union[str, None] = "0025"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "import_task",
+        sa.Column(
+            "recovery_count", sa.Integer(), nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "import_task",
+        sa.Column(
+            "refetched", sa.Boolean(), nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("import_task", "refetched")
+    op.drop_column("import_task", "recovery_count")

backend/app/api/admin.py

@@ -97,11 +97,19 @@ async def images_bulk_delete():
             )
         )
 
-    if dry_run:
-        return jsonify(projected)
-
     sha8 = _bulk_image_confirm_token(image_ids)
     expected = f"delete-images-{sha8}"
+
+    if dry_run:
+        # Hand the canonical Tier-C confirm token back with the
+        # projection so the frontend doesn't have to recompute SHA-256
+        # client-side via crypto.subtle (Secure-Context-gated,
+        # undefined on plain-HTTP origins per the homelab posture).
+        # Operator-flagged 2026-05-27.
+        projected["confirm_token"] = expected
+        return jsonify(projected)
+
+
     if supplied_confirm != expected:
         return _bad(
             "confirm_mismatch",

backend/app/api/cleanup.py

@@ -81,6 +81,13 @@ async def min_dim_preview():
                 s, min_width=min_w, min_height=min_h,
             )
         )
+    # Hand the canonical Tier-C delete token back with the preview so
+    # the frontend doesn't have to recompute SHA-256 client-side.
+    # window.crypto.subtle is Secure-Context-gated and undefined on
+    # plain-HTTP origins (homelab posture); without this the Delete
+    # button silently swallowed the TypeError and never opened the
+    # confirm modal. Operator-flagged 2026-05-27.
+    projection["confirm_token"] = _min_dim_token(min_w, min_h)
     return jsonify(projection)
 
 

backend/app/api/import_admin.py

@@ -114,18 +114,55 @@ async def retry_failed():
                 status="queued", error=None,
                 started_at=None, finished_at=None,
             )
-            .returning(ImportTask.id)
+            .returning(ImportTask.id, ImportTask.task_type)
         )
-        failed_ids = [row[0] for row in result.all()]
-        if not failed_ids:
+        failed = result.all()
+        if not failed:
             return jsonify({"retried": 0})
         await session.commit()
 
-    from ..tasks.import_file import import_media_file
-    for tid in failed_ids:
-        import_media_file.delay(tid)
+    from ..tasks.import_file import enqueue_import
+    for tid, task_type in failed:
+        enqueue_import(tid, task_type)
 
-    return jsonify({"retried": len(failed_ids)})
+    return jsonify({"retried": len(failed)})
+
+
+@import_admin_bp.route("/tasks/<int:task_id>/refetch", methods=["POST"])
+async def refetch_task(task_id: int):
+    """Layer-2 one-shot re-download: delete the (corrupt) file behind a
+    failed import task and re-run its source's downloader to fetch a
+    fresh copy. Only works for files that resolve to an enabled,
+    real-URL subscription Source; filesystem-only imports return
+    no_source.
+
+    Returns one of: refetch_queued (+source_id) / no_source /
+    already_refetched / not_found / not_failed.
+    """
+    async with get_session() as session:
+        result = await session.run_sync(_refetch_task_sync, task_id)
+    if result["status"] == "not_found":
+        return jsonify(result), 404
+    if result["status"] == "not_failed":
+        return jsonify(result), 400
+    return jsonify(result)
+
+
+def _refetch_task_sync(session, task_id: int) -> dict:
+    from pathlib import Path
+
+    from ..models import ImportSettings
+    from ..services.refetch_service import attempt_refetch
+
+    task = session.get(ImportTask, task_id)
+    if task is None:
+        return {"status": "not_found"}
+    if task.status != "failed":
+        return {"status": "not_failed"}
+    settings = session.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    return attempt_refetch(session, task, Path(settings.import_scan_path))
 
 
 @import_admin_bp.route("/clear-stuck", methods=["POST"])

backend/app/models/import_task.py

@@ -8,7 +8,16 @@ been processing longer than the stuck-task threshold.
 
 from datetime import datetime
 
-from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base
@@ -26,6 +35,13 @@ class ImportTask(Base):
     task_type: Mapped[str] = mapped_column(String(16), nullable=False)  # media|archive
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
 
+    # Poison-pill circuit breaker (alembic 0026). recovery_count tracks
+    # how many times the stuck-task sweep has re-queued this row; after
+    # the cap it's failed with a diagnostic instead of looping. refetched
+    # bounds the one-shot re-download remediation to a single attempt.
+    recovery_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    refetched: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
     result_image_id: Mapped[int | None] = mapped_column(
         ForeignKey("image_record.id", ondelete="SET NULL"), nullable=True
     )

backend/app/services/credential_service.py

@@ -148,6 +148,7 @@ class CredentialService:
             return None
         plaintext = self.crypto.decrypt(row.encrypted_blob)
         netscape = _to_netscape(plaintext)
+        netscape = _augment_cookies(platform, netscape)
         self.cookies_dir.mkdir(parents=True, exist_ok=True)
         out = self.cookies_dir / f"{platform}_cookies.txt"
         out.write_text(netscape)
@@ -163,6 +164,19 @@ class CredentialService:
         return self.crypto.decrypt(row.encrypted_blob)
 
 
+def _augment_cookies(platform: str, netscape: str) -> str:
+    """Delegate to the platform's `augment_cookies` hook if one is
+    registered (subscribestar, hentaifoundry, etc. — see
+    `services/platforms/<name>.py`). No-op when the platform doesn't
+    register a hook (Patreon, DeviantArt). Centralizing the
+    quirks-per-platform in the platforms package means adding a new
+    platform's cookie quirks doesn't require touching this file."""
+    info = PLATFORMS.get(platform)
+    if info is None or info.augment_cookies is None:
+        return netscape
+    return info.augment_cookies(netscape)
+
+
 def _to_netscape(plaintext: str) -> str:
     """Accept either Netscape-format text (the extension's output) or a
     JSON array of cookie dicts (a manual-paste edge case); produce

backend/app/services/importer.py

@@ -30,6 +30,7 @@ from ..models import (
     PostAttachment,
     Source,
 )
+from ..utils import safe_probe
 from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name
 from ..utils.phash import compute_phash, find_similar
 from ..utils.sidecar import find_sidecar, parse_sidecar
@@ -407,6 +408,29 @@ class Importer:
         return ImportResult(status="attached")
 
     def _import_archive(self, source: Path) -> ImportResult:
+        # Layer-3 isolation: bomb-size guard + integrity test in a
+        # spawned child BEFORE extracting in this process. A
+        # decompression bomb or a native-lib crash on a malformed
+        # archive is contained to the child; we reject the file cleanly
+        # instead of OOMing/segfaulting the import worker. extract_archive
+        # is already fail-soft for plain exceptions, so this only adds
+        # the hard-crash protection.
+        probe = safe_probe.probe_archive(source)
+        if not probe.ok:
+            if probe.crashed:
+                return ImportResult(
+                    status="failed",
+                    error=f"archive probe crashed/timed out: {probe.reason}",
+                )
+            # Clean rejection (bomb cap exceeded, integrity mismatch):
+            # still preserve the archive file itself as an attachment so
+            # nothing silently vanishes, matching extract_archive's
+            # fail-soft contract.
+            artist = self._resolve_artist(source)
+            post = self._post_for_sidecar(source, artist)
+            self._capture_attachment(source, post=post, artist=artist, resolved=True)
+            return ImportResult(status="attached")
+
         artist = self._resolve_artist(source)
         post = self._post_for_sidecar(source, artist)
         member_ids: list[int] = []
@@ -446,7 +470,25 @@ class Importer:
         # Compute file dimensions (images only) and apply filters.
         width = height = None
         has_alpha = False
-        if not is_video(source):
+        if is_video(source):
+            # Layer-3 isolation: validate the container via ffprobe (a
+            # separate process) before the rest of the pipeline touches
+            # it. A corrupt video that would crash a decoder is rejected
+            # cleanly here, and we capture width/height for free (the
+            # importer didn't previously record video dimensions).
+            probe = safe_probe.probe_video(source)
+            if not probe.ok:
+                if probe.crashed:
+                    return ImportResult(
+                        status="failed",
+                        error=f"video probe crashed/timed out: {probe.reason}",
+                    )
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.invalid_image,
+                    error=probe.reason,
+                )
+            width, height = probe.width, probe.height
+        else:
             try:
                 with Image.open(source) as im:
                     im.verify()

backend/app/services/platforms.py

@@ -1,140 +0,0 @@
-"""FC-3b platforms registry — the single source of truth for what
-FabledCurator supports.
-
-Lifted from GallerySubscriber's
-~/Nextcloud/Projects/GallerySubscriber/backend/app/api/platforms.py
-and ~/.../extension/lib/platforms.js. Six platforms; auth_type and
-URL patterns match GS exactly so the existing browser extension
-hits FC unmodified.
-"""
-
-from dataclasses import dataclass
-from typing import Literal
-
-
-@dataclass(frozen=True)
-class PlatformInfo:
-    key: str
-    name: str
-    description: str
-    auth_type: Literal["cookies", "token"]
-    requires_auth: bool
-    url_pattern: str
-    url_examples: list[str]
-    default_config: dict
-    notes: str | None = None
-
-
-# Common defaults used across most platforms; embedded per-platform
-# below so per-platform overrides remain explicit.
-_DEFAULTS = {
-    "sleep": 3.0,
-    "sleep_request": 1.5,
-    "skip_existing": True,
-    "save_metadata": True,
-    "timeout": 3600,
-}
-
-
-PLATFORMS: dict[str, PlatformInfo] = {
-    "patreon": PlatformInfo(
-        key="patreon",
-        name="Patreon",
-        description="Download posts from Patreon creators",
-        auth_type="cookies",
-        requires_auth=True,
-        url_pattern=r"^https?://(www\.)?patreon\.com/",
-        url_examples=[
-            "https://www.patreon.com/example_artist",
-            "https://www.patreon.com/user?u=12345678",
-        ],
-        default_config={**_DEFAULTS, "content_types": ["images", "attachments"]},
-    ),
-    "subscribestar": PlatformInfo(
-        key="subscribestar",
-        name="SubscribeStar",
-        description="Download posts from SubscribeStar creators",
-        auth_type="cookies",
-        requires_auth=True,
-        url_pattern=r"^https?://(www\.)?subscribestar\.(com|adult)/",
-        url_examples=[
-            "https://subscribestar.adult/example_artist",
-            "https://www.subscribestar.com/example_artist",
-        ],
-        default_config={**_DEFAULTS, "content_types": ["all"]},
-    ),
-    "hentaifoundry": PlatformInfo(
-        key="hentaifoundry",
-        name="Hentai Foundry",
-        description="Download artwork from Hentai Foundry artists",
-        auth_type="cookies",
-        requires_auth=False,
-        url_pattern=r"^https?://(www\.)?hentai-foundry\.com/",
-        url_examples=[
-            "https://www.hentai-foundry.com/user/example_artist",
-            "https://www.hentai-foundry.com/pictures/user/example_artist",
-        ],
-        default_config={**_DEFAULTS, "content_types": ["pictures"]},
-    ),
-    "discord": PlatformInfo(
-        key="discord",
-        name="Discord",
-        description="Download attachments from Discord channels",
-        auth_type="token",
-        requires_auth=True,
-        url_pattern=r"^https?://(www\.)?discord\.com/channels/",
-        url_examples=["https://discord.com/channels/123456789/987654321"],
-        default_config={**_DEFAULTS, "content_types": ["all"]},
-        notes="Requires Discord user token (not bot token).",
-    ),
-    "pixiv": PlatformInfo(
-        key="pixiv",
-        name="Pixiv",
-        description="Download artwork from Pixiv artists",
-        auth_type="token",
-        requires_auth=True,
-        url_pattern=r"^https?://(www\.)?pixiv\.net/",
-        url_examples=[
-            "https://www.pixiv.net/users/12345678",
-            "https://www.pixiv.net/en/users/12345678",
-        ],
-        default_config={**_DEFAULTS, "content_types": ["all"]},
-        notes="Requires OAuth refresh token. Run `gallery-dl oauth:pixiv` to obtain one.",
-    ),
-    "deviantart": PlatformInfo(
-        key="deviantart",
-        name="DeviantArt",
-        description="Download artwork from DeviantArt artists",
-        auth_type="cookies",
-        requires_auth=False,
-        url_pattern=r"^https?://(www\.)?deviantart\.com/",
-        url_examples=[
-            "https://www.deviantart.com/example-artist",
-            "https://www.deviantart.com/example-artist/gallery",
-        ],
-        default_config={**_DEFAULTS, "content_types": ["gallery"]},
-    ),
-}
-
-
-def known_platform_keys() -> frozenset[str]:
-    return frozenset(PLATFORMS.keys())
-
-
-def auth_type_for(platform: str) -> str | None:
-    info = PLATFORMS.get(platform)
-    return info.auth_type if info else None
-
-
-def to_dict(info: PlatformInfo) -> dict:
-    return {
-        "key": info.key,
-        "name": info.name,
-        "description": info.description,
-        "auth_type": info.auth_type,
-        "requires_auth": info.requires_auth,
-        "url_pattern": info.url_pattern,
-        "url_examples": info.url_examples,
-        "default_config": info.default_config,
-        "notes": info.notes,
-    }

backend/app/services/platforms/__init__.py

@@ -0,0 +1,95 @@
+"""FC-3b platforms registry — single source of truth for what
+FabledCurator supports + where each platform's quirks live.
+
+Adding a new platform: drop a new module `<platform>.py` next to this
+one, declare an `INFO = PlatformInfo(...)`, add the import + entry in
+PLATFORMS below. Sidecar parsing, cookie materialization, and
+`/api/platforms` pick it up automatically.
+
+Lifted from GallerySubscriber's
+~/Nextcloud/Projects/GallerySubscriber/backend/app/api/platforms.py
+and ~/.../extension/lib/platforms.js. Six platforms; auth_type and
+URL patterns match GS exactly so the existing browser extension
+hits FC unmodified.
+"""
+
+from .base import (
+    DEFAULT_DESCRIPTION_KEYS,
+    DEFAULT_EXTERNAL_POST_ID_KEYS,
+    PlatformInfo,
+)
+from .deviantart import INFO as _DEVIANTART
+from .discord import INFO as _DISCORD
+from .hentaifoundry import INFO as _HENTAIFOUNDRY
+from .patreon import INFO as _PATREON
+from .pixiv import INFO as _PIXIV
+from .subscribestar import INFO as _SUBSCRIBESTAR
+
+PLATFORMS: dict[str, PlatformInfo] = {
+    info.key: info
+    for info in (
+        _PATREON,
+        _SUBSCRIBESTAR,
+        _HENTAIFOUNDRY,
+        _DISCORD,
+        _PIXIV,
+        _DEVIANTART,
+    )
+}
+
+
+def known_platform_keys() -> frozenset[str]:
+    return frozenset(PLATFORMS.keys())
+
+
+def auth_type_for(platform: str) -> str | None:
+    info = PLATFORMS.get(platform)
+    return info.auth_type if info else None
+
+
+def to_dict(info: PlatformInfo) -> dict:
+    """Serialize a PlatformInfo to a JSON-safe dict for /api/platforms.
+
+    Behavioral fields (callables, sidecar-chain overrides) are
+    intentionally omitted — they aren't useful to API consumers.
+    """
+    return {
+        "key": info.key,
+        "name": info.name,
+        "description": info.description,
+        "auth_type": info.auth_type,
+        "requires_auth": info.requires_auth,
+        "url_pattern": info.url_pattern,
+        "url_examples": info.url_examples,
+        "default_config": info.default_config,
+        "notes": info.notes,
+    }
+
+
+def external_post_id_keys_for(platform: str | None) -> tuple[str, ...]:
+    """Resolve the external_post_id lookup chain for a given platform,
+    falling back to the module default when the platform isn't
+    registered or hasn't overridden the chain."""
+    info = PLATFORMS.get(platform) if platform else None
+    if info is not None and info.external_post_id_keys is not None:
+        return info.external_post_id_keys
+    return DEFAULT_EXTERNAL_POST_ID_KEYS
+
+
+def description_keys_for(platform: str | None) -> tuple[str, ...]:
+    """Resolve the description body lookup chain for a given platform."""
+    info = PLATFORMS.get(platform) if platform else None
+    if info is not None and info.description_keys is not None:
+        return info.description_keys
+    return DEFAULT_DESCRIPTION_KEYS
+
+
+__all__ = [
+    "PLATFORMS",
+    "PlatformInfo",
+    "auth_type_for",
+    "description_keys_for",
+    "external_post_id_keys_for",
+    "known_platform_keys",
+    "to_dict",
+]

backend/app/services/platforms/base.py

@@ -0,0 +1,105 @@
+"""PlatformInfo dataclass + shared defaults + small helpers.
+
+Per-platform modules import from here, register their PlatformInfo via
+INFO, optionally attaching `derive_post_url` and/or `augment_cookies`
+callables for behavior that diverges from gallery-dl's mainline shape
+(Patreon).
+
+Adding a new platform: drop a new module under `services/platforms/`,
+declare an INFO, and add it to the import list in
+`services/platforms/__init__.py`. Sidecar parsing, cookie
+materialization, and the /api/platforms response pick it up
+automatically.
+"""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Literal
+
+# Sidecar parsing defaults. Per-platform PlatformInfo entries can
+# override these by setting `external_post_id_keys=` /
+# `description_keys=`. Most don't need to — the defaults already cover
+# every platform FC supports.
+#
+# external_post_id chain: `post_id` MUST come before `id` because
+# SubscribeStar gallery-dl puts the per-attachment id in `id` and the
+# actual post id in `post_id`; picking `id` first fragments
+# multi-image SubscribeStar posts into N Post rows. Patreon/Pixiv have
+# no `post_id` so `id` still wins for them; HF uses `index`, Discord
+# uses `message_id` — all reached via the remaining chain entries.
+# (Banked 2026-05-27 during the sidecar audit.)
+DEFAULT_EXTERNAL_POST_ID_KEYS: tuple[str, ...] = (
+    "post_id", "id", "index", "message_id",
+)
+
+# Description body chain: Discord's gallery-dl extractor uses `message`
+# (no `content`); appended to the chain so Discord posts surface body
+# text.
+DEFAULT_DESCRIPTION_KEYS: tuple[str, ...] = (
+    "content", "description", "caption", "message",
+)
+
+
+@dataclass(frozen=True)
+class PlatformInfo:
+    # --- Identity / metadata ---
+    key: str
+    name: str
+    description: str
+    auth_type: Literal["cookies", "token"]
+    requires_auth: bool
+    url_pattern: str
+    url_examples: list[str]
+    default_config: dict
+    notes: str | None = None
+
+    # --- Sidecar parsing overrides ---
+    # Each is None to mean "use the module default above"; a platform
+    # only sets one of these when its sidecar shape genuinely differs.
+    external_post_id_keys: tuple[str, ...] | None = None
+    description_keys: tuple[str, ...] | None = None
+
+    # --- Behavioral hooks ---
+    # Synthesize a post permalink from sidecar data. Required when
+    # gallery-dl's `url` field is the file/CDN URL rather than the post
+    # permalink (subscribestar/pixiv/hf/discord). None = trust the bare
+    # `url` field (patreon, deviantart).
+    derive_post_url: Callable[[dict], str | None] | None = None
+
+    # Post-process the materialized cookies.txt for gallery-dl. Used by
+    # platforms whose server gates or extractor quirks need synthetic
+    # cookies the extension can't capture (subscribestar age cookie, HF
+    # host-only PHPSESSID duplicate). None = no-op.
+    augment_cookies: Callable[[str], str] | None = None
+
+
+def str_id_value(v) -> str | None:
+    """Coerce a JSON scalar id into a non-empty string, rejecting bool
+    (Python's bool is an int subclass so `isinstance(True, int)` is
+    True; without this guard a sidecar with `"id": true` would produce
+    external_post_id="True")."""
+    if isinstance(v, bool):
+        return None
+    if isinstance(v, (str, int)) and str(v).strip():
+        return str(v).strip()
+    return None
+
+
+def str_field(v) -> str | None:
+    """Same idea as str_id_value but for plain string fields (no int
+    coercion)."""
+    if isinstance(v, str) and v.strip():
+        return v.strip()
+    return None
+
+
+# Shared gallery-dl invocation defaults. Embedded in each platform's
+# default_config (with platform-specific overrides) so per-platform
+# choices stay explicit.
+GD_DEFAULTS = {
+    "sleep": 3.0,
+    "sleep_request": 1.5,
+    "skip_existing": True,
+    "save_metadata": True,
+    "timeout": 3600,
+}

backend/app/services/platforms/deviantart.py

@@ -0,0 +1,23 @@
+"""DeviantArt — no exercised quirks yet.
+
+No operator-owned DeviantArt archive existed at the 2026-05-27 sidecar
+audit, so we don't know yet whether DA's gallery-dl sidecars are
+well-behaved or have their own quirks. When DA gets exercised for the
+first time, add `derive_post_url` / `augment_cookies` here as needed.
+"""
+
+from .base import GD_DEFAULTS, PlatformInfo
+
+INFO = PlatformInfo(
+    key="deviantart",
+    name="DeviantArt",
+    description="Download artwork from DeviantArt artists",
+    auth_type="cookies",
+    requires_auth=False,
+    url_pattern=r"^https?://(www\.)?deviantart\.com/",
+    url_examples=[
+        "https://www.deviantart.com/example-artist",
+        "https://www.deviantart.com/example-artist/gallery",
+    ],
+    default_config={**GD_DEFAULTS, "content_types": ["gallery"]},
+)

backend/app/services/platforms/discord.py

@@ -0,0 +1,38 @@
+"""Discord — one quirk + one already-default.
+
+post_url: gallery-dl's `url` is the CDN attachment URL. The "permalink"
+for a Discord message uses the (server, channel, message) triple via
+`discord.com/channels/<server>/<channel>/<message>`. Note that
+permalinks are only resolvable for users in the same server — public
+access doesn't work — but the URL is still useful to the operator
+in-app.
+
+Description body is in `message` not `content`. That's already covered
+by the default description chain in base.py (DEFAULT_DESCRIPTION_KEYS
+ends with `message`). No description_keys override needed.
+"""
+
+from .base import GD_DEFAULTS, PlatformInfo, str_id_value
+
+
+def derive_post_url(data: dict) -> str | None:
+    sid = str_id_value(data.get("server_id"))
+    cid = str_id_value(data.get("channel_id"))
+    mid = str_id_value(data.get("message_id"))
+    if sid and cid and mid:
+        return f"https://discord.com/channels/{sid}/{cid}/{mid}"
+    return None
+
+
+INFO = PlatformInfo(
+    key="discord",
+    name="Discord",
+    description="Download attachments from Discord channels",
+    auth_type="token",
+    requires_auth=True,
+    url_pattern=r"^https?://(www\.)?discord\.com/channels/",
+    url_examples=["https://discord.com/channels/123456789/987654321"],
+    default_config={**GD_DEFAULTS, "content_types": ["all"]},
+    notes="Requires Discord user token (not bot token).",
+    derive_post_url=derive_post_url,
+)

backend/app/services/platforms/hentaifoundry.py

@@ -0,0 +1,83 @@
+"""HentaiFoundry — two quirks colocated.
+
+1. post_url: HF sidecars omit `url` entirely; `src` is the image URL.
+   Synthesize the permalink from `user` + `index`
+   (/pictures/user/<user>/<index>).
+
+2. augment_cookies: gallery-dl's HF extractor checks
+   `self.cookies.get("PHPSESSID", domain="www.hentai-foundry.com")` with
+   `requests`' EXACT domain matching. The extension's pre-v1.0.5
+   `cookies.js` aggressively rewrote every captured cookie to the
+   leading-dot subdomain-wide form (`.hentai-foundry.com`), which fails
+   the exact lookup even though the cookie IS sent on actual HTTP
+   requests (RFC 6265 subdomain matching). The extractor falls into
+   an unauthenticated `?enterAgree=1` HEAD that 401s. Inject host-only
+   duplicates of PHPSESSID + YII_CSRF_TOKEN so the lookup succeeds.
+"""
+
+from .base import GD_DEFAULTS, PlatformInfo, str_field, str_id_value
+
+_HOST_ONLY_NAMES = ("PHPSESSID", "YII_CSRF_TOKEN")
+
+
+def derive_post_url(data: dict) -> str | None:
+    user = str_field(data.get("user")) or str_field(data.get("artist"))
+    idx = str_id_value(data.get("index"))
+    if user and idx:
+        return f"https://www.hentai-foundry.com/pictures/user/{user}/{idx}"
+    return None
+
+
+def augment_cookies(netscape: str) -> str:
+    body = netscape.rstrip("\n")
+    if not body:
+        return netscape
+    lines = body.split("\n")
+    existing_host_only: set[str] = set()
+    by_name: dict[str, list[str]] = {}
+    for raw in lines:
+        if not raw or raw.startswith("#"):
+            continue
+        parts = raw.split("\t")
+        if len(parts) < 7:
+            continue
+        domain, _flag, _path, _secure, _exp, name, _value = parts[:7]
+        if name not in _HOST_ONLY_NAMES:
+            continue
+        if domain == "www.hentai-foundry.com":
+            existing_host_only.add(name)
+        elif domain in (".hentai-foundry.com", "hentai-foundry.com"):
+            by_name.setdefault(name, []).append(raw)
+
+    appended: list[str] = []
+    for name in _HOST_ONLY_NAMES:
+        if name in existing_host_only or name not in by_name:
+            continue
+        # Duplicate the first subdomain-wide line as host-only on
+        # www.hentai-foundry.com. Same value + expiry; flag=FALSE marks
+        # the entry host-only in netscape format.
+        parts = by_name[name][0].split("\t")
+        parts[0] = "www.hentai-foundry.com"
+        parts[1] = "FALSE"
+        appended.append("\t".join(parts[:7]))
+
+    if not appended:
+        return netscape
+    return body + "\n" + "\n".join(appended) + "\n"
+
+
+INFO = PlatformInfo(
+    key="hentaifoundry",
+    name="Hentai Foundry",
+    description="Download artwork from Hentai Foundry artists",
+    auth_type="cookies",
+    requires_auth=False,
+    url_pattern=r"^https?://(www\.)?hentai-foundry\.com/",
+    url_examples=[
+        "https://www.hentai-foundry.com/user/example_artist",
+        "https://www.hentai-foundry.com/pictures/user/example_artist",
+    ],
+    default_config={**GD_DEFAULTS, "content_types": ["pictures"]},
+    derive_post_url=derive_post_url,
+    augment_cookies=augment_cookies,
+)

backend/app/services/platforms/patreon.py

@@ -0,0 +1,23 @@
+"""Patreon — no quirks. The reference platform.
+
+Patreon's gallery-dl sidecars are the well-behaved baseline: `url` is a
+real permalink, `id` is the post id, `title` and `content` are
+populated. No cookie quirks (session cookies are domain-wide). No
+derivation overrides.
+"""
+
+from .base import GD_DEFAULTS, PlatformInfo
+
+INFO = PlatformInfo(
+    key="patreon",
+    name="Patreon",
+    description="Download posts from Patreon creators",
+    auth_type="cookies",
+    requires_auth=True,
+    url_pattern=r"^https?://(www\.)?patreon\.com/",
+    url_examples=[
+        "https://www.patreon.com/example_artist",
+        "https://www.patreon.com/user?u=12345678",
+    ],
+    default_config={**GD_DEFAULTS, "content_types": ["images", "attachments"]},
+)

backend/app/services/platforms/pixiv.py

@@ -0,0 +1,32 @@
+"""Pixiv — one quirk.
+
+post_url: gallery-dl's `url` is the image URL on `i.pximg.net`. The
+post permalink follows /artworks/<id>. external_post_id (= `id`) was
+already correct, so no override there.
+"""
+
+from .base import GD_DEFAULTS, PlatformInfo, str_id_value
+
+
+def derive_post_url(data: dict) -> str | None:
+    pid = str_id_value(data.get("id"))
+    if pid:
+        return f"https://www.pixiv.net/artworks/{pid}"
+    return None
+
+
+INFO = PlatformInfo(
+    key="pixiv",
+    name="Pixiv",
+    description="Download artwork from Pixiv artists",
+    auth_type="token",
+    requires_auth=True,
+    url_pattern=r"^https?://(www\.)?pixiv\.net/",
+    url_examples=[
+        "https://www.pixiv.net/users/12345678",
+        "https://www.pixiv.net/en/users/12345678",
+    ],
+    default_config={**GD_DEFAULTS, "content_types": ["all"]},
+    notes="Requires OAuth refresh token. Run `gallery-dl oauth:pixiv` to obtain one.",
+    derive_post_url=derive_post_url,
+)

backend/app/services/platforms/subscribestar.py

@@ -0,0 +1,62 @@
+"""SubscribeStar — three quirks colocated.
+
+1. external_post_id: gallery-dl puts the per-attachment id in `id`
+   (e.g. 711509) and the actual post id in `post_id` (e.g. 360360).
+   The default chain in base.py already prefers `post_id`; this module
+   doesn't need to override it but the comment lives here too so a
+   future reader knows the chain's order was driven by this platform.
+
+2. post_url: gallery-dl's `url` is the file CDN URL
+   (`/post_uploads?payload=...`). Synthesize the post permalink from
+   `post_id`.
+
+3. augment_cookies: the server gates artist pages behind a
+   `_personalization_id` age-confirmation cookie that the user can't
+   easily refresh — SubscribeStar's frontend JS uses localStorage to
+   suppress the age popup once dismissed. gallery-dl's own login flow
+   sidesteps this by setting `18_plus_agreement_generic=true` on
+   `.subscribestar.adult`; we mirror that for cookies captured via the
+   extension.
+"""
+
+from .base import GD_DEFAULTS, PlatformInfo, str_id_value
+
+
+def derive_post_url(data: dict) -> str | None:
+    pid = str_id_value(data.get("post_id"))
+    if pid:
+        return f"https://www.subscribestar.com/posts/{pid}"
+    return None
+
+
+def augment_cookies(netscape: str) -> str:
+    if "18_plus_agreement_generic" in netscape:
+        return netscape
+    # Far-future expiry — gallery-dl's own login flow sets this with no
+    # explicit expiry; the server only checks presence/value.
+    expiry = 4102444800  # 2100-01-01 UTC
+    line = "\t".join([
+        ".subscribestar.adult", "TRUE", "/", "TRUE",
+        str(expiry), "18_plus_agreement_generic", "true",
+    ])
+    body = netscape.rstrip("\n")
+    if not body:
+        body = "# Netscape HTTP Cookie File"
+    return body + "\n" + line + "\n"
+
+
+INFO = PlatformInfo(
+    key="subscribestar",
+    name="SubscribeStar",
+    description="Download posts from SubscribeStar creators",
+    auth_type="cookies",
+    requires_auth=True,
+    url_pattern=r"^https?://(www\.)?subscribestar\.(com|adult)/",
+    url_examples=[
+        "https://subscribestar.adult/example_artist",
+        "https://www.subscribestar.com/example_artist",
+    ],
+    default_config={**GD_DEFAULTS, "content_types": ["all"]},
+    derive_post_url=derive_post_url,
+    augment_cookies=augment_cookies,
+)

backend/app/services/refetch_service.py

@@ -0,0 +1,106 @@
+"""Layer-2 one-shot re-download remediation for corrupt imported files.
+
+When an import fails on a file that came from a known, pollable
+subscription Source, deleting the bad copy and re-running the source's
+downloader can fetch a fresh, unblemished copy. This only helps when:
+
+  - the corruption is in transit / on disk (not at the source), AND
+  - the file resolves to an ENABLED Source with a real feed URL
+    (a `sidecar:<platform>:<slug>` synthetic anchor is not pollable),
+    AND
+  - we haven't already re-fetched this task once (bounded by
+    ImportTask.refetched so source-side corruption can't loop).
+
+Filesystem-only imports with no resolvable Source return 'no_source' —
+the operator's only remediation there is to replace the file on disk.
+
+Operator-requested 2026-05-28 (Layer 2).
+"""
+
+import json
+import logging
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ..models import Artist, ImportTask, Source
+from ..utils.paths import derive_top_level_artist
+from ..utils.sidecar import find_sidecar, parse_sidecar
+from ..utils.slug import slugify
+
+log = logging.getLogger(__name__)
+
+
+def resolve_refetch_source(
+    session: Session, source_path: str, import_root: Path,
+) -> Source | None:
+    """Find an enabled, real-URL Source for the file's (artist, platform),
+    or None when nothing re-pollable resolves."""
+    path = Path(source_path)
+    sc = find_sidecar(path)
+    if sc is None:
+        return None
+    try:
+        data = json.loads(sc.read_text("utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(data, dict):
+        return None
+    sd = parse_sidecar(data)
+    if not sd.platform:
+        return None
+    artist_name = derive_top_level_artist(path, import_root)
+    if not artist_name:
+        return None
+    artist = session.execute(
+        select(Artist).where(Artist.slug == slugify(artist_name))
+    ).scalar_one_or_none()
+    if artist is None:
+        return None
+    src = session.execute(
+        select(Source)
+        .where(
+            Source.artist_id == artist.id,
+            Source.platform == sd.platform,
+            Source.enabled.is_(True),
+        )
+        .order_by(Source.id.asc())
+    ).scalars().first()
+    if src is None:
+        return None
+    if (src.url or "").startswith("sidecar:"):
+        return None  # synthetic anchor — not a pollable feed
+    return src
+
+
+def attempt_refetch(
+    session: Session, task: ImportTask, import_root: Path,
+) -> dict:
+    """Delete the corrupt file, mark the task refetched, and trigger ONE
+    source re-check. Idempotent/bounded: a task already refetched (or
+    with no resolvable Source) is a no-op. Commits."""
+    if task.refetched:
+        return {"status": "already_refetched"}
+    src = resolve_refetch_source(session, task.source_path, import_root)
+    if src is None:
+        return {"status": "no_source"}
+
+    # Remove the bad copy so gallery-dl (skip_existing) re-fetches it on
+    # the source re-check instead of skipping the still-present corrupt
+    # file.
+    try:
+        Path(task.source_path).unlink(missing_ok=True)
+    except OSError as exc:
+        log.warning("refetch unlink failed for %s: %s", task.source_path, exc)
+
+    task.refetched = True
+    session.add(task)
+    session.commit()
+
+    # Lazy import to avoid a tasks→services→tasks import cycle at module
+    # load. download_source.delay() is sync-safe in any context.
+    from ..tasks.download import download_source
+
+    download_source.delay(src.id)
+    return {"status": "refetch_queued", "source_id": src.id}

backend/app/tasks/import_file.py

@@ -64,30 +64,13 @@ def _mark_failed(session, task, error_msg: str) -> None:
             pass
 
 
-@celery.task(
-    name="backend.app.tasks.import_file.import_media_file",
-    bind=True,
-    autoretry_for=(OperationalError, DBAPIError, OSError),
-    retry_backoff=5,
-    retry_backoff_max=60,
-    retry_jitter=True,
-    max_retries=3,
-    soft_time_limit=300,
-    time_limit=360,
-)
-def import_media_file(self, import_task_id: int) -> dict:
-    """Returns a dict so the eager-mode tests can assert without DB.
-
-    Decorator notes:
-      - autoretry_for: transient DB / filesystem errors retry with
-        exponential backoff (5s base, jitter, max 3 attempts). On final
-        give-up the task raises and acks_late=True (set globally on the
-        Celery app) does NOT redeliver — the recovery sweep catches the
-        row instead.
-      - soft_time_limit (300s) raises SoftTimeLimitExceeded in this
-        process so the task can mark its row failed before being killed.
-      - time_limit (360s) is the hard cap; SIGKILL if the soft signal
-        was swallowed.
+def _run_import_task(import_task_id: int) -> dict:
+    """Shared body for import_media_file + import_archive_file. The two
+    tasks differ ONLY in their Celery time limits (a single media file
+    is sub-second; an archive runs the full per-member pipeline inline
+    for every member and can take many minutes). Both flip the row to
+    'processing', dispatch to `_do_import`, and honor the
+    flip-to-terminal resilience contract.
     """
     SessionLocal = _sync_session_factory()
     with SessionLocal() as session:
@@ -103,19 +86,85 @@ def import_media_file(self, import_task_id: int) -> dict:
         try:
             return _do_import(session, task, import_task_id)
         except SoftTimeLimitExceeded:
-            _mark_failed(session, task, "soft_time_limit exceeded (>300s)")
+            _mark_failed(session, task, "soft_time_limit exceeded")
             raise
         except (OperationalError, DBAPIError, OSError):
             # Retryable per the decorator; do NOT mark failed (let
             # autoretry have a clean go at it). If autoretry exhausts,
             # the row stays 'processing' and the maintenance sweep
-            # flips it within 5 min.
+            # flips it.
             raise
         except Exception as exc:  # noqa: BLE001 — pipeline crash, mark + re-raise
             _mark_failed(session, task, f"{type(exc).__name__}: {exc}")
             raise
 
 
+@celery.task(
+    name="backend.app.tasks.import_file.import_media_file",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=300,
+    time_limit=360,
+)
+def import_media_file(self, import_task_id: int) -> dict:
+    """Import ONE media file (or non-media → PostAttachment). Sub-second
+    for the common case; the tight 5-min soft limit keeps a genuinely
+    stuck single-file import detectable fast.
+
+    Decorator notes:
+      - autoretry_for: transient DB / filesystem errors retry with
+        exponential backoff (5s base, jitter, max 3 attempts). On final
+        give-up the task raises and acks_late=True (set globally on the
+        Celery app) does NOT redeliver — the recovery sweep catches the
+        row instead.
+      - soft_time_limit (300s) raises SoftTimeLimitExceeded in-process
+        so the task can mark its row failed before being killed.
+      - time_limit (360s) is the hard SIGKILL cap.
+    """
+    return _run_import_task(import_task_id)
+
+
+@celery.task(
+    name="backend.app.tasks.import_file.import_archive_file",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    # Archives run the full per-member pipeline (sha256 + pHash + dedup
+    # query + copy + provenance) for EVERY media member inline, under a
+    # single task budget. A multi-hundred-member archive blows the
+    # 5-min media limit. soft=30min / hard=35min sizes for a large
+    # archive. Operator-flagged 2026-05-28 (target 1645019 hit the old
+    # shared 300s soft limit). The recovery sweep gives this task its
+    # own 40-min threshold via maintenance.TASK_STUCK_THRESHOLD_MINUTES
+    # so it isn't preempted while legitimately grinding through members.
+    soft_time_limit=1800,
+    time_limit=2100,
+)
+def import_archive_file(self, import_task_id: int) -> dict:
+    """Import an archive: extract + run the per-member media pipeline for
+    every member inline, then preserve the archive as a PostAttachment.
+    Same body as import_media_file (dispatch is by file kind inside
+    Importer.import_one); split out purely for the larger time budget."""
+    return _run_import_task(import_task_id)
+
+
+def enqueue_import(task_id: int, task_type: str) -> None:
+    """Route an ImportTask to the right Celery task by its task_type.
+    Single source of truth for the media-vs-archive dispatch so the
+    scan, retry, and recovery-requeue paths stay in sync."""
+    if task_type == "archive":
+        import_archive_file.delay(task_id)
+    else:
+        import_media_file.delay(task_id)
+
+
 def _do_import(session, task, import_task_id: int) -> dict:
     """Actual work, called from inside the resilience wrapper."""
     settings = session.execute(

backend/app/tasks/maintenance.py

@@ -1,12 +1,13 @@
 """Periodic maintenance: recover stuck import tasks, garbage-collect old finished tasks."""
 
 import logging
+import os
 import subprocess
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import delete, select, update
+from sqlalchemy import and_, delete, or_, select, update
 
 from ..celery_app import celery
 from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun
@@ -16,6 +17,22 @@ from ._sync_engine import sync_session_factory as _sync_session_factory
 log = logging.getLogger(__name__)
 
 STUCK_THRESHOLD_MINUTES = 5
+# Archive ImportTasks run the per-member pipeline inline for every
+# member (import_archive_file: soft=30min/hard=35min). The ImportTask
+# 'processing' recovery sweep must give them a longer threshold or it
+# re-queues a legitimately-running archive mid-import (double-process).
+# 40 min = 5-min buffer past the archive task's hard kill.
+# Operator-flagged 2026-05-28 (target 1645019, a big archive).
+ARCHIVE_STUCK_THRESHOLD_MINUTES = 40
+
+# Poison-pill cap. After being recovered (re-queued from a stuck
+# 'processing' state) MAX_RECOVERY_ATTEMPTS-1 times, the next sweep
+# marks the row 'failed' instead of looping. 3 = two recoveries then
+# give up. A row reaches this only if it leaves NO terminal flip each
+# run — i.e. it hard-crashes the worker (OOM/segfault/SIGKILL), the
+# signature of a corrupt or oversized input. Caught exceptions already
+# flip to terminal 'failed' and never enter this loop.
+MAX_RECOVERY_ATTEMPTS = 3
 ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
@@ -24,17 +41,40 @@ FFPROBE_TIMEOUT_SECONDS = 10
 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
 TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 
+# Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep).
+# Tasks/queues that legitimately run longer than the default 5-min
+# threshold need their own larger value, else the sweep marks in-flight
+# work 'error' before it finishes. Each value MUST be ≥ the relevant
+# task.time_limit + a small buffer. task_name overrides take precedence
+# over queue overrides.
+#
+# ml queue: tag_and_embed video branch (≈20 GPU ops); time_limit=1200.
+# import_archive_file: shares the 'import' queue with the fast
+#   single-file import_media_file, so it needs a task-name override
+#   (the import queue itself stays at the 5-min default for single
+#   files); time_limit=2100.
+QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
+    "ml": 25,
+}
+TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
+    "backend.app.tasks.import_file.import_archive_file": 40,
+}
+
 
 @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
     """Recover stuck ImportTask rows. Two distinct stuck states:
 
-    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.
+    1. 'processing' too long — worker crash mid-import. Re-queue via
+       enqueue_import (routing media vs archive) and let the import
+       retry. Threshold is task-type-aware: media files are sub-second
+       and capped at the 5-min soft limit, so STUCK_THRESHOLD_MINUTES
+       (5) means a confirmed crash; archives run the per-member
+       pipeline inline (import_archive_file, 35-min hard limit) so they
+       get ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid re-queueing a
+       still-running archive. (Media was tightened from 30 min to 5
+       2026-05-24 after a 2224-row zombie pile; archive split out
+       2026-05-28.)
 
     2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
        creates rows with status='pending' (commit), then in a second pass
@@ -51,7 +91,8 @@ def recover_interrupted_tasks() -> int:
     """
     SessionLocal = _sync_session_factory()
     now = datetime.now(UTC)
-    processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    media_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    archive_cutoff = now - timedelta(minutes=ARCHIVE_STUCK_THRESHOLD_MINUTES)
     orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
     with SessionLocal() as session:
         # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which
@@ -59,20 +100,67 @@ def recover_interrupted_tasks() -> int:
         # tens of thousands of rows (operator hit it 2026-05-26 after the
         # /import deep scan piled up orphans). Folding the SELECT into the
         # UPDATE eliminates the IN-list entirely. RETURNING gives us back
-        # exactly the ids that flipped so the stuck sweep can still
-        # .delay() each one.
-        stuck_result = session.execute(
+        # exactly the (id, task_type) pairs that flipped so the requeue
+        # can route media vs archive correctly.
+        #
+        # Media + archive get separate cutoffs: a single media file is
+        # sub-second so 5 min means crash; an archive runs the per-member
+        # pipeline inline and can legitimately take up to its 35-min hard
+        # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid
+        # re-queueing a still-running archive.
+        stuck_predicate = and_(
+            ImportTask.status == "processing",
+            or_(
+                and_(ImportTask.task_type != "archive",
+                     ImportTask.started_at < media_cutoff),
+                and_(ImportTask.task_type == "archive",
+                     ImportTask.started_at < archive_cutoff),
+            ),
+        )
+
+        # POISON-PILL CIRCUIT BREAKER (Layer 1, 2026-05-28). A row that
+        # leaves no terminal flip (hard worker crash: OOM/segfault/SIGKILL
+        # on a corrupt or oversized input) gets re-queued by this sweep —
+        # and would loop forever, re-crashing the worker each pass,
+        # without a cap. Once a row has already been recovered
+        # MAX_RECOVERY_ATTEMPTS-1 times, stop re-queueing it and mark it
+        # 'failed' with a diagnostic so the operator can find + replace
+        # the offending file. This UPDATE runs FIRST so the rows it
+        # claims drop out of 'processing' before the re-queue pass.
+        poison_result = session.execute(
             update(ImportTask)
-            .where(ImportTask.status == "processing")
-            .where(ImportTask.started_at < processing_cutoff)
+            .where(stuck_predicate)
+            .where(ImportTask.recovery_count >= MAX_RECOVERY_ATTEMPTS - 1)
             .values(
-                status="queued",
-                started_at=None,
-                error="recovered from stuck state",
+                status="failed",
+                finished_at=now,
+                error=(
+                    f"crashed or stalled the worker {MAX_RECOVERY_ATTEMPTS} "
+                    f"times without completing — likely a corrupt or "
+                    f"oversized input. Not re-queued. Inspect/replace the "
+                    f"file, then retry via /api/import/retry-failed."
+                ),
             )
             .returning(ImportTask.id)
         )
-        stuck_ids = [row[0] for row in stuck_result.all()]
+        poison_ids = [r[0] for r in poison_result.all()]
+
+        # Re-queue the remaining stuck rows (under the cap) and bump
+        # their recovery_count. RETURNING (id, task_type) so the requeue
+        # routes media vs archive correctly.
+        stuck_result = session.execute(
+            update(ImportTask)
+            .where(stuck_predicate)
+            .where(ImportTask.recovery_count < MAX_RECOVERY_ATTEMPTS - 1)
+            .values(
+                status="queued",
+                started_at=None,
+                recovery_count=ImportTask.recovery_count + 1,
+                error="recovered from stuck state",
+            )
+            .returning(ImportTask.id, ImportTask.task_type)
+        )
+        stuck = stuck_result.all()
 
         orphan_result = session.execute(
             update(ImportTask)
@@ -91,12 +179,34 @@ def recover_interrupted_tasks() -> int:
 
         session.commit()
 
-        if stuck_ids:
-            from .import_file import import_media_file
-            for tid in stuck_ids:
-                import_media_file.delay(tid)
+        if stuck:
+            from .import_file import enqueue_import
+            for tid, task_type in stuck:
+                enqueue_import(tid, task_type)
 
-        return len(stuck_ids) + orphan_count
+        # Layer-2 auto re-download (env-gated, default OFF). For each
+        # poison-pill row that resolves to a pollable Source, delete the
+        # bad file and trigger ONE source re-check to fetch a fresh
+        # copy. Bounded by ImportTask.refetched so source-side
+        # corruption can't loop. The 'failed' row stays as history; the
+        # re-downloaded file re-imports as a fresh task on the next scan.
+        if poison_ids and os.environ.get("FC_AUTO_REFETCH_CORRUPT", "0") == "1":
+            from ..models import ImportSettings
+            from ..services.refetch_service import attempt_refetch
+            import_root = Path(session.execute(
+                select(ImportSettings.import_scan_path)
+                .where(ImportSettings.id == 1)
+            ).scalar_one())
+            for pid in poison_ids:
+                ptask = session.get(ImportTask, pid)
+                if ptask is None:
+                    continue
+                try:
+                    attempt_refetch(session, ptask, import_root)
+                except Exception as exc:  # noqa: BLE001 — best-effort
+                    log.warning("auto-refetch failed for task %s: %s", pid, exc)
+
+        return len(stuck) + len(poison_ids) + orphan_count
 
 
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")
@@ -121,18 +231,29 @@ def cleanup_old_tasks() -> int:
 
 @celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs")
 def recover_stalled_task_runs() -> int:
-    """Flip task_run rows stuck in 'running' for >STUCK_THRESHOLD_MINUTES
-    to 'error'. FC-3i.
+    """Flip task_run rows stuck in 'running' past their queue-specific
+    threshold to 'error'. FC-3i.
 
     A row gets stuck when the worker dies without emitting
     task_postrun / task_failure (e.g. OOM, container restart between
-    signals, signal handler raised+logged). Shares the 5-min threshold
-    with recover_interrupted_tasks for consistency.
+    signals, signal handler raised+logged). The default 5-min threshold
+    fits short-lived queues (import/thumbnail/download); queues that
+    legitimately run longer tasks (ml-video, deep scans) get their
+    own larger threshold via QUEUE_STUCK_THRESHOLD_MINUTES so the
+    sweep doesn't preempt them.
+
+    Runs once per distinct threshold value: each pass updates rows
+    whose queue maps to that threshold.
     """
     SessionLocal = _sync_session_factory()
-    cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
-    with SessionLocal() as session:
-        result = session.execute(
+    now = datetime.now(UTC)
+    override_tasks = set(TASK_STUCK_THRESHOLD_MINUTES.keys())
+    override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys())
+    total = 0
+
+    def _flag(minutes, *extra_where):
+        cutoff = now - timedelta(minutes=minutes)
+        stmt = (
             update(TaskRun)
             .where(TaskRun.status == "running")
             .where(TaskRun.started_at < cutoff)
@@ -140,14 +261,42 @@ def recover_stalled_task_runs() -> int:
                 status="error",
                 error_type="RecoverySweep",
                 error_message=(
-                    f"no completion signal received within "
-                    f"{STUCK_THRESHOLD_MINUTES} min"
+                    f"no completion signal received within {minutes} min"
                 ),
-                finished_at=datetime.now(UTC),
+                finished_at=now,
             )
         )
+        for w in extra_where:
+            stmt = stmt.where(w)
+        return session.execute(stmt).rowcount or 0
+
+    with SessionLocal() as session:
+        # Precedence: task_name override → queue override → default.
+        # Each pass excludes rows claimed by a higher-precedence pass so
+        # every row is touched at most once.
+
+        # 1. Per-task-name overrides (e.g. import_archive_file, which
+        #    shares the 'import' queue with fast single-file imports).
+        for task_name, minutes in TASK_STUCK_THRESHOLD_MINUTES.items():
+            total += _flag(minutes, TaskRun.task_name == task_name)
+
+        # 2. Per-queue overrides, excluding the override task-names.
+        for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items():
+            wheres = [TaskRun.queue == queue]
+            if override_tasks:
+                wheres.append(TaskRun.task_name.notin_(override_tasks))
+            total += _flag(minutes, *wheres)
+
+        # 3. Default — everything not claimed above.
+        default_wheres = []
+        if override_queues:
+            default_wheres.append(TaskRun.queue.notin_(override_queues))
+        if override_tasks:
+            default_wheres.append(TaskRun.task_name.notin_(override_tasks))
+        total += _flag(STUCK_THRESHOLD_MINUTES, *default_wheres)
+
         session.commit()
-        return result.rowcount or 0
+        return total
 
 
 @celery.task(name="backend.app.tasks.maintenance.prune_task_runs")

backend/app/tasks/ml.py

@@ -31,8 +31,15 @@ def _is_video(path: Path) -> bool:
     retry_backoff_max=60,
     retry_jitter=True,
     max_retries=3,
-    soft_time_limit=300,
-    time_limit=420,
+    # Sized for the video branch: sample 10 frames, run tagger +
+    # embedder on each (≈20 GPU ops vs 2 for an image). A loaded
+    # ml-worker can take 5-10 min on a long video; bumped from
+    # 5min/7min on 2026-05-28 after operator-flagged image 6288 (a
+    # .mp4) hit the recovery sweep at 5 min while still legitimately
+    # processing. Image runs return in seconds; the bump doesn't
+    # affect their UX.
+    soft_time_limit=900,   # 15 min
+    time_limit=1200,       # 20 min hard
 )
 def tag_and_embed(self, image_id: int) -> dict:
     """Run Camie + SigLIP on one image; store predictions + embedding;
@@ -64,6 +71,18 @@ def tag_and_embed(self, image_id: int) -> dict:
         embedder = get_embedder()
 
         if _is_video(src):
+            # Layer-3 isolation: ffprobe (a separate process) validates
+            # the container before we burn ~20 GPU ops sampling frames
+            # from it. A corrupt video that would crash the frame
+            # decoder is rejected cleanly here instead of taking down
+            # the ml-worker. Operator-flagged 2026-05-28.
+            from ..utils import safe_probe
+            vprobe = safe_probe.probe_video(src)
+            if not vprobe.ok:
+                return {
+                    "status": "bad_video", "image_id": image_id,
+                    "reason": vprobe.reason,
+                }
             frames = _sample_video_frames(
                 src, int(os.environ.get("VIDEO_ML_FRAMES", "10"))
             )

backend/app/tasks/scan.py

@@ -17,6 +17,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from ..celery_app import celery
 from ..config import get_config
 from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
+from ..services.archive_extractor import is_archive
 from ..services.scheduler_service import select_due_sources
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
@@ -96,7 +97,9 @@ def scan_directory(self, triggered_by: str = "manual",
             task = ImportTask(
                 batch_id=batch_id,
                 source_path=entry_str,
-                task_type="media",
+                # Archives route to import_archive_file (larger time
+                # budget) — they run the per-member pipeline inline.
+                task_type="archive" if is_archive(entry) else "media",
                 status="pending",
                 size_bytes=size,
             )
@@ -115,15 +118,16 @@ def scan_directory(self, triggered_by: str = "manual",
             batch.finished_at = datetime.now(UTC)
         session.commit()
 
-        # Now enqueue import_media_file for each pending task.
+        # Now enqueue each pending task on the right Celery task
+        # (media vs archive) via the shared router.
+        from .import_file import enqueue_import
+
         for task in session.execute(
             select(ImportTask).where(ImportTask.batch_id == batch_id)
         ).scalars():
             task.status = "queued"
             session.add(task)
-            from .import_file import import_media_file
-
-            import_media_file.delay(task.id)
+            enqueue_import(task.id, task.task_type)
         session.commit()
 
         if mode == "deep":

backend/app/utils/safe_probe.py

@@ -0,0 +1,170 @@
+"""Subprocess-isolated media probes (Layer 3 of import resilience).
+
+A malformed video or archive can hard-crash the worker process — a
+decoder OOM, a native-lib segfault, or a decompression bomb. A hard
+crash leaves no terminal flip, so the recovery sweep re-queues the row
+and it crashes again: a poison-pill loop (the Layer-1 cap is the
+backstop, but isolating the crash is better — the file gets a clean
+terminal failure and the worker never dies).
+
+These probes run the risky read in a way that contains the blast:
+
+- Video: `ffprobe` is a separate binary, so a crash decoding the
+  container kills only ffprobe (non-zero exit), never the worker. Also
+  returns width/height, which the importer didn't previously capture
+  for videos.
+- Archive: an uncompressed-size guard (catches decompression bombs
+  before they OOM anything) plus an integrity test in a spawned child
+  (catches native-lib crashes on a malformed archive). A child segfault
+  / OOM shows up as a non-zero exit code, not a dead worker.
+
+Images are intentionally NOT probed here: Pillow raises (it doesn't
+segfault) on the realistic corrupt-image cases, the importer already
+catches that as an invalid_image skip, and a subprocess per image would
+wreck deep-scan throughput on a large library. Add an image branch only
+if a real image-induced worker crash is ever observed.
+
+Operator-requested 2026-05-28 (Layer 3).
+"""
+
+import json
+import multiprocessing as mp
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+VIDEO_PROBE_TIMEOUT_SECONDS = 60
+ARCHIVE_PROBE_TIMEOUT_SECONDS = 120
+# Refuse archives whose total UNCOMPRESSED size exceeds this — the
+# classic decompression-bomb guard (a 4 GB cap comfortably clears real
+# art-pack archives while stopping a few-KB zip that expands to TB).
+MAX_ARCHIVE_UNCOMPRESSED_BYTES = 4 * 1024 * 1024 * 1024
+
+
+@dataclass(frozen=True)
+class ProbeResult:
+    ok: bool
+    # crashed=True means the probe HARD-FAILED (subprocess killed by a
+    # signal, OOM, or timeout) — the poison-pill signature. crashed=False
+    # with ok=False means a clean rejection (corrupt-but-handled,
+    # bomb-size-exceeded, integrity mismatch). Callers map crashed → a
+    # terminal 'failed', clean → a 'skipped'/'failed' of their choosing.
+    crashed: bool = False
+    reason: str | None = None
+    width: int | None = None
+    height: int | None = None
+
+
+def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) -> ProbeResult:
+    """Validate a video container + first video stream via ffprobe."""
+    try:
+        out = subprocess.run(
+            [
+                "ffprobe", "-v", "error",
+                "-select_streams", "v:0",
+                "-show_entries", "stream=width,height",
+                "-of", "json", str(path),
+            ],
+            capture_output=True, text=True, timeout=timeout,
+        )
+    except subprocess.TimeoutExpired:
+        return ProbeResult(ok=False, crashed=True, reason="ffprobe timed out")
+    except OSError as exc:
+        # ffprobe missing / not executable — environmental, not the
+        # file's fault. Treat as a clean non-crash failure so the import
+        # path can decide (it currently proceeds without dims).
+        return ProbeResult(ok=False, crashed=False, reason=f"ffprobe unavailable: {exc}")
+    if out.returncode != 0:
+        return ProbeResult(
+            ok=False, crashed=False,
+            reason=f"ffprobe rejected the file: {out.stderr.strip()[:200]}",
+        )
+    try:
+        streams = (json.loads(out.stdout) or {}).get("streams") or []
+    except json.JSONDecodeError as exc:
+        return ProbeResult(ok=False, crashed=False, reason=f"ffprobe output parse failed: {exc}")
+    if not streams:
+        return ProbeResult(ok=False, crashed=False, reason="no decodable video stream")
+    return ProbeResult(
+        ok=True, width=streams[0].get("width"), height=streams[0].get("height"),
+    )
+
+
+def probe_archive(path: Path, *, timeout: float = ARCHIVE_PROBE_TIMEOUT_SECONDS) -> ProbeResult:
+    """Bomb-size guard + isolated integrity test for an archive."""
+    ctx = mp.get_context("spawn")
+    q = ctx.Queue()
+    proc = ctx.Process(target=_archive_probe_target, args=(str(path), q))
+    proc.start()
+    proc.join(timeout)
+    if proc.is_alive():
+        proc.terminate()
+        proc.join(5)
+        return ProbeResult(ok=False, crashed=True, reason="archive probe timed out")
+    if proc.exitcode != 0:
+        # Negative exitcode = killed by signal (segfault); positive =
+        # the child os._exit'd or was OOM-killed. Either way the file
+        # hard-crashed the probe — the poison-pill signature.
+        return ProbeResult(
+            ok=False, crashed=True,
+            reason=f"archive probe crashed (exit {proc.exitcode})",
+        )
+    try:
+        outcome = q.get(timeout=5)
+    except Exception:  # noqa: BLE001 — empty queue / broken pipe
+        return ProbeResult(ok=False, crashed=True, reason="archive probe produced no result")
+    status, detail = outcome
+    if status == "ok":
+        return ProbeResult(ok=True)
+    return ProbeResult(ok=False, crashed=False, reason=detail)
+
+
+def _archive_probe_target(path_str: str, q) -> None:
+    """Runs in the spawned child. Reads member sizes (bomb guard) then
+    runs the format's integrity test. Puts ('ok', None) or
+    ('error', reason). A crash/OOM here never reaches the queue — the
+    parent reads the non-zero exit code instead."""
+    path = Path(path_str)
+    ext = path.suffix.lower()
+    try:
+        total, test_bad = _inspect_archive(path, ext)
+    except Exception as exc:  # noqa: BLE001 — clean rejection
+        q.put(("error", f"{type(exc).__name__}: {exc}"))
+        return
+    if total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES:
+        gib = total / (1024 ** 3)
+        q.put(("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap"))
+        return
+    if test_bad is not None:
+        q.put(("error", f"integrity test failed at member {test_bad!r}"))
+        return
+    q.put(("ok", None))
+
+
+def _inspect_archive(path: Path, ext: str):
+    """Return (total_uncompressed_bytes | None, first_bad_member | None)
+    for the archive. Format-specific; raises on a structurally-broken
+    container (caught by the child as a clean rejection)."""
+    if ext in (".zip", ".cbz"):
+        import zipfile
+
+        with zipfile.ZipFile(path) as zf:
+            total = sum(zi.file_size for zi in zf.infolist())
+            return total, zf.testzip()
+    if ext == ".rar":
+        import rarfile
+
+        with rarfile.RarFile(path) as rf:
+            total = sum(getattr(ri, "file_size", 0) for ri in rf.infolist())
+            rf.testrar()
+            return total, None
+    if ext == ".7z":
+        import py7zr
+
+        with py7zr.SevenZipFile(path, "r") as zf:
+            info = zf.archiveinfo()
+            total = getattr(info, "uncompressed", None)
+            ok = zf.test()  # True / None when all members pass
+            return total, (None if ok in (True, None) else "7z test reported corruption")
+    # Unknown extension — nothing to test; treat as clean.
+    return None, None

backend/app/utils/sidecar.py

@@ -1,7 +1,9 @@
 """Minimal gallery-dl sidecar parsing (one-time filesystem-import aid).
 
-No per-platform branching: a small common key set with fallbacks; the
-full JSON is kept in raw so anything unmapped is recoverable later.
+Per-platform quirks (post_url synthesis, key-chain overrides) live in
+the platforms registry — `backend/app/services/platforms/`. This module
+is platform-agnostic: it looks up `category` in the sidecar and asks
+the registry for the right behavior.
 """
 
 import re
@@ -9,6 +11,12 @@ from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
 
+from ..services.platforms import (
+    PLATFORMS,
+    description_keys_for,
+    external_post_id_keys_for,
+)
+
 
 @dataclass(frozen=True)
 class SidecarData:
@@ -55,6 +63,19 @@ def _first_str(data: dict, keys: tuple[str, ...]) -> str | None:
     return None
 
 
+def _first_id(data: dict, keys: tuple[str, ...]) -> str | None:
+    """Like `_first_str` but accepts ints and rejects bool (Python's
+    bool subclasses int, so a literal `"id": true` would otherwise
+    yield external_post_id="True")."""
+    for k in keys:
+        v = data.get(k)
+        if isinstance(v, bool):
+            continue
+        if isinstance(v, (str, int)) and str(v).strip():
+            return str(v).strip()
+    return None
+
+
 # Strip HTML tags + collapse whitespace + take the first non-empty line.
 # Used to derive a display title from a body when the platform doesn't
 # expose a separate title field (subscribestar posts always write
@@ -111,22 +132,7 @@ def parse_sidecar(data: dict) -> SidecarData:
     cat = data.get("category")
     platform = cat if isinstance(cat, str) and cat.strip() else None
 
-    # external_post_id lookup order: post_id MUST come before id.
-    # SubscribeStar gallery-dl writes the per-attachment id in `id`
-    # (e.g. 711509) and the actual post id in `post_id` (e.g. 360360);
-    # picking `id` first fragments every multi-image subscribestar post
-    # into N distinct Post rows in FC. Patreon/Pixiv have no `post_id`
-    # so `id` still wins for them; HF uses `index`, Discord uses
-    # `message_id` — all reached via the remaining chain entries.
-    # Operator-flagged 2026-05-27 during the sidecar audit.
-    external_post_id = None
-    for k in ("post_id", "id", "index", "message_id"):
-        v = data.get(k)
-        if isinstance(v, bool):
-            continue
-        if isinstance(v, (str, int)) and str(v).strip():
-            external_post_id = str(v)
-            break
+    external_post_id = _first_id(data, external_post_id_keys_for(platform))
 
     pc = data.get("page_count")
     if isinstance(pc, bool):
@@ -146,30 +152,23 @@ def parse_sidecar(data: dict) -> SidecarData:
             if post_date is not None:
                 break
 
-    # `message` is Discord gallery-dl's body field (no `content`); added
-    # 2026-05-27 to the description fallback chain.
-    description = _first_str(
-        data, ("content", "description", "caption", "message"),
-    )
+    description = _first_str(data, description_keys_for(platform))
 
-    # SubscribeStar posts always write `title: ""` and put the leading
-    # sentence inside `content` (confirmed against the operator's
-    # /mnt/Data/Patreon/Cheunart/subscribestar/ dump 2026-05-27). When
-    # no explicit title is present, synthesize one from the description
-    # body's first non-empty line. Patreon retains its explicit titles
-    # because they're non-empty and short-circuit the fallback.
+    # When `title` is empty (subscribestar always; sometimes elsewhere),
+    # synthesize from the description body's first non-empty text line.
+    # Patreon's explicit titles short-circuit the fallback.
     post_title = _first_str(data, ("title",))
     if post_title is None and description:
         post_title = _first_line_text(description)
 
-    # post_url derivation: SubscribeStar/Pixiv/HF/Discord put the FILE
-    # download URL in `url`, not a post permalink. Synthesize the
-    # permalink from per-platform fields when possible. Patreon's `url`
-    # IS a permalink and is used as-is. For the four file-URL platforms,
-    # the bare `url` is NEVER trusted — derive or return None rather
-    # than persist a CDN URL in post.post_url.
-    if platform in _DERIVED_URL_PLATFORMS:
-        post_url = _derive_post_url(platform, data)
+    # post_url: ask the platform module to synthesize a permalink.
+    # When the platform registers a `derive_post_url`, it owns the
+    # field (the bare `url`/`post_url` value is a file CDN URL and
+    # must NEVER be persisted). When it doesn't register one, trust
+    # the sidecar's `url` (Patreon's case — real permalink).
+    info = PLATFORMS.get(platform) if platform else None
+    if info is not None and info.derive_post_url is not None:
+        post_url = info.derive_post_url(data)
     else:
         post_url = _first_str(data, ("url", "post_url"))
 
@@ -183,39 +182,3 @@ def parse_sidecar(data: dict) -> SidecarData:
         post_date=post_date,
         raw=data,
     )
-
-
-_DERIVED_URL_PLATFORMS = frozenset({
-    "subscribestar", "pixiv", "hentaifoundry", "discord",
-})
-
-
-def _derive_post_url(platform: str, data: dict) -> str | None:
-    """Synthesize the post-permalink URL from per-platform metadata.
-
-    gallery-dl writes the file-download URL in `url` for these four
-    platforms; we need a real permalink for the PostCard "open original"
-    button. Returns None if the platform-specific fields are missing
-    (rare in well-formed sidecars but defensive).
-    """
-    if platform == "subscribestar":
-        pid = data.get("post_id")
-        if isinstance(pid, (str, int)) and str(pid).strip():
-            return f"https://www.subscribestar.com/posts/{pid}"
-    elif platform == "pixiv":
-        pid = data.get("id")
-        if isinstance(pid, (str, int)) and not isinstance(pid, bool) and str(pid).strip():
-            return f"https://www.pixiv.net/artworks/{pid}"
-    elif platform == "hentaifoundry":
-        user = _first_str(data, ("user", "artist"))
-        idx = data.get("index")
-        if user and isinstance(idx, (str, int)) and not isinstance(idx, bool) and str(idx).strip():
-            return f"https://www.hentai-foundry.com/pictures/user/{user}/{idx}"
-    elif platform == "discord":
-        sid = data.get("server_id")
-        cid = data.get("channel_id")
-        mid = data.get("message_id")
-        if all(isinstance(v, (str, int)) and not isinstance(v, bool) and str(v).strip()
-               for v in (sid, cid, mid)):
-            return f"https://discord.com/channels/{sid}/{cid}/{mid}"
-    return None

extension/lib/cookies.js

@@ -38,11 +38,33 @@ function deduplicateCookies(cookies) {
 function toNetscapeFormat(cookies) {
   const lines = ['# Netscape HTTP Cookie File'];
   for (const c of cookies) {
-    let domain = c.domain.replace(/^\.?www\./, '.');
-    if (!domain.startsWith('.')) domain = '.' + domain;
+    // Preserve the browser's actual scope. Earlier versions rewrote
+    // every cookie to a leading-dot subdomain-wide form, which broke
+    // gallery-dl's HF extractor: its `cookies.get(name,
+    // domain="www.hentai-foundry.com")` does EXACT domain matching and
+    // missed host-only PHPSESSID rewritten to `.hentai-foundry.com`.
+    // Operator-flagged 2026-05-27. Backend `_augment_cookies` covers
+    // the already-stored cookies; this fix is forward-compat for fresh
+    // captures.
+    //
+    // Cookie storage semantics (Firefox):
+    //   c.hostOnly === true   → cookie set without a Domain= attribute;
+    //                           applies to the exact host only.
+    //   c.hostOnly === false  → cookie set with Domain=X; applies to
+    //                           that domain and its subdomains.
+    //
+    // Netscape format:
+    //   leading-dot domain + TRUE flag → subdomain-wide
+    //   bare-host  domain + FALSE flag → host-only
+    const hostOnly = c.hostOnly === true;
+    let domain = c.domain;
+    if (!hostOnly && !domain.startsWith('.')) {
+      domain = '.' + domain;
+    }
+    const subdomainFlag = hostOnly ? 'FALSE' : 'TRUE';
     const secure = c.secure ? 'TRUE' : 'FALSE';
     const expiration = c.expirationDate ? Math.floor(c.expirationDate) : 0;
-    lines.push([domain, 'TRUE', c.path || '/', secure, String(expiration), c.name, c.value].join('\t'));
+    lines.push([domain, subdomainFlag, c.path || '/', secure, String(expiration), c.name, c.value].join('\t'));
   }
   return lines.join('\n');
 }

extension/manifest.json

@@ -1,7 +1,7 @@
 {
   "manifest_version": 3,
   "name": "FabledCurator",
-  "version": "1.0.4",
+  "version": "1.0.5",
   "description": "Export cookies from supported platforms to FabledCurator and add creators as sources in one click.",
 
   "browser_specific_settings": {

extension/package.json

@@ -1,6 +1,6 @@
 {
   "name": "fabledcurator-extension",
-  "version": "1.0.4",
+  "version": "1.0.5",
   "private": true,
   "description": "Firefox extension for FabledCurator",
   "scripts": {

frontend/src/components/cleanup/MinDimensionCard.vue

@@ -52,8 +52,8 @@
       v-model="showModal"
       action="delete"
       kind="min-dim"
-      :run-id="tokenSha8"
       tier="C"
+      :expected-token-override="preview?.confirm_token || ''"
       :projected-counts="projectedCounts"
       :description="`Width < ${minW} OR height < ${minH}`"
       @confirm="onConfirmedDelete"
@@ -67,13 +67,20 @@ import { onMounted, ref } from 'vue'
 import DestructiveConfirmModal from '../modal/DestructiveConfirmModal.vue'
 import { useCleanupStore } from '../../stores/cleanup.js'
 
+// Backend's preview response hands the full Tier-C confirm token back
+// as `confirm_token` (e.g. `delete-min-dim-1a2b3c4d`); passed straight
+// to the modal via `expected-token-override`. We previously
+// reconstructed via Web Crypto's SHA-256, but `crypto.subtle` is
+// Secure-Context-gated and undefined on plain-HTTP origins, so the
+// Delete button silently swallowed the TypeError. Operator-flagged
+// 2026-05-27.
+
 const store = useCleanupStore()
 const minW = ref(0)
 const minH = ref(0)
 const preview = ref(null)
 const busy = ref(false)
 const showModal = ref(false)
-const tokenSha8 = ref('')
 const projectedCounts = ref({})
 
 onMounted(async () => {
@@ -82,15 +89,6 @@ onMounted(async () => {
   minH.value = store.defaults.min_height
 })
 
-// SHA-256 truncated to 8 hex chars — matches the backend's
-// _min_dim_token() exactly. Web Crypto rejects MD5 as insecure.
-async function sha8(canon) {
-  const enc = new TextEncoder()
-  const buf = await crypto.subtle.digest('SHA-256', enc.encode(canon))
-  const hex = Array.from(new Uint8Array(buf)).map(b => b.toString(16).padStart(2, '0')).join('')
-  return hex.slice(0, 8)
-}
-
 async function onPreview() {
   busy.value = true
   try {
@@ -102,8 +100,7 @@ async function onPreview() {
   }
 }
 
-async function onDeleteClick() {
-  tokenSha8.value = await sha8(`${minW.value}x${minH.value}`)
+function onDeleteClick() {
   projectedCounts.value = { 'Images to delete': preview.value.count }
   showModal.value = true
 }

frontend/src/components/discovery/MasonryGrid.vue

@@ -4,7 +4,10 @@
       <div v-for="(col, ci) in columns" :key="ci" class="fc-masonry__col">
         <button
           v-for="item in col" :key="item.id"
-          class="fc-masonry__item" type="button"
+          class="fc-masonry__item"
+          :class="{ 'fc-masonry__item--anim': shouldAnimate(item) }"
+          :style="itemStyle(item)"
+          type="button"
           @click="$emit('open', item.id)"
         >
           <img
@@ -33,7 +36,13 @@ import { usePolyMasonry } from '../../composables/usePolyMasonry.js'
 const props = defineProps({
   items: { type: Array, default: () => [] },
   loading: { type: Boolean, default: false },
-  hasMore: { type: Boolean, default: false }
+  hasMore: { type: Boolean, default: false },
+  // Items at indices >= animateFromIndex get the stagger fade-in. Opt-in
+  // — defaults to Infinity (no animation) so views that don't want it
+  // (ArtistView, etc.) don't pay the layout-shift cost. ShowcaseView
+  // uses 0 on initial load / shuffle and prevCount on infinite-scroll
+  // appends.
+  animateFromIndex: { type: Number, default: Number.POSITIVE_INFINITY },
 })
 const emit = defineEmits(['load-more', 'open'])
 
@@ -43,6 +52,25 @@ const { columnCount, distribute } = usePolyMasonry(containerEl)
 
 const columns = computed(() => distribute(props.items, columnCount.value))
 
+// id → index lookup so we can derive the stagger from natural reading
+// order even after the masonry distributes items across columns.
+const idxById = computed(() => {
+  const m = new Map()
+  props.items.forEach((it, i) => m.set(it.id, i))
+  return m
+})
+
+function shouldAnimate(item) {
+  const idx = idxById.value.get(item.id)
+  return idx !== undefined && idx >= props.animateFromIndex
+}
+
+function itemStyle(item) {
+  if (!shouldAnimate(item)) return {}
+  const idx = idxById.value.get(item.id) - props.animateFromIndex
+  return { '--stagger-index': idx }
+}
+
 function aspectStyle(item) {
   const w = Number(item.width)
   const h = Number(item.height)
@@ -81,4 +109,23 @@ onUnmounted(() => observer && observer.disconnect())
   display: flex; justify-content: center; padding: 32px 0; min-height: 60px;
 }
 .fc-masonry__end { text-align: center; padding: 32px 0; }
+
+/* IR-parity stagger fade-in for showcase entry / shuffle. 60ms between
+   items, 250ms each — matches IR's `itemFadeIn` keyframe (style.css
+   ~line 1834). Honors prefers-reduced-motion. */
+@keyframes fc-masonry-item-in {
+  from { opacity: 0; transform: translateY(12px); }
+  to   { opacity: 1; transform: translateY(0); }
+}
+.fc-masonry__item--anim {
+  animation: fc-masonry-item-in 0.25s ease forwards;
+  animation-delay: calc(var(--stagger-index, 0) * 60ms);
+  opacity: 0;
+}
+@media (prefers-reduced-motion: reduce) {
+  .fc-masonry__item--anim {
+    animation: none;
+    opacity: 1;
+  }
+}
 </style>

frontend/src/components/downloads/DownloadEventRow.vue

@@ -1,47 +1,118 @@
 <template>
-  <div class="fc-dl-row" @click="$emit('open', event.id)">
-    <v-icon :icon="statusIcon" :color="statusColor" size="small" />
+  <div
+    class="fc-dl-row"
+    :class="[`fc-dl-row--${event.status || 'unknown'}`]"
+    @click="$emit('open', event.id)"
+  >
+    <!-- Colored left edge marks the run's status; matches the row's
+         status-chip color but reads at a glance without needing to
+         parse the chip text. -->
+    <div class="fc-dl-row__bar" />
+
+    <v-chip
+      :color="statusColor"
+      size="small"
+      variant="tonal"
+      :prepend-icon="statusIcon"
+      class="fc-dl-row__status"
+    >{{ statusLabel }}</v-chip>
+
     <RouterLink
       v-if="event.artist_slug"
       :to="`/artist/${event.artist_slug}`"
       class="fc-dl-row__artist"
       @click.stop
     >{{ event.artist_name }}</RouterLink>
-    <span v-else class="fc-dl-row__artist">—</span>
-    <v-chip size="x-small" variant="tonal">{{ event.platform || '—' }}</v-chip>
-    <span class="fc-dl-row__time">{{ fmtTime(event.started_at) }}</span>
-    <span class="fc-dl-row__files">{{ event.files_count }} files</span>
-    <span class="fc-dl-row__duration">{{ fmtDuration(event.summary?.duration_seconds) }}</span>
-    <span v-if="event.error" class="fc-dl-row__error">{{ event.error }}</span>
+    <span v-else class="fc-dl-row__artist fc-dl-row__artist--missing">—</span>
+
+    <PlatformChip
+      v-if="event.platform"
+      :platform="event.platform"
+      size="x-small"
+      class="fc-dl-row__platform"
+    />
+    <span v-else class="fc-dl-row__platform-missing">—</span>
+
+    <span class="fc-dl-row__time" :title="event.started_at">
+      {{ fmtTime(event.started_at) }}
+    </span>
+
+    <v-chip
+      v-if="event.files_count > 0"
+      size="x-small" variant="tonal" color="info"
+      prepend-icon="mdi-image-multiple"
+      class="fc-dl-row__files"
+    >{{ event.files_count }}</v-chip>
+    <span v-else class="fc-dl-row__no-files" aria-label="no new files">·</span>
+
+    <span class="fc-dl-row__duration">
+      {{ fmtDuration(event.summary?.duration_seconds) }}
+    </span>
+
+    <v-chip
+      v-if="event.error"
+      color="error" size="x-small" variant="tonal"
+      prepend-icon="mdi-alert-octagon"
+      class="fc-dl-row__error"
+      :title="event.error"
+    >{{ truncateError(event.error) }}</v-chip>
+    <span v-else class="fc-dl-row__error-spacer" />
+
+    <div class="fc-dl-row__actions" @click.stop>
+      <v-btn
+        v-if="event.status === 'error' && event.source_id"
+        icon size="x-small" variant="text" color="warning"
+        :loading="retrying"
+        @click.stop="onRetry"
+      >
+        <v-icon size="small">mdi-refresh</v-icon>
+        <v-tooltip activator="parent" location="top">Retry source check</v-tooltip>
+      </v-btn>
+      <v-btn icon size="x-small" variant="text" @click.stop="$emit('open', event.id)">
+        <v-icon size="small">mdi-information-outline</v-icon>
+        <v-tooltip activator="parent" location="top">Details</v-tooltip>
+      </v-btn>
+      <v-btn
+        v-if="event.artist_slug"
+        icon size="x-small" variant="text"
+        :to="`/artist/${event.artist_slug}`"
+        @click.stop
+      >
+        <v-icon size="small">mdi-account-circle</v-icon>
+        <v-tooltip activator="parent" location="top">Open artist</v-tooltip>
+      </v-btn>
+    </div>
   </div>
 </template>
 
 <script setup>
-import { computed } from 'vue'
+import { computed, ref } from 'vue'
 import { RouterLink } from 'vue-router'
 
+import PlatformChip from '../subscriptions/PlatformChip.vue'
+import { useSourcesStore } from '../../stores/sources.js'
+
 const props = defineProps({ event: { type: Object, required: true } })
 defineEmits(['open'])
 
-const statusIcon = computed(() => ({
-  ok: 'mdi-check-circle',
-  error: 'mdi-alert-circle',
-  running: 'mdi-progress-clock',
-  pending: 'mdi-clock-outline',
-  skipped: 'mdi-minus-circle',
-}[props.event.status] || 'mdi-help-circle'))
+const sourcesStore = useSourcesStore()
+const retrying = ref(false)
 
-const statusColor = computed(() => ({
-  ok: 'success',
-  error: 'error',
-  running: 'info',
-  pending: 'secondary',
-  skipped: 'warning',
-}[props.event.status] || undefined))
+const _STATUS = {
+  ok:      { color: 'success', icon: 'mdi-check-circle',    label: 'Completed' },
+  error:   { color: 'error',   icon: 'mdi-alert-circle',    label: 'Failed' },
+  running: { color: 'info',    icon: 'mdi-progress-clock',  label: 'Running' },
+  pending: { color: 'grey',    icon: 'mdi-clock-outline',   label: 'Queued' },
+  skipped: { color: 'warning', icon: 'mdi-skip-next',       label: 'Skipped' },
+}
+const statusColor = computed(() => _STATUS[props.event.status]?.color || 'grey')
+const statusIcon = computed(() => _STATUS[props.event.status]?.icon || 'mdi-help-circle')
+const statusLabel = computed(() => _STATUS[props.event.status]?.label || props.event.status)
 
 function fmtTime(iso) {
   if (!iso) return '—'
-  return iso.slice(0, 19).replace('T', ' ')
+  // 2026-05-27 23:36 — second granularity is in the row's title attr
+  return iso.slice(0, 16).replace('T', ' ')
 }
 function fmtDuration(sec) {
   if (sec == null) return '—'
@@ -49,34 +120,107 @@ function fmtDuration(sec) {
   const m = Math.floor(sec / 60), s = Math.floor(sec % 60)
   return `${m}m ${s}s`
 }
+function truncateError(msg) {
+  const s = String(msg || '')
+  if (s.length <= 60) return s
+  return s.slice(0, 57) + '…'
+}
+
+async function onRetry() {
+  if (!props.event.source_id) return
+  retrying.value = true
+  try {
+    await sourcesStore.checkNow(props.event.source_id)
+    globalThis.window?.__fcToast?.({
+      text: `Source check re-queued`, type: 'success',
+    })
+  } catch (e) {
+    const isInFlight = !!e?.body?.download_event_id
+    globalThis.window?.__fcToast?.({
+      text: isInFlight ? 'Already running' : `Retry failed: ${e?.detail || e?.message || e}`,
+      type: isInFlight ? 'info' : 'error',
+    })
+  } finally {
+    retrying.value = false
+  }
+}
 </script>
 
 <style scoped>
 .fc-dl-row {
+  position: relative;
   display: grid;
-  grid-template-columns: 24px 1fr 96px 160px 80px 80px 1fr;
-  gap: 0.75rem;
+  grid-template-columns:
+    /* bar */    4px
+    /* status */ 120px
+    /* artist */ minmax(120px, 1.2fr)
+    /* plat   */ 140px
+    /* time   */ 140px
+    /* files  */ 60px
+    /* dur    */ 70px
+    /* error  */ minmax(0, 1.5fr)
+    /* actions*/ 120px;
+  gap: 0.6rem;
   align-items: center;
-  padding: 0.5rem 0.75rem;
-  border-bottom: 1px solid rgb(var(--v-theme-on-surface-variant) / 0.18);
+  padding: 0.55rem 0.75rem 0.55rem 0;
+  border-bottom: 1px solid rgb(var(--v-theme-on-surface-variant) / 0.15);
   cursor: pointer;
+  transition: background 0.12s ease;
 }
-.fc-dl-row:hover { background: rgb(var(--v-theme-surface) / 0.5); }
+.fc-dl-row:hover {
+  background: rgb(var(--v-theme-on-surface) / 0.04);
+}
+.fc-dl-row__bar {
+  width: 4px;
+  align-self: stretch;
+  border-radius: 0 2px 2px 0;
+}
+.fc-dl-row--ok      .fc-dl-row__bar { background: rgb(var(--v-theme-success)); }
+.fc-dl-row--error   .fc-dl-row__bar { background: rgb(var(--v-theme-error));   }
+.fc-dl-row--running .fc-dl-row__bar { background: rgb(var(--v-theme-info));    }
+.fc-dl-row--skipped .fc-dl-row__bar { background: rgb(var(--v-theme-warning)); }
+.fc-dl-row--pending .fc-dl-row__bar { background: rgb(var(--v-theme-on-surface-variant) / 0.4); }
+
+.fc-dl-row__status { justify-self: start; }
 .fc-dl-row__artist {
   color: rgb(var(--v-theme-on-surface));
   text-decoration: none;
+  font-weight: 500;
+  overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
+}
+.fc-dl-row__artist--missing,
+.fc-dl-row__platform-missing,
+.fc-dl-row__no-files {
+  color: rgb(var(--v-theme-on-surface-variant));
+  opacity: 0.5;
 }
 .fc-dl-row__artist:hover { color: rgb(var(--v-theme-accent)); }
-.fc-dl-row__time, .fc-dl-row__files, .fc-dl-row__duration {
+.fc-dl-row__platform { justify-self: start; }
+.fc-dl-row__time,
+.fc-dl-row__duration {
   color: rgb(var(--v-theme-on-surface-variant));
   font-size: 0.85rem;
   font-variant-numeric: tabular-nums;
-}
-.fc-dl-row__error {
-  color: rgb(var(--v-theme-error));
-  font-size: 0.85rem;
-  overflow: hidden;
-  text-overflow: ellipsis;
   white-space: nowrap;
 }
+.fc-dl-row__no-files {
+  text-align: center;
+  font-size: 1.1rem;
+}
+.fc-dl-row__error {
+  justify-self: start;
+  max-width: 100%;
+}
+.fc-dl-row__error :deep(.v-chip__content) {
+  overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
+}
+.fc-dl-row__error-spacer { /* keeps the grid column reserved */ }
+
+.fc-dl-row__actions {
+  display: flex; gap: 2px;
+  justify-self: end;
+  opacity: 0.5;
+  transition: opacity 0.12s ease;
+}
+.fc-dl-row:hover .fc-dl-row__actions { opacity: 1; }
 </style>

frontend/src/components/gallery/BulkEditorPanel.vue

@@ -74,7 +74,7 @@
       v-model="deleteModalOpen"
       action="delete"
       kind="images-selection"
-      :run-id="bulkToken"
+      :expected-token-override="bulkProjected?.confirm_token || ''"
       tier="C"
       :projected-counts="bulkProjectedCounts"
       :description="bulkDescription"
@@ -130,7 +130,6 @@ watch(() => sel.order.length, () => {
 const deleting = ref(false)
 const deleteModalOpen = ref(false)
 const bulkProjected = ref(null)
-const bulkToken = ref('')
 
 const bulkProjectedCounts = computed(() => bulkProjected.value
   ? {
@@ -147,24 +146,22 @@ const bulkDescription = computed(
     : '',
 )
 
-async function _computeSha8(ids) {
-  const canon = [...ids].sort((a, b) => a - b).join(',')
-  const buf = new TextEncoder().encode(canon)
-  const hashBuf = await crypto.subtle.digest('SHA-256', buf)
-  const bytes = new Uint8Array(hashBuf)
-  let hex = ''
-  for (let i = 0; i < 4; i++) {
-    hex += bytes[i].toString(16).padStart(2, '0')
-  }
-  return hex
-}
+// The dry-run response hands the canonical Tier-C confirm token back
+// as `confirm_token` (e.g. `delete-images-1a2b3c4d`), passed straight
+// to the modal via `expected-token-override`. We used to compute the
+// hash client-side via `crypto.subtle.digest`, but (1) that's
+// Secure-Context-gated and undefined on plain-HTTP origins
+// (homelab posture), so the click silently threw TypeError and the
+// modal never opened, and (2) the modal's `kind="images-selection"`
+// produced `delete-images-selection-<sha8>` while the backend
+// expected `delete-images-<sha8>` — so it never would have worked
+// even on HTTPS. Operator-flagged 2026-05-27.
 
 async function onDeleteClick() {
   if (!sel.order.length) return
   deleting.value = true
   try {
     bulkProjected.value = await adminStore.projectBulkImageDelete(sel.order)
-    bulkToken.value = await _computeSha8(sel.order)
     deleteModalOpen.value = true
   } finally {
     deleting.value = false

frontend/src/components/modal/DestructiveConfirmModal.vue

@@ -58,17 +58,27 @@ import { computed, ref, watch } from 'vue'
 const props = defineProps({
   modelValue: { type: Boolean, required: true },
   action: { type: String, required: true },          // 'restore' | 'delete'
-  kind: { type: String, required: true },            // 'db' | 'images' | 'artist' | 'tag' | 'images-selection'
+  kind: { type: String, required: true },            // 'db' | 'images' | 'artist' | 'tag' | 'images-selection' | 'audit' | 'min-dim'
   runId: { type: [Number, String], default: '' },    // numeric id or sha8 string
   description: { type: String, default: '' },
   tier: { type: String, default: 'C' },              // 'B' | 'C'
   projectedCounts: { type: Object, default: null },
+  // Override the `${action}-${kind}-${runId}` token formula. Use when
+  // the backend computes the canonical confirm token (e.g. bulk-delete
+  // and min-dim cleanup both return `confirm_token` from their dry-run
+  // endpoints) and the UI's kind/runId would otherwise produce a
+  // mismatched string. Operator-flagged 2026-05-27 after the
+  // BulkEditor's kind="images-selection" produced
+  // `delete-images-selection-<sha8>` while the backend expected
+  // `delete-images-<sha8>`.
+  expectedTokenOverride: { type: String, default: '' },
 })
 const emit = defineEmits(['update:modelValue', 'confirm'])
 
 const typed = ref('')
 const expectedToken = computed(
-  () => `${props.action}-${props.kind}-${props.runId}`,
+  () => props.expectedTokenOverride
+    || `${props.action}-${props.kind}-${props.runId}`,
 )
 const titleVerb = computed(
   () => props.action === 'restore' ? 'Restore' : 'Delete',

frontend/src/components/settings/ImportTaskList.vue

@@ -51,6 +51,19 @@
           title="Click for full error"
         >{{ shorten(item.error, 60) }}</button>
       </template>
+      <template #item.actions="{ item }">
+        <v-btn
+          v-if="item.status === 'failed'"
+          icon size="x-small" variant="text"
+          :loading="refetching === item.id"
+          @click="onRefetch(item)"
+        >
+          <v-icon size="small">mdi-cloud-refresh</v-icon>
+          <v-tooltip activator="parent" location="top">
+            Re-fetch original (re-download from source)
+          </v-tooltip>
+        </v-btn>
+      </template>
     </v-data-table-virtual>
     <div v-if="store.hasMore" class="d-flex justify-center py-3">
       <v-btn variant="text" size="small" @click="onLoadMore">Load more</v-btn>
@@ -149,9 +162,29 @@ const headers = [
   { title: 'Source', key: 'source_path', sortable: false },
   { title: 'Size', key: 'size_bytes', sortable: false, width: 90 },
   { title: 'Created', key: 'created_at', sortable: false, width: 150 },
-  { title: 'Note', key: 'error', sortable: false }
+  { title: 'Note', key: 'error', sortable: false },
+  { title: '', key: 'actions', sortable: false, width: 56 }
 ]
 
+const refetching = ref(null)
+const _REFETCH_MSG = {
+  refetch_queued: { text: 'Re-fetch queued — re-downloading from source', type: 'success' },
+  no_source: { text: 'No re-fetchable source (filesystem import — replace the file manually)', type: 'info' },
+  already_refetched: { text: 'Already re-fetched once', type: 'info' },
+}
+async function onRefetch(item) {
+  refetching.value = item.id
+  try {
+    const res = await store.refetchTask(item.id)
+    const msg = _REFETCH_MSG[res.status] || { text: `Re-fetch: ${res.status}`, type: 'info' }
+    window.__fcToast?.(msg)
+  } catch (e) {
+    window.__fcToast?.({ text: `Re-fetch failed: ${e.message}`, type: 'error' })
+  } finally {
+    refetching.value = null
+  }
+}
+
 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'

frontend/src/components/subscriptions/DownloadsTab.vue

@@ -20,15 +20,44 @@
       <v-progress-circular indeterminate color="accent" size="36" />
     </div>
 
-    <div v-else-if="store.events.length === 0" class="fc-dl__empty">
+    <div v-else-if="filteredEvents.length === 0" class="fc-dl__empty">
       <p>No download events match the current filter.</p>
     </div>
 
     <div v-else>
-      <DownloadEventRow
-        v-for="e in filteredEvents" :key="e.id" :event="e"
-        @open="openDetail"
-      />
+      <section
+        v-for="g in groups" :key="g.key"
+        class="fc-dl__group"
+      >
+        <header
+          class="fc-dl__group-head"
+          role="button" tabindex="0"
+          @click="toggle(g.key)" @keydown.enter="toggle(g.key)"
+        >
+          <v-icon size="small" class="fc-dl__group-chev">
+            {{ collapsed[g.key] ? 'mdi-chevron-right' : 'mdi-chevron-down' }}
+          </v-icon>
+          <span class="fc-dl__group-label">{{ g.label }}</span>
+          <span class="fc-dl__group-counts">
+            <v-chip
+              v-if="g.failedCount > 0"
+              size="x-small" color="error" variant="tonal"
+              prepend-icon="mdi-alert-circle"
+            >{{ g.failedCount }}</v-chip>
+            <v-chip size="x-small" variant="tonal">
+              {{ g.items.length }}
+              {{ g.items.length === 1 ? 'event' : 'events' }}
+            </v-chip>
+          </span>
+        </header>
+        <div v-if="!collapsed[g.key]" class="fc-dl__group-body">
+          <DownloadEventRow
+            v-for="e in g.items" :key="e.id" :event="e"
+            @open="openDetail"
+          />
+        </div>
+      </section>
+
       <div class="fc-dl__sentinel">
         <v-btn v-if="store.hasMore" variant="text" @click="store.loadMore()" :loading="store.loading">
           Load more
@@ -45,7 +74,7 @@
 </template>
 
 <script setup>
-import { computed, onMounted, ref, watch } from 'vue'
+import { computed, onMounted, reactive, ref, watch } from 'vue'
 import { useRoute } from 'vue-router'
 
 import { useDownloadsStore } from '../../stores/downloads.js'
@@ -59,6 +88,16 @@ const route = useRoute()
 const store = useDownloadsStore()
 const filterModel = ref({ ...store.filter })
 
+// Each group's collapsed state persists across refreshes for the
+// lifetime of the SubscriptionsView (operator-friendly default: all
+// expanded; collapse what you don't care about right now).
+const collapsed = reactive({
+  today: false, yesterday: false, week: false, earlier: false,
+})
+function toggle(key) {
+  collapsed[key] = !collapsed[key]
+}
+
 async function refresh() {
   await Promise.all([
     store.loadFirst(),
@@ -91,6 +130,48 @@ const filteredEvents = computed(() => {
   return arr
 })
 
+// Group events by relative date bucket and pin failed runs to the
+// top of each bucket. Buckets boundaries are computed against the
+// operator's local-time start-of-day so "Today" matches their
+// intuition regardless of the event's stored UTC timestamp.
+const groups = computed(() => {
+  const now = new Date()
+  const startOfToday = new Date(
+    now.getFullYear(), now.getMonth(), now.getDate(),
+  ).getTime()
+  const startOfYesterday = startOfToday - 24 * 3600 * 1000
+  const startOfWeek = startOfToday - 7 * 24 * 3600 * 1000
+
+  const buckets = { today: [], yesterday: [], week: [], earlier: [] }
+  for (const e of filteredEvents.value) {
+    const t = new Date(e.started_at).getTime()
+    if (t >= startOfToday) buckets.today.push(e)
+    else if (t >= startOfYesterday) buckets.yesterday.push(e)
+    else if (t >= startOfWeek) buckets.week.push(e)
+    else buckets.earlier.push(e)
+  }
+
+  function withFailedPinned(items) {
+    const fail = items.filter((e) => e.status === 'error')
+    const rest = items.filter((e) => e.status !== 'error')
+    return [...fail, ...rest]
+  }
+
+  const meta = [
+    { key: 'today', label: 'Today' },
+    { key: 'yesterday', label: 'Yesterday' },
+    { key: 'week', label: 'Last 7 days' },
+    { key: 'earlier', label: 'Earlier' },
+  ]
+  return meta
+    .map(({ key, label }) => {
+      const items = withFailedPinned(buckets[key])
+      const failedCount = items.filter((e) => e.status === 'error').length
+      return { key, label, items, failedCount }
+    })
+    .filter((g) => g.items.length > 0)
+})
+
 watch(filterModel, async (m) => {
   await store.applyFilter({
     status: m.status,
@@ -120,4 +201,29 @@ async function openDetail(id) {
 .fc-dl__sentinel {
   display: flex; justify-content: center; padding: 1rem 0;
 }
+
+.fc-dl__group { margin-bottom: 12px; }
+.fc-dl__group-head {
+  display: flex; align-items: center; gap: 8px;
+  padding: 6px 8px;
+  background: rgb(var(--v-theme-on-surface) / 0.04);
+  border-radius: 4px;
+  cursor: pointer;
+  user-select: none;
+}
+.fc-dl__group-head:hover {
+  background: rgb(var(--v-theme-on-surface) / 0.08);
+}
+.fc-dl__group-chev {
+  color: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-dl__group-label {
+  font-weight: 600;
+  color: rgb(var(--v-theme-on-surface));
+  flex: 1;
+}
+.fc-dl__group-counts {
+  display: flex; gap: 6px; align-items: center;
+}
+.fc-dl__group-body { margin-top: 4px; }
 </style>

frontend/src/stores/import.js

@@ -146,6 +146,15 @@ export const useImportStore = defineStore('import', () => {
     return body
   }
 
+  // Layer-2 one-shot re-download for a failed task's (corrupt) file.
+  // Returns the endpoint's status dict (refetch_queued / no_source /
+  // already_refetched). Caller surfaces it as a toast.
+  async function refetchTask(taskId) {
+    const body = await api.post(`/api/import/tasks/${taskId}/refetch`)
+    await loadTasks(true)
+    return body
+  }
+
   const hasMore = computed(() => tasksNextCursor.value !== null)
 
   return {
@@ -155,6 +164,7 @@ export const useImportStore = defineStore('import', () => {
     triggerError,
     loadSettings, patchSettings,
     refreshStatus, triggerScan,
-    loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck
+    loadTasks, setStatusFilter, retryFailed, clearCompleted, clearStuck,
+    refetchTask,
   }
 })

frontend/src/views/ShowcaseView.vue

@@ -18,6 +18,7 @@
       :items="store.images"
       :loading="store.loading"
       :has-more="store.hasMore"
+      :animate-from-index="animateFromIndex"
       @load-more="store.fetchPage()"
       @open="openImage"
     />
@@ -25,17 +26,52 @@
 </template>
 
 <script setup>
-import { onMounted } from 'vue'
-import { useShowcaseStore } from '../stores/showcase.js'
-import { useModalStore } from '../stores/modal.js'
+import { onMounted, onUnmounted, ref, watch } from 'vue'
+
 import MasonryGrid from '../components/discovery/MasonryGrid.vue'
+import { useModalStore } from '../stores/modal.js'
+import { useShowcaseStore } from '../stores/showcase.js'
 
 const store = useShowcaseStore()
 const modal = useModalStore()
 
-onMounted(() => { if (store.images.length === 0) store.fetchPage() })
+// Track when items were appended vs replaced so MasonryGrid only animates
+// items new to the current batch (mirrors IR's behavior: animate on
+// initial load and on shuffle, but skip silent infinite-scroll appends).
+const animateFromIndex = ref(0)
+let prevCount = 0
+watch(() => store.images.length, (newCount) => {
+  if (newCount < prevCount || prevCount === 0) {
+    // Reset (shuffle) or initial load — animate everything from 0.
+    animateFromIndex.value = 0
+  } else {
+    // Append — only animate the newly-added tail.
+    animateFromIndex.value = prevCount
+  }
+  prevCount = newCount
+})
 
 function openImage(id) {
   modal.open(id)
 }
+
+// IR-parity keyboard shuffle: press R anywhere on the page (not inside a
+// text input, not while a Vuetify overlay is open) to reshuffle.
+function onKeydown(e) {
+  const t = e.target
+  if (t && (t.tagName === 'INPUT' || t.tagName === 'TEXTAREA' || t.isContentEditable)) return
+  // Vuetify marks the active overlay (v-dialog, v-menu) on the body when
+  // open. Skip shuffle when a modal is in the way.
+  if (document.querySelector('.v-overlay--active')) return
+  if (e.key === 'r' || e.key === 'R') {
+    e.preventDefault()
+    store.shuffle()
+  }
+}
+
+onMounted(() => {
+  if (store.images.length === 0) store.fetchPage()
+  window.addEventListener('keydown', onKeydown)
+})
+onUnmounted(() => window.removeEventListener('keydown', onKeydown))
 </script>

tests/test_api_admin.py

@@ -140,6 +140,11 @@ async def test_bulk_delete_dry_run_returns_counts(client, db, tmp_path):
     assert body["images_found"] == 2
     assert body["bytes_on_disk"] == 30
     assert body["missing_ids"] == [9_999_999]
+    # Dry-run hands the canonical Tier-C confirm token back so the
+    # frontend doesn't recompute SHA-256 client-side (crypto.subtle
+    # is Secure-Context-gated; FC runs over plain HTTP).
+    assert body["confirm_token"].startswith("delete-images-")
+    assert len(body["confirm_token"]) == len("delete-images-") + 8
 
 
 @pytest.mark.asyncio

tests/test_api_cleanup.py

@@ -66,6 +66,10 @@ async def test_min_dimension_preview_returns_count(client, db, tmp_path):
     assert resp.status_code == 200
     body = await resp.get_json()
     assert body["count"] == 1
+    # Preview hands the canonical Tier-C confirm token back so the
+    # frontend doesn't have to recompute SHA-256 client-side
+    # (crypto.subtle is Secure-Context-gated; FC runs over plain HTTP).
+    assert body["confirm_token"] == _sha256_min_dim_token(200, 200)
 
 
 @pytest.mark.asyncio

tests/test_api_import_admin.py

@@ -171,6 +171,115 @@ async def test_trigger_still_rejects_unknown_mode(client):
     assert resp.status_code == 400
 
 
+@pytest.mark.asyncio
+async def test_refetch_404_for_unknown_task(client):
+    resp = await client.post("/api/import/tasks/999999/refetch")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_refetch_400_for_non_failed_task(client, db):
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    task = ImportTask(
+        batch_id=batch.id, source_path="/x.jpg", task_type="media",
+        status="complete", finished_at=datetime.now(UTC),
+    )
+    db.add(task)
+    await db.commit()
+    resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert resp.status_code == 400
+    assert (await resp.get_json())["status"] == "not_failed"
+
+
+@pytest.mark.asyncio
+async def test_refetch_no_source_when_unresolvable(client, db):
+    """A failed task whose file has no sidecar / no resolvable Source
+    returns no_source (filesystem-only import — nothing to re-poll)."""
+    batch = ImportBatch(triggered_by="manual", source_path="/import", scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    task = ImportTask(
+        batch_id=batch.id, source_path="/import/nowhere/x.jpg",
+        task_type="media", status="failed", finished_at=datetime.now(UTC),
+    )
+    db.add(task)
+    await db.commit()
+    resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["status"] == "no_source"
+
+
+@pytest.mark.asyncio
+async def test_refetch_queued_with_resolvable_source(client, db, tmp_path, monkeypatch):
+    """A failed task whose file resolves (via sidecar → artist+platform)
+    to an enabled, real-URL Source: the file is deleted, the task is
+    marked refetched, and ONE source re-check is queued."""
+    import json as _json
+
+    from sqlalchemy import update as _update
+
+    from backend.app.models import Artist, ImportSettings, Source
+    from backend.app.tasks import download as download_mod
+
+    # Stub the downloader so the eager test doesn't run a real fetch.
+    dispatched = []
+    monkeypatch.setattr(download_mod.download_source, "delay", dispatched.append)
+
+    # import_root/<ArtistName>/post.jpg + sidecar identifying the platform.
+    import_root = tmp_path / "import"
+    artist_dir = import_root / "Maewix"
+    artist_dir.mkdir(parents=True)
+    media = artist_dir / "post.jpg"
+    media.write_bytes(b"corrupt-bytes")
+    (artist_dir / "post.jpg.json").write_text(
+        _json.dumps({"category": "patreon", "post_id": 123})
+    )
+
+    # import_settings(id=1) is migration-seeded; point its scan path at
+    # our tmp import root rather than inserting a conflicting row.
+    await db.execute(
+        _update(ImportSettings).where(ImportSettings.id == 1)
+        .values(import_scan_path=str(import_root))
+    )
+    artist = Artist(name="Maewix", slug="maewix")
+    db.add(artist)
+    await db.flush()
+    db.add(Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://www.patreon.com/maewix", enabled=True,
+        config_overrides={},
+    ))
+    batch = ImportBatch(triggered_by="manual", source_path=str(import_root), scan_mode="quick")
+    db.add(batch)
+    await db.flush()
+    task = ImportTask(
+        batch_id=batch.id, source_path=str(media), task_type="media",
+        status="failed", finished_at=datetime.now(UTC),
+    )
+    db.add(task)
+    await db.commit()
+
+    resp = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["status"] == "refetch_queued"
+    assert len(dispatched) == 1
+    assert not media.exists()  # corrupt copy removed for re-fetch
+
+    from sqlalchemy import select as _select
+    refetched = (await db.execute(
+        _select(ImportTask.refetched).where(ImportTask.id == task.id)
+    )).scalar_one()
+    assert refetched is True
+
+    # Second attempt is a no-op (bounded to one).
+    resp2 = await client.post(f"/api/import/tasks/{task.id}/refetch")
+    assert (await resp2.get_json())["status"] == "already_refetched"
+    assert len(dispatched) == 1
+
+
 @pytest.mark.asyncio
 async def test_trigger_accepts_verify(client, monkeypatch):
     # Stub the verify task's dispatch so the API contract is asserted

tests/test_credential_service.py

@@ -110,6 +110,114 @@ async def test_get_cookies_path_none_for_token_kind(db, crypto, tmp_path):
     assert await svc.get_cookies_path("discord") is None
 
 
+@pytest.mark.asyncio
+async def test_get_cookies_path_subscribestar_injects_age_cookie(db, crypto, tmp_path):
+    """SubscribeStar's server gates artist pages behind a _personalization_id
+    cookie; the browser-stored cookie expires annually and can't be easily
+    refreshed (the JS age popup is suppressed by localStorage). Mirror
+    gallery-dl's own login-flow workaround by injecting
+    `18_plus_agreement_generic=true` on `.subscribestar.adult` whenever
+    cookies for subscribestar are materialized."""
+    netscape_in = (
+        "# Netscape HTTP Cookie File\n"
+        ".subscribestar.adult\tTRUE\t/\tTRUE\t1700000000\tsession_id\txyz\n"
+    )
+    svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
+    await svc.upsert(platform="subscribestar", credential_type="cookies", data=netscape_in)
+    path = await svc.get_cookies_path("subscribestar")
+    contents = path.read_text()
+    assert "18_plus_agreement_generic\ttrue" in contents
+    assert ".subscribestar.adult" in contents
+    # Original session_id cookie preserved.
+    assert "session_id\txyz" in contents
+
+
+@pytest.mark.asyncio
+async def test_get_cookies_path_subscribestar_idempotent_when_present(db, crypto, tmp_path):
+    """If the operator's captured cookies ALREADY contain the age cookie
+    (e.g. a manual paste, or a re-login), don't double-inject."""
+    netscape_in = (
+        "# Netscape HTTP Cookie File\n"
+        ".subscribestar.adult\tTRUE\t/\tTRUE\t1700000000\t18_plus_agreement_generic\ttrue\n"
+        ".subscribestar.adult\tTRUE\t/\tTRUE\t1700000000\tsession_id\txyz\n"
+    )
+    svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
+    await svc.upsert(platform="subscribestar", credential_type="cookies", data=netscape_in)
+    path = await svc.get_cookies_path("subscribestar")
+    contents = path.read_text()
+    assert contents.count("18_plus_agreement_generic") == 1
+
+
+@pytest.mark.asyncio
+async def test_get_cookies_path_non_subscribestar_unchanged(db, crypto, tmp_path):
+    """The age-cookie injection MUST NOT fire for non-subscribestar
+    platforms — Patreon/etc. don't need it and shouldn't carry a
+    foreign-domain cookie in their cookies.txt."""
+    svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
+    await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE)
+    path = await svc.get_cookies_path("patreon")
+    contents = path.read_text()
+    assert "18_plus_agreement_generic" not in contents
+    assert "subscribestar" not in contents
+
+
+@pytest.mark.asyncio
+async def test_get_cookies_path_hf_injects_host_only_phpsessid(db, crypto, tmp_path):
+    """HF: extension writes session cookies as subdomain-wide
+    (`.hentai-foundry.com`), but gallery-dl's extractor uses
+    `cookies.get(name, domain='www.hentai-foundry.com')` with EXACT
+    domain matching. Emit host-only duplicates of PHPSESSID +
+    YII_CSRF_TOKEN on `www.hentai-foundry.com` so the lookup matches."""
+    netscape_in = (
+        "# Netscape HTTP Cookie File\n"
+        ".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123\n"
+        ".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tYII_CSRF_TOKEN\ttoken456\n"
+    )
+    svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
+    await svc.upsert(platform="hentaifoundry", credential_type="cookies", data=netscape_in)
+    path = await svc.get_cookies_path("hentaifoundry")
+    contents = path.read_text()
+    # Subdomain-wide originals preserved.
+    assert ".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123" in contents
+    # Host-only duplicates appended for both names.
+    assert "www.hentai-foundry.com\tFALSE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123" in contents
+    assert "www.hentai-foundry.com\tFALSE\t/\tTRUE\t1900000000\tYII_CSRF_TOKEN\ttoken456" in contents
+
+
+@pytest.mark.asyncio
+async def test_get_cookies_path_hf_idempotent_when_host_only_present(db, crypto, tmp_path):
+    """If the captured cookies already include a host-only PHPSESSID
+    on www.hentai-foundry.com (e.g. a future extension fix that
+    preserves browser hostOnly state), don't double-inject."""
+    netscape_in = (
+        "# Netscape HTTP Cookie File\n"
+        ".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123\n"
+        "www.hentai-foundry.com\tFALSE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123\n"
+    )
+    svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
+    await svc.upsert(platform="hentaifoundry", credential_type="cookies", data=netscape_in)
+    path = await svc.get_cookies_path("hentaifoundry")
+    contents = path.read_text()
+    # Should count PHPSESSID exactly twice — the original two lines, no third.
+    assert contents.count("PHPSESSID\tsess123") == 2
+
+
+@pytest.mark.asyncio
+async def test_get_cookies_path_hf_ignores_unrelated_cookies(db, crypto, tmp_path):
+    """The injection should only target session/CSRF cookies. Other HF
+    cookies (e.g. analytics) stay subdomain-wide as captured."""
+    netscape_in = (
+        "# Netscape HTTP Cookie File\n"
+        ".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\t_ga\tGA1.2.x\n"
+    )
+    svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
+    await svc.upsert(platform="hentaifoundry", credential_type="cookies", data=netscape_in)
+    path = await svc.get_cookies_path("hentaifoundry")
+    contents = path.read_text()
+    assert "www.hentai-foundry.com" not in contents
+    assert contents.count("_ga") == 1
+
+
 @pytest.mark.asyncio
 async def test_get_token_decrypts(db, crypto):
     svc = CredentialService(db, crypto)

tests/test_maintenance.py

@@ -164,6 +164,52 @@ def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch
     assert dispatched == [stuck.id]  # stuck rows re-enqueue; orphans don't
 
 
+def test_recover_interrupted_poison_pill_caps_at_max(db_sync, monkeypatch):
+    """A stuck row that's already been recovered MAX_RECOVERY_ATTEMPTS-1
+    times is marked 'failed' (with a diagnostic) instead of re-queued —
+    the circuit breaker against an input that hard-crashes the worker
+    every run. Operator-flagged 2026-05-28."""
+    from backend.app.tasks import import_file
+    from backend.app.tasks.maintenance import (
+        MAX_RECOVERY_ATTEMPTS,
+        recover_interrupted_tasks,
+    )
+    dispatched: list[int] = []
+    monkeypatch.setattr(
+        import_file.import_media_file, "delay", dispatched.append
+    )
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    # At the cap already (recovered MAX-1 times) → fail, don't re-queue.
+    poison = ImportTask(
+        batch_id=batch_id, source_path="/import/poison.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+        recovery_count=MAX_RECOVERY_ATTEMPTS - 1,
+    )
+    # One recovery short of the cap → re-queue + increment.
+    recoverable = ImportTask(
+        batch_id=batch_id, source_path="/import/ok.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+        recovery_count=MAX_RECOVERY_ATTEMPTS - 2,
+    )
+    db_sync.add_all([poison, recoverable])
+    db_sync.commit()
+
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2  # one failed + one re-queued
+
+    db_sync.refresh(poison)
+    db_sync.refresh(recoverable)
+    assert poison.status == "failed"
+    assert "corrupt or" in (poison.error or "")
+    assert recoverable.status == "queued"
+    assert recoverable.recovery_count == MAX_RECOVERY_ATTEMPTS - 1
+    # Only the recoverable row re-enqueues; the poison pill does not.
+    assert dispatched == [recoverable.id]
+
+
 def test_cleanup_old_deletes_finished_old(db_sync):
     batch_id = _make_batch(db_sync)
     now = datetime.now(UTC)
@@ -195,12 +241,13 @@ def test_cleanup_old_deletes_finished_old(db_sync):
 
 
 def _make_task_run(db_sync, *, status, started_at, finished_at=None,
-                   error_type=None):
+                   error_type=None, queue="default",
+                   task_name="backend.app.tasks.fake.t"):
     from backend.app.models import TaskRun
     row = TaskRun(
         celery_task_id="x",
-        queue="ml",
-        task_name="backend.app.tasks.fake.t",
+        queue=queue,
+        task_name=task_name,
         target_id=1,
         started_at=started_at,
         finished_at=finished_at,
@@ -262,6 +309,94 @@ def test_recover_stalled_task_runs_skips_fresh_running(db_sync):
     assert status == "running"
 
 
+def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
+    """ml-queue tasks (tag_and_embed video branch) legitimately run
+    past the default 5-min threshold. The sweep must NOT flag an
+    ml-queue task that's only been running 10 min — the override
+    threshold (25 min via QUEUE_STUCK_THRESHOLD_MINUTES) protects
+    in-flight video tagging. Operator-flagged 2026-05-28 after
+    image 6288 (mp4) was marked failed at the 5-min tick mid-run."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    now = datetime.now(UTC)
+    # 10-min-old ml-queue row: stale by the default 5-min rule but
+    # fresh by the 25-min ml override. Must survive the sweep.
+    ml_fresh_id = _make_task_run(
+        db_sync, status="running", queue="ml",
+        started_at=now - timedelta(minutes=10),
+    )
+    # 30-min-old ml-queue row: past even the ml override. Must be
+    # flagged.
+    ml_stale_id = _make_task_run(
+        db_sync, status="running", queue="ml",
+        started_at=now - timedelta(minutes=30),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 1
+
+    db_sync.expire_all()
+    ml_fresh_status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == ml_fresh_id)
+    ).scalar_one()
+    ml_stale_status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.id == ml_stale_id)
+    ).scalar_one()
+    assert ml_fresh_status == "running"
+    assert ml_stale_status == "error"
+
+
+def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
+    """import_archive_file shares the 'import' queue with fast
+    single-file import_media_file, so it gets a per-task-name override
+    (40 min) while the import queue stays at the 5-min default. A
+    10-min-old archive task-run must survive; a 50-min-old one is
+    flagged. Operator-flagged 2026-05-28."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    archive_name = "backend.app.tasks.import_file.import_archive_file"
+    now = datetime.now(UTC)
+    # Fast single-file import on the same queue, 10 min old → flagged
+    # by the default 5-min rule.
+    media_id = _make_task_run(
+        db_sync, status="running", queue="import",
+        task_name="backend.app.tasks.import_file.import_media_file",
+        started_at=now - timedelta(minutes=10),
+    )
+    # Archive on the same queue, 10 min old → survives (40-min override).
+    archive_fresh_id = _make_task_run(
+        db_sync, status="running", queue="import",
+        task_name=archive_name,
+        started_at=now - timedelta(minutes=10),
+    )
+    # Archive 50 min old → past even the 40-min override → flagged.
+    archive_stale_id = _make_task_run(
+        db_sync, status="running", queue="import",
+        task_name=archive_name,
+        started_at=now - timedelta(minutes=50),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 2  # media + stale archive
+
+    db_sync.expire_all()
+    def _status(_id):
+        return db_sync.execute(
+            select(TaskRun.status).where(TaskRun.id == _id)
+        ).scalar_one()
+    assert _status(media_id) == "error"
+    assert _status(archive_fresh_id) == "running"
+    assert _status(archive_stale_id) == "error"
+
+
 def test_prune_task_runs_deletes_ok_older_than_24h(db_sync):
     from sqlalchemy import select
 

tests/test_safe_probe.py

@@ -0,0 +1,71 @@
+"""Layer-3 subprocess-isolated probe tests.
+
+The bomb-guard cap is exercised against `_archive_probe_target` directly
+(in-process, where a monkeypatch on the module constant takes effect) —
+spawn re-imports the module in the child, so a parent-process
+monkeypatch wouldn't reach the spawned worker.
+"""
+
+import multiprocessing as mp
+import zipfile
+from pathlib import Path
+
+from backend.app.utils import safe_probe
+
+
+def _zip(path, entries):
+    with zipfile.ZipFile(path, "w") as zf:
+        for name, data in entries.items():
+            zf.writestr(name, data)
+
+
+def test_probe_archive_valid_zip(tmp_path):
+    z = tmp_path / "ok.zip"
+    _zip(z, {"a.jpg": b"hello", "b.png": b"world"})
+    res = safe_probe.probe_archive(z)
+    assert res.ok is True
+    assert res.crashed is False
+
+
+def test_probe_archive_corrupt_zip_clean_rejection(tmp_path):
+    z = tmp_path / "broken.zip"
+    z.write_bytes(b"PK\x03\x04 not really a zip past here")
+    res = safe_probe.probe_archive(z)
+    assert res.ok is False
+    # Corrupt-but-handled (zipfile raises BadZipFile in the child) — a
+    # clean rejection, not a hard crash.
+    assert res.crashed is False
+    assert res.reason
+
+
+def test_inspect_archive_reports_size_and_clean_integrity(tmp_path):
+    z = tmp_path / "sized.zip"
+    _zip(z, {"a.txt": b"x" * 100, "b.txt": b"y" * 50})
+    total, bad = safe_probe._inspect_archive(z, ".zip")
+    assert total == 150
+    assert bad is None
+
+
+def test_archive_probe_target_bomb_guard(tmp_path, monkeypatch):
+    """In-process call to the child target so the monkeypatched cap
+    takes effect. A normal zip whose uncompressed size exceeds the
+    (lowered) cap is rejected with the bomb-guard reason."""
+    monkeypatch.setattr(safe_probe, "MAX_ARCHIVE_UNCOMPRESSED_BYTES", 10)
+    z = tmp_path / "bomb.zip"
+    _zip(z, {"big.txt": b"x" * 5000})  # 5000 uncompressed > 10-byte cap
+    q = mp.get_context("spawn").Queue()
+    safe_probe._archive_probe_target(str(z), q)
+    status, detail = q.get(timeout=5)
+    assert status == "error"
+    assert "bomb-guard cap" in detail
+
+
+def test_probe_video_non_video_is_not_ok(tmp_path):
+    """A text file is not a decodable video. Whether ffprobe is present
+    (returncode != 0) or absent (OSError → 'unavailable'), the result is
+    ok=False. We don't assert on crashed/reason so the test is robust to
+    ffprobe presence in CI."""
+    f = tmp_path / "nope.txt"
+    f.write_text("definitely not a video container")
+    res = safe_probe.probe_video(f)
+    assert res.ok is False

tests/test_tasks_register.py

@@ -21,6 +21,10 @@ def test_import_media_file_registered():
     assert "backend.app.tasks.import_file.import_media_file" in celery.tasks
 
 
+def test_import_archive_file_registered():
+    assert "backend.app.tasks.import_file.import_archive_file" in celery.tasks
+
+
 def test_generate_thumbnail_registered():
     assert "backend.app.tasks.thumbnail.generate_thumbnail" in celery.tasks