feat(translation): 8h sweep + drain button + detection probe (#1376)

Throughput: translate_posts now runs every 8h (was daily) as the
steady-state cadence for newly-imported posts, and the Settings
"Translate now" button runs it in drain mode (run-until-done, no reset)
so one press clears the whole untranslated backlog instead of a single
300-post chunk. The interrupt/backoff re-enqueue now preserves the drain
flag so a bulk drain resumes cleanly after an Interpreter restart.

Misdetection groundwork: surface the detector's confidence from the
Interpreter client (it was in the detectedLanguage payload but discarded)
and add a read-only "Test translation" box — POST /settings/translation/
probe + TranslationCard UI — that shows detected language + confidence +
engine + result for pasted text, without saving. Lets the operator see
why a short/abbreviation-heavy English title gets mis-detected so the
detection guard (min-length + confidence floor) can be tuned from real
numbers. The guard itself follows once the mis-detected cases are probed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CgZP9v2otxVJymiYsnVuMy

backend/app/api/settings.py

@@ -364,9 +364,49 @@ async def translation_test():
     return jsonify({"healthy": healthy})
 
 
+@settings_bp.route("/settings/translation/probe", methods=["POST"])
+async def translation_probe():
+    """Diagnostic for the Settings 'Test translation' box: translate a pasted
+    snippet WITHOUT saving anything, returning what Interpreter *detected*
+    (language + confidence) alongside the result. Lets the operator see why a
+    given string was (mis-)detected — e.g. a short English title flagged as
+    another language — so a detection guard can be tuned from real numbers.
+    Read-only: no post is touched. Uses the currently-saved base URL + target."""
+    body = await request.get_json(silent=True)
+    body = body if isinstance(body, dict) else {}
+    text = (body.get("text") or "").strip()
+    if not text:
+        return jsonify({"error": "provide text to translate"}), 400
+    async with get_session() as session:
+        cfg = await ImportSettings.load(session)
+    base_url = (cfg.interpreter_base_url or "").strip()
+    if not base_url:
+        return jsonify({"error": "no Interpreter base URL is set"}), 400
+    target = (cfg.translation_target_lang or "en").strip() or "en"
+    try:
+        res = await asyncio.to_thread(
+            ic.translate, [text], base_url=base_url, target=target,
+        )
+    except ic.InterpreterUnavailable as e:
+        return jsonify({"error": f"Interpreter unavailable: {e}"}), 503
+    except ic.InterpreterBadRequest as e:
+        return jsonify({"error": f"Interpreter rejected the request: {e}"}), 400
+    translations = res.get("translations") or []
+    return jsonify({
+        "target": target,
+        "detected_lang": res.get("detected_lang"),
+        "detected_confidence": res.get("detected_confidence"),
+        "engine": res.get("engine"),
+        "engine_version": res.get("engine_version"),
+        "translated": translations[0] if translations else None,
+    })
+
+
 @settings_bp.route("/settings/translation/run", methods=["POST"])
 async def translation_run():
-    """Enqueue the translate sweep now (the Settings 'Translate now' button)."""
+    """Enqueue the translate sweep now (the Settings 'Translate now' button).
+    Runs in drain mode — run-until-done — so one press chases the whole
+    untranslated backlog to zero rather than a single 300-post chunk."""
     async with get_session() as session:
         cfg = await ImportSettings.load(session)
     if not cfg.translation_enabled or not (cfg.interpreter_base_url or "").strip():
@@ -375,7 +415,7 @@ async def translation_run():
         ), 400
     from ..tasks.translation import translate_posts
 
-    r = translate_posts.delay()
+    r = translate_posts.delay(drain=True)
     return jsonify({"celery_task_id": r.id}), 202
 
 

backend/app/celery_app.py

@@ -178,9 +178,13 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.ml.prune_presentation_reviews",
                 "schedule": 86400.0,  # retention: drop resolved review flags >30d
             },
-            "translate-posts-daily": {
+            "translate-posts-8h": {
                 "task": "backend.app.tasks.translation.translate_posts",
-                "schedule": 86400.0,  # no-op unless translation configured + healthy
+                "schedule": 28800.0,  # every 8h: steady-state cadence for the
+                # trickle of newly-imported posts (no-op unless translation
+                # configured + healthy). One bounded 300-chunk per fire — the
+                # one-time backlog drains via the "Translate now" button
+                # (drain=True, run-until-done), not this sweep.
             },
             "snapshot-head-metrics-daily": {
                 "task": "backend.app.tasks.maintenance.snapshot_head_metrics",

backend/app/services/interpreter_client.py

@@ -96,8 +96,9 @@ def translate(
 ) -> dict:
     """Translate a batch. Returns::
 
-        {"translations": [str, ...],   # SAME order & length as `texts`
-         "detected_lang": str | None,  # aggregate: describes the FIRST item
+        {"translations": [str, ...],        # SAME order & length as `texts`
+         "detected_lang": str | None,       # aggregate: describes the FIRST item
+         "detected_confidence": float | None,  # detector's confidence, if given
          "engine": str | None,
          "engine_version": str | None}
 
@@ -110,6 +111,7 @@ def translate(
     """
     if not texts:
         return {"translations": [], "detected_lang": None,
+                "detected_confidence": None,
                 "engine": None, "engine_version": None}
     try:
         r = session.post(
@@ -140,9 +142,11 @@ def translate(
     if not isinstance(out, list):
         out = [out]
     interp = body.get("interpreter") or {}
+    detected = body.get("detectedLanguage") or {}
     return {
         "translations": out,
-        "detected_lang": (body.get("detectedLanguage") or {}).get("language"),
+        "detected_lang": detected.get("language"),
+        "detected_confidence": detected.get("confidence"),
         "engine": interp.get("engine"),
         "engine_version": interp.get("engine_version"),
     }

backend/app/tasks/translation.py

@@ -7,8 +7,10 @@ same pattern as the other backfill sweeps. No-op unless translation is enabled,
 a base URL is set, and the service is healthy.
 
 Two entry points:
-- ``translate_posts`` — the daily untranslated sweep (translated_source_lang IS
-  NULL only); also the Settings "Translate now" button.
+- ``translate_posts`` — the periodic untranslated sweep (every 8h;
+  translated_source_lang IS NULL only), one bounded chunk per fire. The Settings
+  "Translate now" button calls it with ``drain=True`` to chase the tail
+  run-until-done and clear the whole backlog in a single press.
 - ``retranslate_posts`` — after a model change, resets the stored translation
   columns for a scoped set of posts (all, or a given set of artists) so the
   untranslated selection re-runs them, then chases the tail until drained
@@ -56,13 +58,19 @@ def _interrupt_backoff(retry_after) -> int:
     name="backend.app.tasks.translation.translate_posts",
     soft_time_limit=1800, time_limit=2100,
 )
-def translate_posts() -> str:
+def translate_posts(drain: bool = False) -> str:
     """Translate untranslated non-English post title/description (default target
     en) via Interpreter. Per-post [title, description] batch → per-post detected
     language. Passthrough / already-target posts are marked handled (source ==
     target) with no stored translation. 503 or a connection error interrupts the
     run (retry next cycle); 400 stops it (fix config); posts done so far stay
-    committed. No-op unless configured + healthy. Returns a summary string."""
+    committed. No-op unless configured + healthy. Returns a summary string.
+
+    ``drain`` (the Settings "Translate now" button) chases the tail
+    run-until-done: on a clean chunk with work still remaining it re-enqueues
+    itself until the untranslated backlog is zero, like ``retranslate_posts``. The
+    periodic beat leaves it False → one bounded chunk per fire (the 8h cadence
+    drives the rest, enough for the trickle of newly-imported posts)."""
     SessionLocal = _sync_session_factory()
     with SessionLocal() as session:
         ready = _translation_config(session)
@@ -73,14 +81,28 @@ def translate_posts() -> str:
         status, translated, retry_after = _translate_batch(
             session, posts, base_url, target,
         )
-        # Steady-state daily sweep: one chunk per run on success (the beat drives
-        # the rest — no tail-chase). But if Interpreter drained mid-chunk, don't
-        # make a manual "Translate now" wait a whole day — re-enqueue after its
-        # Retry-After hint (or the default backoff). Self-terminating: once the
-        # service is down the health gate returns "interpreter unavailable" early.
-        if status == "interrupted" and _count_untranslated(session, None):
+        # Drain mode (manual "Translate now"): chase the tail until the backlog is
+        # zero, so one press clears the whole pile instead of one 300-chunk.
+        # Termination is guaranteed — every handled post leaves the untranslated
+        # set, so the remaining count strictly decreases each chunk.
+        if status == "ok" and drain:
+            remaining = _count_untranslated(session, None)
+            if remaining:
+                translate_posts.apply_async(
+                    (), {"drain": True}, countdown=_RETRANSLATE_COUNTDOWN,
+                )
+                log.info(
+                    "translate_posts: draining — %d remaining → re-enqueued",
+                    remaining,
+                )
+        # Interpreter drained mid-chunk (graceful restart): don't make a manual
+        # "Translate now" wait a whole beat cycle — re-enqueue after its
+        # Retry-After hint (or the default backoff), preserving `drain` so a bulk
+        # drain resumes cleanly. Self-terminating: once the service is down the
+        # health gate returns "interpreter unavailable" early.
+        elif status == "interrupted" and _count_untranslated(session, None):
             delay = _interrupt_backoff(retry_after)
-            translate_posts.apply_async((), {}, countdown=delay)
+            translate_posts.apply_async((), {"drain": drain}, countdown=delay)
             log.info(
                 "translate_posts: interrupted (service draining) → retry in %ss",
                 delay,