fix(downloads): enqueue thumbnail + ML tasks per attached image

Operator-flagged 2026-06-01: downloaded images stayed at
thumbnail_path=NULL until a periodic backfill sweep picked them up,
surfacing as broken-thumbnail tiles in the gallery for hours after
the download landed.

Importer.attach_in_place deliberately skips inline thumbnail
generation (importer.py:591-592) so the import queue stays moving —
the CALLING task is responsible for the enqueue. tasks/import_file.py
already did this (line 228-239). tasks/download.py / download_service
did not — every gallery-dl-attached image landed un-thumbnailed.

Fix in download_service._phase3_persist: after each
`attach_in_place` returning status in (imported, superseded), fan out
`generate_thumbnail.delay()` + `tag_and_embed.delay()` for each
image_id. Lazy import avoids circular-import risk between
download_service and the celery task modules that depend on it.

Mirrors the existing pattern verbatim — single source of truth for
"what fires after a successful attach" remains a comment in two
places (filesystem-import task, download orchestrator) rather than
a shared helper, because the contexts differ enough (sync session vs
async orchestrator) that abstracting would obscure more than it'd
share.

Test covers the happy-path with two attached files: both get the
thumbnail enqueue AND the ML enqueue, with image_ids drawn from
ImportResult (so future supersede-on-attach paths stay covered).

backend/app/services/download_service.py

@@ -260,6 +260,25 @@ class DownloadService:
                     bytes_downloaded += path.stat().st_size  # noqa: ASYNC240
                 except OSError:
                     pass
+                # Enqueue thumbnail + ML for newly-attached images, matching
+                # the filesystem-import path (tasks/import_file.py:228-239).
+                # Importer.attach_in_place deliberately skips inline thumb
+                # generation to keep the import queue moving; the calling
+                # task is responsible for the enqueue. Operator-flagged
+                # 2026-06-01: without this, every downloaded image stayed
+                # at thumbnail_path=NULL until a periodic backfill swept
+                # it up, surfacing as broken-thumbnail tiles in the gallery
+                # for hours after a download landed. Lazy import to avoid
+                # circular-import risk between this service and the
+                # tasks/* modules that import it.
+                from ..tasks.ml import tag_and_embed
+                from ..tasks.thumbnail import generate_thumbnail
+                ids = list(result.member_image_ids)
+                if result.image_id is not None and result.image_id not in ids:
+                    ids.append(result.image_id)
+                for img_id in ids:
+                    generate_thumbnail.delay(img_id)
+                    tag_and_embed.delay(img_id)
             elif result.status == "skipped" and result.skip_reason and result.skip_reason.value in (
                 "duplicate_hash", "duplicate_phash",
             ):

tests/test_download_service.py

@@ -546,3 +546,70 @@ async def test_partial_error_type_maps_to_ok_status(
         select(Source.consecutive_failures).where(Source.id == source.id)
     )).scalar_one()
     assert src_after == 0
+
+
+@pytest.mark.asyncio
+async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    """Operator-flagged 2026-06-01: downloaded images stayed at
+    thumbnail_path=NULL until periodic backfill swept them up, surfacing
+    as broken-thumbnail tiles in the gallery. Importer.attach_in_place
+    deliberately skips inline generation; the calling code MUST enqueue
+    the thumbnail + ML tasks per attached image (matching the pattern in
+    tasks/import_file.py)."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+
+    images_root = tmp_path / "images"
+    f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
+    f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
+    _make_jpg(f1, split="h")
+    _make_jpg(f2, split="v")
+
+    fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
+        success=True, written_paths=[str(f1), str(f2)],
+        files_downloaded=2, stdout=f"{f1}\n{f2}\n",
+    ))
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    sync_settings.phash_threshold = 0
+    importer = Importer(
+        session=db_sync, images_root=images_root, import_root=images_root,
+        thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
+    )
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+
+    # Capture the IDs that the orchestrator hands off to each Celery task.
+    # The .delay() shim runs inside DownloadService._phase3_persist (lazy
+    # imports under ..tasks.thumbnail / ..tasks.ml), so monkeypatch the
+    # symbols on those modules.
+    thumb_calls: list[int] = []
+    ml_calls: list[int] = []
+    from backend.app.tasks import ml as ml_mod
+    from backend.app.tasks import thumbnail as thumb_mod
+    monkeypatch.setattr(
+        thumb_mod.generate_thumbnail, "delay",
+        lambda image_id: thumb_calls.append(image_id),
+    )
+    monkeypatch.setattr(
+        ml_mod.tag_and_embed, "delay",
+        lambda image_id: ml_calls.append(image_id),
+    )
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    await svc.download_source(source.id)
+
+    # Two files attached → two thumbnail enqueues + two ML enqueues, IDs
+    # match the actually-imported records (not ad-hoc — drawn from the
+    # importer's returned image_id so future supersede paths stay covered).
+    assert len(thumb_calls) == 2
+    assert len(ml_calls) == 2
+    assert sorted(thumb_calls) == sorted(ml_calls)