Merge pull request 'Ship: title-based WIP auto-tagging (#1458)' (#222) from dev into main

alembic/versions/0085_wip_title_tagging.py

@@ -0,0 +1,35 @@
+"""title-based WIP auto-tagging (task #1458) — ImportSettings toggle
+
+ImportSettings gains wip_title_tagging_enabled (ON by default): when a freshly
+imported post's title explicitly declares work-in-progress ("WIP" / "work in
+progress"), the importer applies the `wip` system tag to its images. No new
+table — the tag itself is the seeded `wip` system tag (migration 0075) and the
+application reuses image_tag with source='wip_title'.
+
+Revision ID: 0085
+Revises: 0084
+Create Date: 2026-07-12
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0085"
+down_revision: Union[str, None] = "0084"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "import_settings",
+        sa.Column(
+            "wip_title_tagging_enabled", sa.Boolean(), nullable=False,
+            server_default=sa.text("true"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("import_settings", "wip_title_tagging_enabled")

backend/app/api/settings.py

@@ -48,6 +48,7 @@ _EDITABLE_FIELDS = (
     "interpreter_base_url",
     "translation_target_lang",
     "translation_min_confidence",
+    "wip_title_tagging_enabled",
 )
 
 # Per-host external-download toggles — all plain booleans, validated uniformly.
@@ -89,6 +90,7 @@ async def get_import_settings():
         "interpreter_base_url": row.interpreter_base_url,
         "translation_target_lang": row.translation_target_lang,
         "translation_min_confidence": row.translation_min_confidence,
+        "wip_title_tagging_enabled": row.wip_title_tagging_enabled,
     })
 
 
@@ -171,6 +173,12 @@ async def update_import_settings():
             return jsonify(
                 {"error": "series_suggest_threshold must be a number in [0, 1]"}
             ), 400
+    if "wip_title_tagging_enabled" in body and not isinstance(
+        body["wip_title_tagging_enabled"], bool
+    ):
+        return jsonify(
+            {"error": "wip_title_tagging_enabled must be a boolean"}
+        ), 400
 
     async with get_session() as session:
         row = await ImportSettings.load(session)
@@ -182,6 +190,18 @@ async def update_import_settings():
     return await get_import_settings()
 
 
+@settings_bp.route("/settings/wip-title/scan", methods=["POST"])
+async def wip_title_scan():
+    """Enqueue the back-catalogue WIP-title scan (task #1458 Settings button):
+    apply the `wip` system tag to EXISTING posts whose title declares
+    work-in-progress. New imports are tagged live by the importer; this catches
+    the existing library. Returns the Celery task id (202)."""
+    from ..tasks.maintenance import backfill_wip_title_tags
+
+    r = backfill_wip_title_tags.delay()
+    return jsonify({"celery_task_id": r.id}), 202
+
+
 @settings_bp.route("/system/stats", methods=["GET"])
 async def system_stats():
     async with get_session() as session:

backend/app/models/import_settings.py

@@ -116,6 +116,17 @@ class ImportSettings(Base):
         Float, nullable=False, default=0.9, server_default="0.9",
     )
 
+    # Title-based WIP auto-tagging (task #1458). When a freshly-imported post's
+    # TITLE explicitly declares work-in-progress ("WIP" / "work in progress"),
+    # the importer applies the `wip` system tag to its images — the artist's own
+    # label, used to keep unfinished pieces out of the Explore/gallery browse. ON
+    # by default (rule 26 — the feature works out of the box). Gates only the
+    # LIVE import hook; the existing catalogue is caught by the operator-triggered
+    # "Scan existing posts" backfill (which runs regardless of this flag).
+    wip_title_tagging_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, server_default="true",
+    )
+
     @classmethod
     async def load(cls, session) -> ImportSettings:
         """The singleton settings row (id=1), via an async session."""

backend/app/services/importer.py

@@ -47,9 +47,14 @@ from .attachment_store import AttachmentStore
 from .audits import single_color
 from .link_extract import extract_external_links
 from .thumbnailer import Thumbnailer
+from .wip_title import apply_wip_image_tags, matches_wip_title, resolve_wip_tag_id
 
 log = logging.getLogger(__name__)
 
+# Sentinel for the lazily-resolved wip tag id (distinguishes "not resolved yet"
+# from a genuine None = tag absent, so absence is cached and not re-queried).
+_UNSET = object()
+
 
 class SkipReason(StrEnum):
     too_small = "too_small"
@@ -183,6 +188,10 @@ class Importer:
         # invalidated mid-Importer (Importer instances are per-task /
         # per-archive-import so cross-instance staleness is harmless).
         self._phash_candidates: list[tuple] | None = None
+        # Lazily-resolved `wip` system tag id for title-based WIP auto-tagging
+        # (task #1458). Sentinel _UNSET so a genuine None (tag absent) is cached
+        # and not re-queried per media. Importer is per-task, so this can't stale.
+        self._wip_tag_id: int | None = _UNSET
 
     def _phash_candidates_cache(self) -> list[tuple]:
         """Cached `(phash, width, height, id)` rows from image_record.
@@ -933,6 +942,10 @@ class Importer:
         # Thumbnail is queued separately by the calling task; the importer
         # does not generate thumbnails inline so the import queue stays moving.
 
+        # Title-based WIP auto-tag (task #1458): fresh import only, after the
+        # sidecar has linked the post so record.primary_post_id / its title exist.
+        self._maybe_apply_wip_title(record)
+
         self.session.commit()
         return ImportResult(status="imported", image_id=record.id)
 
@@ -976,6 +989,36 @@ class Importer:
         self.session.commit()
         return ImportResult(status="refreshed", image_id=existing.id)
 
+    def _maybe_apply_wip_title(self, record: ImageRecord) -> None:
+        """Auto-apply the `wip` system tag to a FRESHLY-imported image when its
+        primary post's TITLE explicitly declares work-in-progress (task #1458 —
+        the artist's own "WIP" / "work in progress" label).
+
+        Called ONLY from the two new-record paths (never deep-scan / supersede),
+        so a manually-removed WIP tag is never re-applied by a routine re-scan —
+        removal sticks. The existing catalogue is covered separately by the
+        operator-triggered backfill sweep. Gated by the settings toggle, and
+        best-effort: any failure is logged, never allowed to fail the import."""
+        if not self.settings.wip_title_tagging_enabled:
+            return
+        if record.primary_post_id is None:
+            return
+        try:
+            title = self.session.execute(
+                select(Post.post_title).where(Post.id == record.primary_post_id)
+            ).scalar_one_or_none()
+            if not matches_wip_title(title):
+                return
+            if self._wip_tag_id is _UNSET:
+                self._wip_tag_id = resolve_wip_tag_id(self.session)
+            if self._wip_tag_id is None:
+                return
+            apply_wip_image_tags(self.session, [record.id], self._wip_tag_id)
+        except Exception as exc:  # noqa: BLE001 — a tag must never fail an import
+            log.warning(
+                "wip-title auto-tag failed for image %s: %s", record.id, exc
+            )
+
     def _apply_post_fields(self, post: Post, sd) -> None:
         """Write a parsed sidecar's post-level fields onto a Post — the SINGLE
         predicate shared by BOTH ingest paths: the per-media path (_apply_sidecar)
@@ -1253,6 +1296,10 @@ class Importer:
         # per-post Source row.
         self._apply_sidecar(record, path, artist, explicit_source=source)
 
+        # Title-based WIP auto-tag (task #1458): fresh import only, see the
+        # matching call in _import_media.
+        self._maybe_apply_wip_title(record)
+
         self.session.commit()
         return ImportResult(status="imported", image_id=record.id)
 

backend/app/services/wip_title.py

@@ -0,0 +1,101 @@
+"""Title-based WIP auto-tagging (task #1458).
+
+Deterministic heuristic: when a post's TITLE explicitly declares work-in-progress
+(the artist's own "WIP" / "work in progress" label), the ``wip`` system tag is
+applied to that post's images — a cheap, high-precision complement to the
+image-based ML ``wip`` head. WIP images are excluded from the Explore/gallery
+browse (see gallery_service ``excluded_system_tags``), so honouring the artist's
+own label keeps unfinished pieces out of the main browse right at import.
+
+Precision over recall — a false WIP tag HIDES a finished post — so matching is
+token-anchored: ``swipe`` / ``wiped`` / ``wiping`` never trip it (a letter on the
+boundary blocks the match).
+
+Sync-only: both consumers (the importer and the backfill Celery task) run on a
+sync Session. Application is idempotent-additive (ON CONFLICT DO NOTHING) and
+stamps a distinct ``image_tag.source`` so a later pass can tell where a wip tag
+came from — the "manual" / "head_auto" / "ccip_auto" / "ml_accepted" provenance
+family gains one member.
+"""
+import re
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.orm import Session
+
+from ..models.tag import WIP_SYSTEM_TAG, Tag, image_tag
+
+# image_tag.source stamped on title-heuristic WIP tags — distinct from the other
+# apply sources so provenance stays legible and a future undo can target only these.
+WIP_TITLE_SOURCE = "wip_title"
+
+# A standalone "WIP" / "W.I.P" token, or the phrase "work in progress"
+# (space/underscore/hyphen separated). The letter-boundary lookarounds are what
+# make this precision-first: `s|wip|e`, `|wip|ed`, `|wip|ing` all have a letter
+# abutting the token, so they're rejected. A trailing digit is allowed so
+# "WIP2" (= WIP part 2) still matches.
+_WIP_RE = re.compile(
+    r"(?<![A-Za-z])(?:w\.?i\.?p\.?|work[\s_-]+in[\s_-]+progress)(?![A-Za-z])",
+    re.IGNORECASE,
+)
+
+# Coarse SQL prefilter for the backfill sweep — narrows the post scan to rows that
+# COULD match before the precise regex confirms. Case-insensitive ILIKE patterns.
+# MUST stay a SUPERSET of _WIP_RE (every regex match contains "wip" or
+# "work…progress") or the sweep would silently miss posts.
+WIP_TITLE_SQL_PREFILTER = ("%wip%", "%work%progress%")
+
+# Chunk bulk inserts so a large sweep can't blow past psycopg's 65535-parameter
+# ceiling (3 params/row → ~21k rows max; 5k stays comfortably under).
+_INSERT_CHUNK = 5000
+
+
+def matches_wip_title(title: str | None) -> bool:
+    """True when a post title explicitly marks it work-in-progress."""
+    if not title:
+        return False
+    return _WIP_RE.search(title) is not None
+
+
+def resolve_wip_tag_id(session: Session) -> int | None:
+    """The seeded ``wip`` system tag's id (migration 0075), or None if absent."""
+    return session.execute(
+        select(Tag.id).where(Tag.name == WIP_SYSTEM_TAG, Tag.is_system.is_(True))
+    ).scalar_one_or_none()
+
+
+def apply_wip_image_tags(session: Session, image_ids, tag_id: int) -> int:
+    """Attach ``tag_id`` (source='wip_title') to each image id, idempotently —
+    never disturbs an existing tag or its source. Returns the number of image_tag
+    rows newly inserted. Does NOT commit.
+
+    The insert count is computed from a pre-SELECT of already-tagged ids rather
+    than the statement's ``rowcount``: psycopg reports -1 for a multi-row
+    ON CONFLICT DO NOTHING insert (it runs via an executemany path), so rowcount
+    is unusable here. The SELECT is accurate within this single transaction (no
+    concurrent writer touches these (image, wip) rows); ON CONFLICT DO NOTHING
+    stays as a race-safety belt so a rare concurrent insert can't error."""
+    ids = list({int(i) for i in image_ids})
+    if not ids:
+        return 0
+    inserted = 0
+    for start in range(0, len(ids), _INSERT_CHUNK):
+        chunk = ids[start:start + _INSERT_CHUNK]
+        already = set(session.execute(
+            select(image_tag.c.image_record_id)
+            .where(image_tag.c.tag_id == tag_id)
+            .where(image_tag.c.image_record_id.in_(chunk))
+        ).scalars())
+        to_insert = [iid for iid in chunk if iid not in already]
+        if not to_insert:
+            continue
+        session.execute(
+            pg_insert(image_tag)
+            .values([
+                {"image_record_id": iid, "tag_id": tag_id, "source": WIP_TITLE_SOURCE}
+                for iid in to_insert
+            ])
+            .on_conflict_do_nothing(index_elements=["image_record_id", "tag_id"])
+        )
+        inserted += len(to_insert)
+    return inserted

backend/app/tasks/maintenance.py

@@ -76,6 +76,8 @@ DOWNLOAD_STALL_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
 VERIFY_PAGE = 200
+# Title-based WIP backfill (task #1458): posts scanned per keyset page.
+WIP_BACKFILL_PAGE = 500
 FFPROBE_TIMEOUT_SECONDS = 10
 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600       # 24 h
 TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
@@ -1045,6 +1047,75 @@ def cleanup_old_download_events() -> int:
         return result.rowcount or 0
 
 
+@celery.task(
+    name="backend.app.tasks.maintenance.backfill_wip_title_tags",
+    # Coarse-prefiltered scan over posts; the candidate set is small on a typical
+    # library, but bound it like the other full-library sweeps.
+    soft_time_limit=1800, time_limit=2100,
+)
+def backfill_wip_title_tags() -> int:
+    """Scan EXISTING posts for explicit WIP titles and apply the `wip` system tag
+    to their images — the operator-triggered back-catalogue catch-up for
+    title-based WIP tagging (task #1458). New imports are tagged live by the
+    importer; this covers everything already in the library.
+
+    Keyset-paginated over posts (restart-safe). A coarse SQL prefilter narrows to
+    titles that COULD match; the precise regex (matches_wip_title) confirms.
+    Idempotent-additive (ON CONFLICT DO NOTHING) — never disturbs an existing tag.
+
+    Deliberately NOT scheduled as a beat: a periodic re-run would re-apply to
+    matching posts and silently undo a manual WIP removal, so it stays an explicit
+    operator action (Settings → "Scan existing posts for WIP titles"). Returns the
+    number of image-tag rows newly applied.
+    """
+    from ..models import Post
+    from ..models.image_provenance import ImageProvenance
+    from ..services.wip_title import (
+        WIP_TITLE_SQL_PREFILTER,
+        apply_wip_image_tags,
+        matches_wip_title,
+        resolve_wip_tag_id,
+    )
+
+    SessionLocal = _sync_session_factory()
+    applied = 0
+    last_id = 0
+    with SessionLocal() as session:
+        tag_id = resolve_wip_tag_id(session)
+        if tag_id is None:
+            log.warning(
+                "backfill_wip_title_tags: no `wip` system tag present; nothing to do"
+            )
+            return 0
+        like_a, like_b = WIP_TITLE_SQL_PREFILTER
+        while True:
+            rows = session.execute(
+                select(Post.id, Post.post_title)
+                .where(Post.id > last_id)
+                .where(Post.post_title.is_not(None))
+                .where(or_(
+                    Post.post_title.ilike(like_a),
+                    Post.post_title.ilike(like_b),
+                ))
+                .order_by(Post.id.asc())
+                .limit(WIP_BACKFILL_PAGE)
+            ).all()
+            if not rows:
+                break
+            last_id = rows[-1][0]
+            match_ids = [pid for pid, title in rows if matches_wip_title(title)]
+            if match_ids:
+                image_ids = session.execute(
+                    select(ImageProvenance.image_record_id)
+                    .where(ImageProvenance.post_id.in_(match_ids))
+                ).scalars().all()
+                applied += apply_wip_image_tags(session, image_ids, tag_id)
+                session.commit()
+    if applied:
+        log.info("backfill_wip_title_tags: applied wip to %d image(s)", applied)
+    return applied
+
+
 @celery.task(name="backend.app.tasks.maintenance.vacuum_analyze")
 def vacuum_analyze() -> dict:
     """Periodic VACUUM (ANALYZE) over the high-churn tables (VACUUM_TABLES) to

frontend/src/components/settings/ImportFiltersForm.vue

@@ -79,6 +79,35 @@
         </v-col>
       </v-row>
 
+      <v-divider class="my-5" />
+
+      <!-- Title-based WIP auto-tagging (task #1458). The switch gates the LIVE
+           import hook; the button runs the one-off back-catalogue scan (an
+           explicit action — it is deliberately not a scheduled sweep so it can't
+           silently re-apply a WIP tag you removed by hand). -->
+      <div class="fc-wip">
+        <div class="fc-wip__title">WIP auto-tagging</div>
+        <v-switch
+          v-model="local.wip_title_tagging_enabled"
+          label="Tag work-in-progress from post titles"
+          density="compact" hide-details color="primary" @change="save"
+        />
+        <div class="fc-help mb-3">
+          When a post's title says <strong>“WIP”</strong> or
+          <strong>“work in progress”</strong>, new imports get the
+          <code>wip</code> tag automatically — keeping unfinished pieces out of
+          the Explore browse. Applies to new imports; run the scan below to catch
+          posts already in your library.
+        </div>
+        <v-btn
+          variant="tonal" color="primary" size="small"
+          :loading="store.wipScanBusy" prepend-icon="mdi-magnify"
+          @click="store.scanWipTitles()"
+        >
+          Scan existing posts for WIP titles
+        </v-btn>
+      </div>
+
       <v-alert v-if="store.settingsError" type="error" variant="tonal" class="mt-2" closable>
         {{ store.settingsError }}
       </v-alert>
@@ -109,6 +138,7 @@ const local = reactive({
   skip_transparent: false, transparency_threshold: 0.9,
   skip_single_color: false, single_color_threshold: 0.95,
   phash_threshold: 10,
+  wip_title_tagging_enabled: true,
 })
 
 watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true })
@@ -124,11 +154,18 @@ async function save() {
   color: rgb(var(--v-theme-on-surface-variant));
   margin-top: 2px;
 }
-.fc-phash__title {
+.fc-phash__title,
+.fc-wip__title {
   font-size: 0.95rem;
   font-weight: 600;
   color: rgb(var(--v-theme-on-surface));
 }
+.fc-wip code {
+  font-size: 0.85em;
+  padding: 1px 4px;
+  border-radius: 3px;
+  background: rgb(var(--v-theme-surface-variant));
+}
 /* Headroom so the tick labels (Exact/Strict/Default/Loose) aren't clipped. */
 .fc-phash__slider { margin-bottom: 18px; }
 </style>

frontend/src/stores/import.js

@@ -16,6 +16,7 @@ export const useImportStore = defineStore('import', () => {
   const settings = ref(null)
   const settingsLoading = ref(false)
   const settingsError = ref(null)
+  const wipScanBusy = ref(false)
 
   async function loadSettings() {
     settingsLoading.value = true
@@ -40,8 +41,25 @@ export const useImportStore = defineStore('import', () => {
     }
   }
 
+  // Enqueue the back-catalogue WIP-title scan (task #1458). New imports are
+  // tagged live; this catches posts already in the library. Fire-and-forget —
+  // the sweep runs on the maintenance worker and its run shows in Activity.
+  async function scanWipTitles() {
+    wipScanBusy.value = true
+    try {
+      const r = await api.post('/api/settings/wip-title/scan')
+      toast({ text: 'Scanning existing posts for WIP titles…', type: 'success' })
+      return r
+    } catch (e) {
+      toast({ text: `WIP scan failed: ${e.message}`, type: 'error' })
+      throw e
+    } finally {
+      wipScanBusy.value = false
+    }
+  }
+
   return {
-    settings, settingsLoading, settingsError,
-    loadSettings, patchSettings,
+    settings, settingsLoading, settingsError, wipScanBusy,
+    loadSettings, patchSettings, scanWipTitles,
   }
 })

tests/test_wip_title.py

@@ -0,0 +1,49 @@
+"""Title-based WIP matcher (task #1458) — pure unit tests for
+``matches_wip_title``, the precision-first heuristic that decides whether a post
+title explicitly declares work-in-progress. No DB, no network.
+
+Precision matters more than recall here: a false positive applies the ``wip``
+system tag, which HIDES a finished post from the Explore browse — so the
+negative cases (substrings like ``swipe`` / ``wiped``) are the load-bearing ones.
+"""
+import pytest
+
+from backend.app.services.wip_title import matches_wip_title
+
+
+@pytest.mark.parametrize("title", [
+    "WIP",
+    "wip",
+    "WiP",
+    "Nami Heroines - WIP Part 1",   # the real-world example from the gate work
+    "sketch (WIP)",
+    "[WIP] new piece",
+    "WIP: colour test",
+    "cool art WIP2",                # trailing digit = "WIP part 2", still WIP
+    "W.I.P.",
+    "W.I.P",
+    "work in progress",
+    "Work In Progress",
+    "commission work-in-progress",
+    "big_project_work_in_progress",
+    "final touches, wip",
+])
+def test_matches_positive(title):
+    assert matches_wip_title(title) is True
+
+
+@pytest.mark.parametrize("title", [
+    None,
+    "",
+    "a quick swipe left",           # 'wip' inside swipe — must NOT match
+    "she wiped the counter",        # wiped
+    "wiping down the desk",         # wiping
+    "unwiped surface",
+    "progressive rock cover",       # 'progress' inside progressive, no 'work in'
+    "workin on it",                 # not the full phrase
+    "finished at last",
+    "Kawips diner",                 # 'wip' mid-word
+    "swipright",
+])
+def test_matches_negative(title):
+    assert matches_wip_title(title) is False

tests/test_wip_title_tagging.py

@@ -0,0 +1,106 @@
+"""Title-based WIP auto-tagging (task #1458) — integration tests for the apply
+helpers and the operator-triggered backfill sweep against a real DB. The
+importer's live hook is a thin call to these same tested pieces (matcher +
+apply), so the DB-facing behaviour is covered here.
+"""
+import pytest
+from sqlalchemy import select
+
+from backend.app.celery_app import celery
+from backend.app.models import Artist, ImageProvenance, ImageRecord, Post, Source
+from backend.app.models.tag import image_tag
+from backend.app.services.wip_title import apply_wip_image_tags, resolve_wip_tag_id
+
+pytestmark = pytest.mark.integration
+
+_N = 0
+
+
+@pytest.fixture(autouse=True)
+def eager():
+    celery.conf.task_always_eager = True
+    yield
+    celery.conf.task_always_eager = False
+
+
+def _img(db_sync):
+    global _N
+    _N += 1
+    rec = ImageRecord(
+        path=f"/images/w/{_N}.jpg", sha256=f"w{_N:063d}",
+        size_bytes=1, mime="image/jpeg", width=1, height=1,
+        origin="imported_filesystem", integrity_status="unknown",
+    )
+    db_sync.add(rec)
+    db_sync.flush()
+    return rec
+
+
+def _post(db_sync, *, title, slug, ext):
+    a = Artist(name=slug.upper(), slug=slug)
+    db_sync.add(a)
+    db_sync.flush()
+    s = Source(artist_id=a.id, platform="patreon", url=f"https://patreon.test/{slug}")
+    db_sync.add(s)
+    db_sync.flush()
+    p = Post(source_id=s.id, artist_id=a.id, external_post_id=ext, post_title=title)
+    db_sync.add(p)
+    db_sync.flush()
+    return s, p
+
+
+def _link(db_sync, rec, post, source):
+    db_sync.add(ImageProvenance(
+        image_record_id=rec.id, post_id=post.id, source_id=source.id,
+    ))
+    db_sync.flush()
+
+
+def _wip_source(db_sync, image_id, tag_id):
+    return db_sync.execute(
+        select(image_tag.c.source).where(
+            image_tag.c.image_record_id == image_id,
+            image_tag.c.tag_id == tag_id,
+        )
+    ).scalar_one_or_none()
+
+
+def test_resolve_wip_tag_id_present(db_sync):
+    # The `wip` system tag is seeded by migration 0075 and restored between tests.
+    assert resolve_wip_tag_id(db_sync) is not None
+
+
+def test_apply_is_idempotent_and_stamps_source(db_sync):
+    tag_id = resolve_wip_tag_id(db_sync)
+    rec = _img(db_sync)
+    db_sync.commit()
+
+    assert apply_wip_image_tags(db_sync, [rec.id], tag_id) == 1
+    # Second apply is a no-op (ON CONFLICT DO NOTHING) — never disturbs the row.
+    assert apply_wip_image_tags(db_sync, [rec.id], tag_id) == 0
+    assert _wip_source(db_sync, rec.id, tag_id) == "wip_title"
+
+
+def test_backfill_tags_only_wip_titled_posts(db_sync):
+    from backend.app.tasks.maintenance import backfill_wip_title_tags
+
+    tag_id = resolve_wip_tag_id(db_sync)
+    wip_rec = _img(db_sync)
+    plain_rec = _img(db_sync)
+    swipe_rec = _img(db_sync)
+    s1, wip_post = _post(db_sync, title="Cool piece - WIP Part 1", slug="wipy", ext="1")
+    s2, plain_post = _post(db_sync, title="Finished commission", slug="doney", ext="2")
+    s3, swipe_post = _post(db_sync, title="a quick swipe", slug="swipey", ext="3")
+    _link(db_sync, wip_rec, wip_post, s1)
+    _link(db_sync, plain_rec, plain_post, s2)
+    _link(db_sync, swipe_rec, swipe_post, s3)   # 'swipe' must NOT trip the matcher
+    db_sync.commit()
+
+    assert backfill_wip_title_tags.apply().get() == 1
+
+    assert _wip_source(db_sync, wip_rec.id, tag_id) == "wip_title"
+    assert _wip_source(db_sync, plain_rec.id, tag_id) is None
+    assert _wip_source(db_sync, swipe_rec.id, tag_id) is None
+
+    # Idempotent: a second sweep finds the tag already present and applies nothing.
+    assert backfill_wip_title_tags.apply().get() == 0