Merge pull request 'Recapture image→post linking (#1288) + ugoira timings + system suggestion category' (#192) from dev into main

backend/app/services/download_service.py

@@ -418,7 +418,8 @@ class DownloadService:
         # the duplicate file). Empty outside recapture mode.
         relink_pairs = getattr(dl_result, "relink_source_paths", None) or []
         relinked = 0
-        for rel_str, rel_url in relink_pairs:
+        post_linked = 0
+        for rel_str, rel_url, rel_post_id in relink_pairs:
             rel_path = Path(rel_str)
             if not rel_path.exists():  # noqa: ASYNC240
                 continue
@@ -428,13 +429,27 @@ class DownloadService:
 
             if await loop.run_in_executor(None, _relink):
                 relinked += 1
+
+            # #1288: link the on-disk image to its Post. Recapture disk-skips the
+            # media (never re-imported), so a pre-existing image (e.g. one pulled
+            # under the old gallery-dl path) otherwise stays orphaned even after
+            # the post record is written. Idempotent for already-linked images.
+            def _link(p=rel_path, pid=rel_post_id):
+                return self.importer.link_existing_image_to_post(
+                    p, pid, source=source_row, artist=artist,
+                )
+
+            if await loop.run_in_executor(None, _link):
+                post_linked += 1
         if relink_pairs:
             # recapture diagnostic: how many on-disk images got their
-            # source_filehash backfilled (inline-image localization). < total is
+            # source_filehash backfilled (inline-image localization) and how many
-            # normal — files already carrying a filehash are skipped (NULL-only).
+            # got (re)linked to their Post. < total for source_filehash is normal
+            # (files already carrying a filehash are skipped, NULL-only).
             log.info(
-                "recap: relinked source_filehash on %d/%d on-disk image(s)",
+                "recap: relinked source_filehash on %d and linked %d/%d on-disk "
-                relinked, len(relink_pairs),
+                "image(s) to their post",
+                relinked, post_linked, len(relink_pairs),
             )
 
         # Kick the off-platform file-host downloader for any links this run

backend/app/services/gallery_dl.py

@@ -157,7 +157,7 @@ class DownloadResult:
     # pairs for already-present media whose ImageRecord.source_filehash should be
     # backfilled (inline-image localization) WITHOUT re-download or unlink. Empty
     # on the gallery-dl path and outside recapture.
-    relink_source_paths: list[tuple[str, str]] = field(default_factory=list)
+    relink_source_paths: list[tuple[str, str, str]] = field(default_factory=list)
     stdout: str = ""
     stderr: str = ""
     return_code: int = 0

backend/app/services/importer.py

@@ -1313,6 +1313,113 @@ class Importer:
         self.session.commit()
         return True
 
+    def link_existing_image_to_post(
+        self, path: Path, external_post_id: str, *,
+        source: Source | None = None, artist: Artist | None = None,
+    ) -> bool:
+        """Recapture back-link: associate an ALREADY-on-disk image (matched by
+        its stored path) with its post — the link _apply_sidecar normally makes
+        at per-media import time.
+
+        Recapture disk-skips downloaded media, so a pre-existing image (e.g. one
+        pulled under the old gallery-dl path, imported as a bare record with no
+        post) never gets its `image_provenance` row / `primary_post_id`. This
+        backfills it from the walk's (on-disk path, post external id) pairing:
+        find the ImageRecord by path, find/attach the Post by (source,
+        external_post_id), upsert provenance. Idempotent (issue #1288). Returns
+        True when a record matched and was linked; no-op when the file has no
+        record or no id is given."""
+        if not external_post_id:
+            return False
+        record = self.session.execute(
+            select(ImageRecord).where(ImageRecord.path == str(path))
+        ).scalar_one_or_none()
+        if record is None:
+            return False
+        artist_id = record.artist_id or (artist.id if artist else None)
+        if artist_id is None:
+            return False
+        if record.artist_id is None:
+            record.artist_id = artist_id
+        post = self._find_or_create_post(
+            source_id=source.id if source else None,
+            external_post_id=str(external_post_id),
+            artist_id=artist_id,
+        )
+        self._attach_provenance(
+            record, post, source_id=source.id if source else None,
+        )
+        self.session.commit()
+        return True
+
+    def _attach_provenance(
+        self, record: ImageRecord, post: Post, *,
+        source_id: int | None, captured_metadata: dict | None = None,
+    ) -> None:
+        """Upsert the (image, post) `image_provenance` link + keep
+        `primary_post_id` and the denormalized gallery sort keys aligned. Shared
+        by _apply_sidecar (fresh per-media import) and link_existing_image_to_post
+        (recapture back-link, #1288) so the two paths can't diverge. Idempotent.
+
+        Race-safe (image_record_id, post_id) upsert — mirrors the
+        _find_or_create_source/post savepoint pattern. The plain
+        SELECT-then-INSERT pattern lost a race when two workers ran on the same
+        (image, post) pair (e.g. the 5-min recovery sweep re-enqueued a
+        still-running long import), planting duplicates that then broke
+        .scalar_one_or_none() on every later deep-scan rederive
+        (MultipleResultsFound). Alembic 0021's uq_image_provenance_image_post
+        UNIQUE makes this savepoint trip on collision."""
+        exists = self.session.execute(
+            select(ImageProvenance.id).where(
+                ImageProvenance.image_record_id == record.id,
+                ImageProvenance.post_id == post.id,
+            )
+        ).scalar_one_or_none()
+        if exists is None:
+            sp = self.session.begin_nested()
+            try:
+                self.session.add(
+                    ImageProvenance(
+                        image_record_id=record.id,
+                        post_id=post.id,
+                        source_id=source_id,
+                        captured_metadata=captured_metadata,
+                    )
+                )
+                self.session.flush()
+                sp.commit()
+            except IntegrityError:
+                sp.rollback()
+        if record.primary_post_id is None:
+            record.primary_post_id = post.id
+        # Keep the denormalized gallery sort key (alembic 0035) aligned with
+        # the primary post's publish date so /scroll orders off
+        # ix_image_record_effective_date instead of COALESCE-ing across the
+        # post join. Only override when THIS post is the primary AND carries
+        # a date; otherwise the column keeps its created_at-equivalent server
+        # default (matches the old COALESCE(post_date, created_at) fallback).
+        if record.primary_post_id == post.id and post.post_date is not None:
+            record.effective_date = post.post_date
+        # earliest_post_date (alembic 0071) = MIN(post_date) across ALL of this
+        # image's provenance posts, not just the primary — so the gallery can
+        # sort by original publish rather than the download/repost the primary
+        # points at. Recompute from provenance whenever a dated post is linked;
+        # the provenance row for THIS post was committed above, so the MIN
+        # includes it. Leaves the created_at default when no linked post is dated.
+        if post.post_date is not None:
+            earliest = self.session.execute(
+                select(func.min(Post.post_date))
+                .select_from(ImageProvenance)
+                .join(Post, Post.id == ImageProvenance.post_id)
+                .where(
+                    ImageProvenance.image_record_id == record.id,
+                    Post.post_date.is_not(None),
+                )
+            ).scalar_one_or_none()
+            if earliest is not None:
+                record.earliest_post_date = earliest
+        self.session.flush()
+
     def _apply_sidecar(
         self,
         record: ImageRecord,
@@ -1386,65 +1493,12 @@ class Importer:
         if not self.post_first:
             self._apply_post_fields(post, sd)
 
-        # Race-safe (image_record_id, post_id) upsert — mirrors the
+        # Link the (image, post) provenance + keep primary_post_id / the gallery
-        # _find_or_create_source/post savepoint pattern. The plain
+        # sort keys aligned — shared with the recapture back-link path (#1288).
-        # SELECT-then-INSERT pattern lost a race when two workers ran
+        self._attach_provenance(
-        # _apply_sidecar on the same (image, post) pair (e.g. the 5-min
+            record, post, source_id=src.id if src else None,
-        # recovery sweep re-enqueued a still-running long import), planting
+            captured_metadata=sd.raw,
-        # duplicates that then broke .scalar_one_or_none() on every later
+        )
-        # deep-scan rederive (MultipleResultsFound). Alembic 0021 adds the
-        # uq_image_provenance_image_post UNIQUE so this savepoint actually
-        # trips on collision.
-        exists = self.session.execute(
-            select(ImageProvenance.id).where(
-                ImageProvenance.image_record_id == record.id,
-                ImageProvenance.post_id == post.id,
-            )
-        ).scalar_one_or_none()
-        if exists is None:
-            sp = self.session.begin_nested()
-            try:
-                self.session.add(
-                    ImageProvenance(
-                        image_record_id=record.id,
-                        post_id=post.id,
-                        source_id=src.id if src else None,
-                        captured_metadata=sd.raw,
-                    )
-                )
-                self.session.flush()
-                sp.commit()
-            except IntegrityError:
-                sp.rollback()
-        if record.primary_post_id is None:
-            record.primary_post_id = post.id
-        # Keep the denormalized gallery sort key (alembic 0035) aligned with
-        # the primary post's publish date so /scroll orders off
-        # ix_image_record_effective_date instead of COALESCE-ing across the
-        # post join. Only override when THIS post is the primary AND carries
-        # a date; otherwise the column keeps its created_at-equivalent server
-        # default (matches the old COALESCE(post_date, created_at) fallback).
-        if record.primary_post_id == post.id and post.post_date is not None:
-            record.effective_date = post.post_date
-        # earliest_post_date (alembic 0071) = MIN(post_date) across ALL of this
-        # image's provenance posts, not just the primary — so the gallery can
-        # sort by original publish rather than the download/repost the primary
-        # points at. Recompute from provenance whenever a dated post is linked;
-        # the provenance row for THIS post was committed above, so the MIN
-        # includes it. Leaves the created_at default when no linked post is dated.
-        if post.post_date is not None:
-            earliest = self.session.execute(
-                select(func.min(Post.post_date))
-                .select_from(ImageProvenance)
-                .join(Post, Post.id == ImageProvenance.post_id)
-                .where(
-                    ImageProvenance.image_record_id == record.id,
-                    Post.post_date.is_not(None),
-                )
-            ).scalar_one_or_none()
-            if earliest is not None:
-                record.earliest_post_date = earliest
-        self.session.flush()
 
     def _copy_to_library(
         self, source: Path, sha: str, attribution_path: Path

backend/app/services/ingest_core.py

@@ -179,10 +179,11 @@ class Ingester:
         written: list[str] = []
         post_records: list[str] = []
         quarantined_paths: list[str] = []
-        # #830 recapture: (on-disk path, CDN source_url) pairs for already-present
+        # #830 recapture: (on-disk path, CDN source_url, post_id) triples for
-        # media, so phase 3 can backfill the ImageRecord's source_filehash WITHOUT
+        # already-present media, so phase 3 can (a) backfill the ImageRecord's
-        # re-downloading or unlinking the file. Empty outside recapture mode.
+        # source_filehash and (b) link the on-disk image to its Post (#1288) —
-        relink: list[tuple[str, str]] = []
+        # WITHOUT re-downloading or unlinking the file. Empty outside recapture.
+        relink: list[tuple[str, str, str]] = []
         downloaded = 0
         errors = 0
         quarantined = 0
@@ -410,12 +411,15 @@ class Ingester:
                         to_clear.append(key)
                         skipped_count += 1
                         consecutive_seen += 1
-                        # #830 recapture: surface (on-disk path, CDN url) so phase
+                        # #830/#1288 recapture: surface (on-disk path, CDN url,
-                        # 3 can backfill source_filehash for inline-image
+                        # post_id) so phase 3 can backfill source_filehash AND link
-                        # localization — a SEPARATE non-deleting channel, never the
+                        # the on-disk image to its Post — a SEPARATE non-deleting
-                        # import list (which would unlink the file, per above).
+                        # channel, never the import list (which would unlink the
+                        # file, per above).
                         if recapture and outcome.path is not None:
-                            relink.append((str(outcome.path), media_item.url))
+                            relink.append(
+                                (str(outcome.path), media_item.url, media_item.post_id)
+                            )
                     elif outcome.status == "skipped_seen":
                         skipped_count += 1
                         consecutive_seen += 1

backend/app/services/ml/heads.py

@@ -326,7 +326,10 @@ async def _current_heads(session: AsyncSession, embedding_version: str):
             {
                 "tag_id": r.tag_id,
                 "name": r.name,
-                "category": _CATEGORY.get(r.kind, "general"),
+                # System tags (wip/banner/editor) are kind=general but group under
+                # their OWN "system" suggestion category so the operator reviews
+                # them apart from content tags (they still train as general heads).
+                "category": "system" if r.is_system else _CATEGORY.get(r.kind, "general"),
                 "auto_apply_threshold": r.auto_apply_threshold,
                 "is_system": bool(r.is_system),
             }

backend/app/services/pixiv_client.py

@@ -435,24 +435,59 @@ class PixivClient:
             )
         ]
 
-    def _ugoira_media(self, work: dict, pid: str) -> list[MediaItem]:
+    def _ugoira_meta(self, work: dict, pid: str) -> dict | None:
-        """The ugoira frame zip (gallery-dl's default non-original mode):
+        """Fetch + memoize the ugoira metadata (frames + zip urls) for a work.
-        `/v1/ugoira/metadata` → zip_urls.medium with the 600x600→1920x1080
+
-        swap. Frame delays are memoized onto the work so the post record
+        Idempotent and cached on the work dict, so the post record and the
-        captures them (a future ugoira→video conversion needs the timings —
+        media extraction share ONE `/v1/ugoira/metadata` call regardless of
-        the zip alone has none). A metadata failure downgrades to 'no media'
+        which runs first (the core writes the post record BEFORE it extracts
-        with a warning (matching gallery-dl) instead of failing the walk —
+        media). Returns None — and caches the miss — on a non-auth failure
-        except auth failures, which stay loud."""
+        (matching gallery-dl's downgrade); auth failures stay loud."""
+        if "_ugoira_meta" in work:
+            return work["_ugoira_meta"]
         try:
             body = self._call("/v1/ugoira/metadata", {"illust_id": pid})
             meta = body["ugoira_metadata"]
-            zip_url = meta["zip_urls"]["medium"]
         except PixivAuthError:
             raise
         except (PixivAPIError, KeyError, TypeError) as exc:
             log.warning("Pixiv ugoira metadata failed for %s: %s", pid, exc)
-            return []
+            work["_ugoira_meta"] = None
+            return None
+        work["_ugoira_meta"] = meta
+        # Frame delays: a future ugoira→video conversion needs the timings (the
+        # zip alone has none), so the post record captures them.
         work["_ugoira_frames"] = meta.get("frames") or []
+        return meta
+
+    def fetch_ugoira_frames(self, post: dict) -> None:
+        """Populate `post['_work']['_ugoira_frames']` for an ugoira post (no-op
+        otherwise). The core writes the post record BEFORE extract_media, so
+        without this the frame timings would never reach the record; this
+        fetches (and memoizes, so extract_media reuses it) the metadata. Injected
+        into the downloader by the ingester, mirroring Patreon's content_fetcher.
+        Auth errors propagate; other failures leave frames unset."""
+        work = post.get("_work") or {}
+        if work.get("type") != "ugoira":
+            return
+        pid = str(post.get("id") or "")
+        if pid:
+            self._ugoira_meta(work, pid)
+
+    def _ugoira_media(self, work: dict, pid: str) -> list[MediaItem]:
+        """The ugoira frame zip (gallery-dl's default non-original mode):
+        `/v1/ugoira/metadata` → zip_urls.medium with the 600x600→1920x1080
+        swap. A metadata failure downgrades to 'no media' with a warning
+        (matching gallery-dl) instead of failing the walk — except auth
+        failures, which stay loud."""
+        meta = self._ugoira_meta(work, pid)
+        if meta is None:
+            return []
+        try:
+            zip_url = meta["zip_urls"]["medium"]
+        except (KeyError, TypeError) as exc:
+            log.warning("Pixiv ugoira zip url missing for %s: %s", pid, exc)
+            return []
         url = zip_url.replace("_ugoira600x600", "_ugoira1920x1080", 1)
         return [
             MediaItem(

backend/app/services/pixiv_downloader.py

@@ -102,11 +102,16 @@ class PixivDownloader(BaseNativeDownloader):
         validate: bool = True,
         rate_limit: float = 0.0,
         session: requests.Session | None = None,
+        ugoira_frames_fetcher: Callable[[dict], None] | None = None,
     ):
         super().__init__(
             images_root, cookies_path, platform="pixiv",
             validate=validate, rate_limit=rate_limit, session=session,
         )
+        # Injected by the ingester (client.fetch_ugoira_frames) so write_post_record
+        # can populate frame timings — which extract_media memoizes, but the core
+        # writes the post record FIRST. Mirrors Patreon's content_fetcher.
+        self._ugoira_frames_fetcher = ugoira_frames_fetcher
         if session is None:
             # i.pximg.net 403s any GET without the app Referer; mirror the
             # client's full app-header profile (gallery-dl serves media off
@@ -248,7 +253,16 @@ class PixivDownloader(BaseNativeDownloader):
                 "account": user.get("account"),
                 "name": user.get("name"),
             }
-        # Memoized by extract_media's ugoira branch; the zip has no timings.
+        # Ugoira frame timings. extract_media memoizes these, but the core writes
+        # the post record BEFORE extracting media, so fetch them here (shared +
+        # idempotent via the client's memoization) so the record actually keeps
+        # them — the zip carries no timings.
+        if (
+            work.get("type") == "ugoira"
+            and not work.get("_ugoira_frames")
+            and self._ugoira_frames_fetcher is not None
+        ):
+            self._ugoira_frames_fetcher(post)
         frames = work.get("_ugoira_frames")
         if frames:
             data["ugoira_frames"] = frames

backend/app/services/pixiv_ingester.py

@@ -81,6 +81,10 @@ class PixivIngester(Ingester):
             if downloader is not None
             else PixivDownloader(
                 self.images_root, cookies_path, validate=validate, rate_limit=rate_limit,
+                # write_post_record runs before extract_media in the core, so it
+                # fetches ugoira frame timings via the SAME client (shared,
+                # memoized) — else the record's ugoira_frames stays empty.
+                ugoira_frames_fetcher=resolved_client.fetch_ugoira_frames,
             )
         )
         super().__init__(

frontend/src/components/modal/SuggestionsPanel.vue

@@ -16,6 +16,16 @@
          of the rail (ImageViewer side layout), so a long suggestion set no
          longer needs an internal scroll cap to keep Related reachable. -->
     <div v-else class="fc-suggestions__list">
+      <!-- System hygiene tags (wip/banner/editor) surface in their own group,
+           first, so false positives are easy to spot and reject (reject-to-train
+           for these heads). Small set — collapsible, open by default. -->
+      <SuggestionsCategoryGroup
+        v-if="store.byCategory.system && store.byCategory.system.length"
+        label="System" :items="store.byCategory.system"
+        collapsible :default-open="true"
+        @accept="onAccept" @alias="onAlias" @remove-alias="onRemoveAlias"
+        @dismiss="onDismiss" @undismiss="onUndismiss"
+      />
       <SuggestionsCategoryGroup
         v-for="cat in peopleCats" :key="cat"
         v-show="store.byCategory[cat] && store.byCategory[cat].length"

frontend/src/components/modal/TagAutocomplete.vue

@@ -154,6 +154,9 @@ const KNOWN_KINDS = new Set([
 const KIND_ICONS = {
   general: 'mdi-tag', character: 'mdi-account-circle',
   fandom: 'mdi-book-open-page-variant', series: 'mdi-bookshelf',
+  // 'system' is a suggestion category (wip/banner/editor), not a tag kind —
+  // matches the shield marker on system-tag chips.
+  system: 'mdi-shield-outline',
 }
 function iconFor (k) { return KIND_ICONS[k] || 'mdi-tag' }
 

frontend/src/stores/suggestions.js

@@ -5,11 +5,13 @@ import { useApi } from '../composables/useApi.js'
 import { useAsyncAction } from '../composables/useAsyncAction.js'
 import { useInflightToken } from '../composables/useInflightToken.js'
 
-// Category display order: people first, general last.
+// Category display order: system (hygiene flags) first for quick review, then
-// 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired — only
+// people, general last. 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01)
-// character and general surface as suggestion categories now.
+// retired. 'system' groups the wip/banner/editor hygiene tags apart from content
-export const CATEGORY_ORDER = ['character', 'general']
+// tags (they're kind=general on the backend but surface under their own category).
+export const CATEGORY_ORDER = ['system', 'character', 'general']
 export const CATEGORY_LABELS = {
+  system: 'System',
   character: 'Character',
   general: 'General'
 }

tests/test_ml_suggestions.py

@@ -124,9 +124,11 @@ async def test_system_tag_surfaces_at_flat_floor_despite_high_threshold(db):
     await _head(db, tag.id, slot=0, suggest_threshold=0.9)
     await db.commit()
     sl = await SuggestionService(db).for_image(img.id)
+    # System tags surface under their OWN "system" category, not "general".
     assert any(
-        s.canonical_tag_id == tag.id for s in sl.by_category.get("general", [])
+        s.canonical_tag_id == tag.id for s in sl.by_category.get("system", [])
     )
+    assert not sl.by_category.get("general")
 
 
 @pytest.mark.asyncio
@@ -139,7 +141,7 @@ async def test_system_tag_below_floor_stays_hidden(db):
     await _head(db, tag.id, slot=0, suggest_threshold=0.9)
     await db.commit()
     sl = await SuggestionService(db).for_image(img.id)
-    assert not sl.by_category.get("general")
+    assert not sl.by_category.get("system")
 
 
 @pytest.mark.asyncio

tests/test_patreon_ingester.py

@@ -667,10 +667,12 @@ async def test_backfill_skips_already_captured_post_but_recapture_forces_it(
     assert downloader2.post_records == 1
     assert res_recap.files_downloaded == 0          # no media re-download
     assert downloader2.download_calls == 0
-    # The on-disk media is surfaced as a (path, CDN url) relink pair.
+    # The on-disk media is surfaced as a (path, CDN url, post_id) relink triple
+    # (post_id drives the #1288 image→post back-link in phase 3).
     assert len(res_recap.relink_source_paths) == 1
-    rel_path, rel_url = res_recap.relink_source_paths[0]
+    rel_path, rel_url, rel_post_id = res_recap.relink_source_paths[0]
     assert rel_url == m1.url
+    assert rel_post_id == m1.post_id
     # Diagnostic summary surfaces the post-record + relink counts (operator reads
     # this off the event stdout to see what a recapture actually did).
     assert "1 post-record(s), 1 relinked" in res_recap.stdout

tests/test_pixiv_client.py

@@ -200,6 +200,38 @@ def test_extract_media_ugoira_zip_swap(client, page1, monkeypatch):
     assert post["_work"]["_ugoira_frames"] == frames
 
 
+def test_fetch_ugoira_frames_memoizes_and_shares_one_call(client, page1, monkeypatch):
+    # fetch_ugoira_frames (called by write_post_record, which the core runs
+    # BEFORE extract_media) populates frames; extract_media then REUSES the
+    # memoized metadata — exactly one /v1/ugoira/metadata call total.
+    frames = [{"file": "000000.jpg", "delay": 90}]
+    calls = []
+
+    def fake_call(endpoint, params):
+        calls.append(endpoint)
+        return {"ugoira_metadata": {
+            "zip_urls": {"medium": (
+                "https://i.pximg.net/img-zip-ugoira/x/333_ugoira600x600.zip"
+            )},
+            "frames": frames,
+        }}
+
+    monkeypatch.setattr(client, "_call", fake_call)
+    post = _post_for(client, page1, 333)
+    client.fetch_ugoira_frames(post)
+    assert post["_work"]["_ugoira_frames"] == frames
+    items = client.extract_media(post, {})          # reuses memoized meta
+    assert items[0].media_id == "ugoira"
+    assert calls == ["/v1/ugoira/metadata"]          # ONE fetch, not two
+
+
+def test_fetch_ugoira_frames_noop_for_non_ugoira(client, page1, monkeypatch):
+    monkeypatch.setattr(client, "_call", lambda e, p: pytest.fail("should not fetch"))
+    post = _post_for(client, page1, 222)             # a single-page illust
+    client.fetch_ugoira_frames(post)
+    assert "_ugoira_frames" not in post["_work"]
+
+
 def test_extract_media_ugoira_metadata_failure_downgrades(
     client, page1, monkeypatch
 ):

tests/test_pixiv_downloader.py

@@ -268,6 +268,40 @@ def test_write_post_record_includes_ugoira_frames(tmp_path):
     assert data["ugoira_frames"] == frames
 
 
+def test_write_post_record_fetches_ugoira_frames_when_absent(tmp_path):
+    # The core writes the post record BEFORE extract_media, so frames aren't
+    # memoized yet — the injected fetcher must populate them so the record keeps
+    # the timings (regression: they were silently always empty).
+    post = _post_only(333)
+    assert "_ugoira_frames" not in post["_work"]
+    frames = [{"file": "000000.jpg", "delay": 120}]
+    calls = []
+
+    def fetcher(p):
+        calls.append(p)
+        p["_work"]["_ugoira_frames"] = frames
+
+    dl = PixivDownloader(
+        tmp_path, validate=False, session=_FakeSession(),
+        ugoira_frames_fetcher=fetcher,
+    )
+    outcome = dl.write_post_record(post, "artist-a")
+    data = json.loads(outcome.path.read_text())
+    assert data["ugoira_frames"] == frames
+    assert len(calls) == 1
+
+
+def test_write_post_record_non_ugoira_does_not_call_fetcher(tmp_path):
+    post = _post_only(111)                       # a plain multi-page illust
+    calls = []
+    dl = PixivDownloader(
+        tmp_path, validate=False, session=_FakeSession(),
+        ugoira_frames_fetcher=lambda p: calls.append(p),
+    )
+    dl.write_post_record(post, "artist-a")
+    assert calls == []
+
+
 def test_write_post_record_without_id(tmp_path):
     dl = _downloader(tmp_path)
     outcome = dl.write_post_record({"id": None, "attributes": {}}, "artist-a")

tests/test_sidecar_import.py

@@ -158,6 +158,60 @@ def test_relink_source_filehash_no_match_is_noop(importer, import_layout):
     ) is False
 
 
+def test_link_existing_image_to_post_backfills_provenance(importer, import_layout, db_sync):
+    """#1288: an on-disk image imported BARE (no sidecar → no post link, as the
+    pre-cutover gallery-dl pixiv images were) gets linked to its post from the
+    recapture walk's (path, external_post_id) pairing — image_provenance +
+    primary_post_id — without re-importing. Idempotent."""
+    import_root, _ = import_layout
+    m = import_root / "Alice" / "img.jpg"
+    _split(m, "v")
+    rec = importer.session.get(ImageRecord, importer.import_one(m).image_id)
+    assert rec.primary_post_id is None          # bare: no post link yet
+    artist = db_sync.get(Artist, rec.artist_id)
+    src = Source(
+        artist_id=artist.id, platform="pixiv",
+        url="https://www.pixiv.net/users/1", enabled=True,
+    )
+    db_sync.add(src)
+    db_sync.flush()
+    # The post already exists (upsert_post_record wrote it earlier in recapture).
+    post = Post(
+        source_id=src.id, artist_id=artist.id,
+        external_post_id="146132304", post_title="t",
+    )
+    db_sync.add(post)
+    db_sync.commit()
+
+    assert importer.link_existing_image_to_post(
+        Path(rec.path), "146132304", source=src, artist=artist,
+    ) is True
+    importer.session.refresh(rec)
+    assert rec.primary_post_id == post.id
+    prov = db_sync.execute(select(ImageProvenance).where(
+        ImageProvenance.image_record_id == rec.id,
+        ImageProvenance.post_id == post.id,
+    )).scalar_one()
+    assert prov.source_id == src.id
+
+    # Idempotent: a second call re-affirms without duplicating the row.
+    assert importer.link_existing_image_to_post(
+        Path(rec.path), "146132304", source=src, artist=artist,
+    ) is True
+    n = db_sync.execute(select(func.count(ImageProvenance.id)).where(
+        ImageProvenance.image_record_id == rec.id,
+        ImageProvenance.post_id == post.id,
+    )).scalar_one()
+    assert n == 1
+
+
+def test_link_existing_image_to_post_no_match_is_noop(importer):
+    """A path with no ImageRecord, or an empty external id, is a clean no-op."""
+    assert importer.link_existing_image_to_post(
+        Path("/nonexistent/x.jpg"), "999",
+    ) is False
+
+
 def test_reimport_same_post_idempotent(importer, import_layout):
     import_root, _ = import_layout
     # Threshold 0: only an exact phash match collapses. Orthogonal splits