feat(wip): soft title tier — sketch/doodle vocab + ring-loud audit (#1474)

Extends WIP title-tagging to lower-precision cues (sketch/doodle/scribble) safely.

- wip_title.py: soft matcher (word-anchored; sketchbook/kadoodle don't trip it);
  WIP_TITLE_SOFT_SOURCE + soft SQL prefilter; apply_wip_image_tags takes a source arg.
- training_data._AUTO_SOURCES += 'wip_title_soft' → the soft tier is PROVISIONAL and
  never trains the wip head (a finished "sketch" can't pollute it). Only the hard
  tier (wip_title) + manual train.
- ImportSettings.wip_soft_title_tagging_enabled (OFF by default, opt-in). Migration 0087.
- importer: hard tier wins, soft is the fallback (source wip_title_soft).
- backfill: refactored into a shared _backfill_wip_tier; hard always, soft when enabled.
- heads.soft_wip_conflict_audit + daily beat: score soft-tagged images against content
  heads, flag ring-loud ones (PresentationReview mode=process) for the review strip —
  the operator's "measure if they got falsely tagged" safety.
- api settings toggle; ImportFiltersForm soft toggle.
- tests: soft matcher pos/neg; soft source not a training positive; audit flags
  ring-loud + spares quiet.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/tasks/maintenance.py

@@ -1047,6 +1047,41 @@ def cleanup_old_download_events() -> int:
         return result.rowcount or 0
 
 
+def _backfill_wip_tier(session, tag_id, prefilter, matcher, source) -> int:
+    """One keyset-paginated pass over posts whose title matches a WIP tier, applying
+    `tag_id` (stamped `source`) to their images. Shared by the hard + soft tiers
+    (#1458 / #1474). Coarse `prefilter` (ILIKE superset) narrows the scan; the precise
+    `matcher` confirms. Idempotent-additive (ON CONFLICT DO NOTHING). Returns the row
+    count newly applied."""
+    from ..models import Post
+    from ..models.image_provenance import ImageProvenance
+    from ..services.wip_title import apply_wip_image_tags
+
+    applied = 0
+    last_id = 0
+    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(p) for p in prefilter]))
+            .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 matcher(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, source=source)
+            session.commit()
+    return applied
+
+
 @celery.task(
     name="backend.app.tasks.maintenance.backfill_wip_title_tags",
     # Coarse-prefiltered scan over posts; the candidate set is small on a typical
@@ -1054,32 +1089,32 @@ def cleanup_old_download_events() -> int:
     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.
+    """Scan EXISTING posts for WIP titles and apply the `wip` system tag to their
+    images — the operator-triggered back-catalogue catch-up (task #1458 hard tier +
+    #1474 soft tier). New imports are tagged live by the importer; this covers the
+    existing 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.
+    HARD tier ("WIP"/"work in progress") always runs (the operator triggered the
+    scan); the SOFT tier (sketch/doodle, provisional source) runs only when
+    wip_soft_title_tagging_enabled, AFTER hard so a title matching both keeps the
+    trained hard tag (ON CONFLICT DO NOTHING). Keyset-paginated, restart-safe.
 
-    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.
+    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 rows applied.
     """
-    from ..models import Post
-    from ..models.image_provenance import ImageProvenance
+    from ..models import ImportSettings
     from ..services.wip_title import (
+        SOFT_WIP_TITLE_SQL_PREFILTER,
+        WIP_TITLE_SOFT_SOURCE,
+        WIP_TITLE_SOURCE,
         WIP_TITLE_SQL_PREFILTER,
-        apply_wip_image_tags,
+        matches_soft_wip_title,
         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:
@@ -1087,30 +1122,16 @@ def backfill_wip_title_tags() -> int:
                 "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()
+        settings = ImportSettings.load_sync(session)
+        applied = _backfill_wip_tier(
+            session, tag_id, WIP_TITLE_SQL_PREFILTER, matches_wip_title,
+            WIP_TITLE_SOURCE,
+        )
+        if settings.wip_soft_title_tagging_enabled:
+            applied += _backfill_wip_tier(
+                session, tag_id, SOFT_WIP_TITLE_SQL_PREFILTER, matches_soft_wip_title,
+                WIP_TITLE_SOFT_SOURCE,
+            )
     if applied:
         log.info("backfill_wip_title_tags: applied wip to %d image(s)", applied)
     return applied

backend/app/tasks/ml.py

@@ -629,6 +629,24 @@ def scheduled_process_auto_apply() -> str:
     return f"applied={result['n_applied']} flagged={result['n_flagged']}"
 
 
+@celery.task(
+    name="backend.app.tasks.ml.scheduled_soft_wip_conflict_audit",
+    soft_time_limit=1800, time_limit=2100,
+)
+def scheduled_soft_wip_conflict_audit() -> str:
+    """Ring-loud audit over the SOFT WIP-title cohort (#1474) — flag sketch/doodle
+    auto-tags that ALSO look like real content for review. No-op when there are no
+    content heads; idempotent (already-flagged images skipped). Runs regardless of
+    the process-sweep toggle, since soft-title tags come from the importer, not that
+    sweep. Wall-clock bounded by the task time limits."""
+    from ..services.ml.heads import soft_wip_conflict_audit
+
+    SessionLocal = _sync_session_factory()
+    with SessionLocal() as session:
+        result = soft_wip_conflict_audit(session)
+    return f"scanned={result['n_scanned']} flagged={result['n_flagged']}"
+
+
 @celery.task(name="backend.app.tasks.ml.prune_presentation_reviews")
 def prune_presentation_reviews() -> str:
     """Retention (rule 89): drop RESOLVED presentation-review flags older than 30