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,
@@ -250,29 +272,39 @@ def _summary(status: str, translated: int, scanned: int) -> str:
     return f"translated={translated} scanned={scanned}"
 
 
+def _translate_field(text: str, base_url: str, target: str):
+    """Translate ONE field independently. Returns (translated, source_lang,
+    engine_version), or (None, None, None) when there's nothing to store — empty
+    text, already the target language, or a passthrough (engine "none"). Per-field
+    (not aggregate first-item) detection so a non-English description still gets
+    translated when the title is already English (mixed-language posts)."""
+    if not text:
+        return None, None, None
+    res = ic.translate([text], base_url=base_url, target=target)
+    detected = res["detected_lang"] or target
+    if detected == target or res["engine"] == "none":
+        return None, None, None
+    return res["translations"][0], detected, res["engine_version"]
+
+
 def _translate_one(session, post, base_url: str, target: str) -> int:
-    """Translate one post's title/description in place. Returns 1 if it stored a
-    translation, 0 for passthrough/empty (which still marks the post handled)."""
+    """Translate a post's title/description in place, each field independently.
+    Returns 1 if it stored at least one translation, 0 when the post is all
+    passthrough/empty (still marks it handled so the sweep won't revisit it)."""
     title = (post.post_title or "").strip()
     desc = (html_to_plain(post.description) if post.description else "") or ""
     desc = desc.strip()
-    fields: list[tuple[str, str]] = []
-    if title:
-        fields.append(("title", title))
-    if desc:
-        fields.append(("desc", desc))
-    if not fields:
-        post.translated_source_lang = target   # nothing to translate → handled
+    title_tr, title_lang, title_ev = _translate_field(title, base_url, target)
+    desc_tr, desc_lang, desc_ev = _translate_field(desc, base_url, target)
+    if title_tr is None and desc_tr is None:
+        # Nothing non-target (or nothing to translate) → handled, store nothing.
+        post.translated_source_lang = target
         return 0
-    res = ic.translate([t for _, t in fields], base_url=base_url, target=target)
-    detected = res["detected_lang"] or target
-    if detected == target or res["engine"] == "none":
-        post.translated_source_lang = target   # already target language → handled
-        return 0
-    by_field = {f: res["translations"][i] for i, (f, _) in enumerate(fields)}
-    post.post_title_translated = by_field.get("title")
-    post.description_translated = by_field.get("desc")
-    post.translated_source_lang = detected
-    post.translation_engine_version = res["engine_version"]
+    post.post_title_translated = title_tr
+    post.description_translated = desc_tr
+    # Source lang = whichever field was actually non-target (for a mixed post,
+    # the translated field's language, not the already-English one).
+    post.translated_source_lang = title_lang or desc_lang
+    post.translation_engine_version = title_ev or desc_ev
     post.translated_at = datetime.now(UTC)
     return 1

frontend/src/components/modal/TagChip.vue

@@ -143,7 +143,19 @@ function iconFor (k) { return KIND_ICONS[k] || 'mdi-tag' }
    chip's hover title. */
 .fc-tag-chip { display: inline-flex; align-items: center; gap: 1px; max-width: 100%; min-width: 0; }
 .fc-tag-chip__nav { max-width: 100%; min-width: 0; }
+/* Tonal chips (esp. character = info) wash out against the dark rail — the fill
+   is intentionally faint. Give every tag chip a thin border in its OWN kind
+   colour (currentColor = the tonal chip's themed foreground) so the edge reads
+   clearly without touching the fill (operator-asked 2026-07-08). Theme-aware:
+   currentColor tracks the kind colour in either light or dark. */
+.fc-tag-chip__nav { border: thin solid color-mix(in srgb, currentColor 55%, transparent); }
 .fc-tag-chip__nav :deep(.v-chip__content) { min-width: 0; overflow: hidden; }
+/* The bigger size=default chip widens Vuetify's negative start-margin on the
+   leading kind-icon, pulling it LEFT of the content box above — where that
+   overflow:hidden then clips the icon's left edge as a vertical slice
+   (operator-flagged 2026-07-08). Zero the pull so the icon sits inside the clip
+   box; the chip's own 12px padding keeps a comfortable left inset. */
+.fc-tag-chip__nav :deep(.v-icon--start) { margin-inline-start: 0; }
 .fc-tag-chip__name {
   overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
   min-width: 0; flex: 0 1 auto;

frontend/src/components/settings/TranslationCard.vue

@@ -78,6 +78,61 @@
       artist, use the Re-translate button on that artist's page.
     </p>
 
+    <v-divider class="my-3" />
+
+    <!-- Diagnostic probe (task #1376): paste text → what Interpreter DETECTS
+         (language + confidence) and returns, without saving. Used to see why a
+         short/abbreviation-heavy English title gets mis-detected, and to pick a
+         detection-confidence / min-length guard from real numbers. -->
+    <div class="mb-1">
+      <span class="fc-section-h">Test translation</span>
+      <p class="fc-muted text-caption mt-1 mb-2">
+        Paste a title or snippet to see what Interpreter detects and returns —
+        handy for checking why a short string is mis-detected. Nothing is saved.
+      </p>
+    </div>
+    <v-textarea
+      v-model="probeText" label="Text to test" rows="2" auto-grow
+      density="compact" hide-details class="mb-2" :disabled="!baseUrl"
+    />
+    <div class="d-flex align-center flex-wrap mb-2" style="gap: 8px;">
+      <v-btn
+        size="small" variant="tonal" rounded="pill" prepend-icon="mdi-magnify"
+        :loading="probing" :disabled="!baseUrl || !probeText.trim()"
+        @click="onProbe"
+      >Test translation</v-btn>
+    </div>
+    <v-sheet
+      v-if="probeResult" rounded class="pa-3 mb-1 text-body-2"
+      color="rgba(var(--v-theme-accent), 0.06)"
+    >
+      <div class="d-flex flex-wrap" style="gap: 4px 16px;">
+        <span>
+          <span class="fc-muted">Detected:</span>
+          <strong>{{ probeResult.detected_lang || '—' }}</strong>
+        </span>
+        <span v-if="probeResult.detected_confidence != null">
+          <span class="fc-muted">Confidence:</span>
+          <strong>{{ fmtConf(probeResult.detected_confidence) }}</strong>
+        </span>
+        <span>
+          <span class="fc-muted">Engine:</span>
+          {{ probeResult.engine || '—' }}<template
+            v-if="probeResult.engine_version"
+          > ({{ probeResult.engine_version }})</template>
+        </span>
+        <span><span class="fc-muted">Target:</span> {{ probeResult.target }}</span>
+      </div>
+      <v-divider class="my-2" />
+      <div class="fc-muted mb-1">
+        Result<template v-if="probeResult.detected_lang === probeResult.target">
+          — already {{ probeResult.target }}, so the sweep leaves it as-is</template>:
+      </div>
+      <p class="mb-0" style="white-space: pre-wrap;">{{
+        probeResult.translated || '(no change)'
+      }}</p>
+    </v-sheet>
+
     <v-alert
       v-if="err" type="error" variant="tonal" density="compact"
       class="mt-3" closable @click:close="err = null"
@@ -129,6 +184,17 @@ const testResult = ref(null)   // null = not tested this session; true/false = l
 const status = ref(null)
 const err = ref(null)
 
+// #1376 "Test translation" probe: paste text → detected lang/confidence + result.
+const probeText = ref('')
+const probing = ref(false)
+const probeResult = ref(null)
+
+// Confidence scale is Interpreter-defined (0–100 or 0–1) — show it as-is, just
+// trimmed to 2 decimals, so the operator reads the true number.
+function fmtConf (c) {
+  return c == null ? '' : Math.round(c * 100) / 100
+}
+
 const statusColor = computed(() => {
   if (!enabled.value || !baseUrl.value) return 'grey'
   return status.value?.healthy ? 'success' : 'error'
@@ -170,6 +236,21 @@ async function onTest() {
   }
 }
 
+async function onProbe () {
+  probing.value = true
+  err.value = null
+  probeResult.value = null
+  try {
+    probeResult.value = await api.post('/api/settings/translation/probe', {
+      body: { text: probeText.value.trim() },
+    })
+  } catch (e) {
+    err.value = e.message
+  } finally {
+    probing.value = false
+  }
+}
+
 onMounted(async () => {
   try {
     await store.loadSettings()

tests/test_api_settings.py

@@ -113,9 +113,10 @@ async def test_translation_test_connection(client, monkeypatch):
 
 @pytest.mark.asyncio
 async def test_translation_run_enqueues_when_configured(client, monkeypatch):
+    sink = {}
     monkeypatch.setattr(
         "backend.app.tasks.translation.translate_posts.delay",
-        lambda *a, **k: type("R", (), {"id": "t1"})(),
+        lambda *a, **k: sink.update(k) or type("R", (), {"id": "t1"})(),
     )
     await client.patch("/api/settings/import", json={
         "translation_enabled": True, "interpreter_base_url": "http://i.lan",
@@ -123,6 +124,46 @@ async def test_translation_run_enqueues_when_configured(client, monkeypatch):
     resp = await client.post("/api/settings/translation/run")
     assert resp.status_code == 202
     assert (await resp.get_json())["celery_task_id"] == "t1"
+    assert sink["drain"] is True   # "Translate now" chases the whole backlog
+
+
+@pytest.mark.asyncio
+async def test_translation_probe_requires_text(client):
+    resp = await client.post(
+        "/api/settings/translation/probe", json={"text": "  "})
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_translation_probe_requires_base_url(client):
+    # Text given but no Interpreter URL saved → nothing to probe against.
+    resp = await client.post(
+        "/api/settings/translation/probe", json={"text": "hello"})
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_translation_probe_returns_detection(client, monkeypatch):
+    # The diagnostic surfaces detected language + confidence + result, unsaved.
+    monkeypatch.setattr(
+        "backend.app.api.settings.ic.translate",
+        lambda texts, **k: {
+            "translations": ["Work in Progress"],
+            "detected_lang": "de", "detected_confidence": 71.5,
+            "engine": "llm", "engine_version": "v1",
+        },
+    )
+    await client.patch("/api/settings/import", json={
+        "translation_enabled": True, "interpreter_base_url": "http://i.lan",
+    })
+    resp = await client.post(
+        "/api/settings/translation/probe", json={"text": "WIP"})
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert body["detected_lang"] == "de"
+    assert body["detected_confidence"] == 71.5
+    assert body["translated"] == "Work in Progress"
+    assert body["target"] == "en"
 
 
 def _fake_retranslate(monkeypatch, sink):

tests/test_interpreter_client.py

@@ -34,6 +34,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
     out = ic.translate(["ねこが可愛い", "金髪ギャル！"], base_url="http://i.lan")
     assert out["translations"] == ["The cat is cute", "Blonde gal!"]
     assert out["detected_lang"] == "ja"
+    assert out["detected_confidence"] == 0.98   # surfaced for the detection guard
     assert out["engine_version"] == "ollama:x:12b"
     assert captured["json"]["q"] == ["ねこが可愛い", "金髪ギャル！"]
     assert captured["json"]["target"] == "en"

tests/test_translate_posts.py

@@ -5,7 +5,11 @@ from sqlalchemy import select
 
 from backend.app.models import Artist, ImportSettings, Post
 from backend.app.services import interpreter_client as ic
-from backend.app.tasks.translation import retranslate_posts, translate_posts
+from backend.app.tasks.translation import (
+    _RETRANSLATE_COUNTDOWN,
+    retranslate_posts,
+    translate_posts,
+)
 
 pytestmark = pytest.mark.integration
 
@@ -115,6 +119,35 @@ def test_translate_posts_passthrough_english_marks_handled(db_sync, monkeypatch)
     assert p.post_title_translated is None       # ...but nothing to show
 
 
+def test_translate_posts_mixed_language_translates_nonenglish_field(db_sync, monkeypatch):
+    # English title + Japanese description: the description is still translated
+    # even though the title is already English (per-field detection, not the old
+    # aggregate first-item bail). Source lang comes from the translated field.
+    _patch(monkeypatch, db_sync)
+    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
+
+    def fake_translate(texts, **k):
+        t = texts[0]
+        if t.isascii():   # already English → passthrough
+            return {"translations": [t], "detected_lang": "en",
+                    "engine": "none", "engine_version": None}
+        return {"translations": [f"EN:{t}"], "detected_lang": "ja",
+                "engine": "llm", "engine_version": "v9"}
+
+    monkeypatch.setattr("backend.app.tasks.translation.ic.translate", fake_translate)
+    a = _artist(db_sync)
+    p = _post(db_sync, a.id, title="hello", desc="ねこ")
+    _enable(db_sync)
+    db_sync.commit()
+
+    assert "translated=1" in translate_posts()
+    db_sync.refresh(p)
+    assert p.post_title_translated is None          # English title untouched
+    assert p.description_translated == "EN:ねこ"      # Japanese desc translated
+    assert p.translated_source_lang == "ja"          # from the translated field
+    assert p.translation_engine_version == "v9"
+
+
 def test_translate_posts_disabled_is_noop(db_sync, monkeypatch):
     _patch(monkeypatch, db_sync)
     a = _artist(db_sync)
@@ -162,6 +195,63 @@ def test_translate_posts_interrupt_reenqueues_after_backoff(db_sync, monkeypatch
     assert calls[0][1]["countdown"] == 30
 
 
+def _mock_ok(monkeypatch):
+    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
+    monkeypatch.setattr(
+        "backend.app.tasks.translation.ic.translate",
+        lambda texts, **k: {
+            "translations": [f"EN:{t}" for t in texts],
+            "detected_lang": "ja", "engine": "llm", "engine_version": "v1",
+        },
+    )
+
+
+def test_translate_posts_drain_chases_tail(db_sync, monkeypatch):
+    # "Translate now" (drain=True): with the chunk capped at 1 and 2 untranslated
+    # posts, the first chunk re-enqueues itself — drain preserved — to finish the
+    # backlog rather than waiting for the next beat.
+    _patch(monkeypatch, db_sync)
+    _mock_ok(monkeypatch)
+    monkeypatch.setattr("backend.app.tasks.translation._MAX_POSTS_PER_RUN", 1)
+    calls = []
+    monkeypatch.setattr(
+        "backend.app.tasks.translation.translate_posts.apply_async",
+        lambda *a, **k: calls.append((a, k)),
+    )
+    a = _artist(db_sync)
+    _post(db_sync, a.id, title="ねこ", ext="p1")
+    _post(db_sync, a.id, title="いぬ", ext="p2")
+    _enable(db_sync)
+    db_sync.commit()
+
+    translate_posts(drain=True)
+    assert len(calls) == 1                       # one re-enqueue for the tail
+    args, kw = calls[0]
+    assert args[1] == {"drain": True}            # drain flag carried forward
+    assert kw["countdown"] == _RETRANSLATE_COUNTDOWN
+
+
+def test_translate_posts_beat_single_chunk_no_tail_chase(db_sync, monkeypatch):
+    # The periodic beat (drain defaults False) does exactly ONE chunk and never
+    # re-enqueues — the 8h cadence drives the rest.
+    _patch(monkeypatch, db_sync)
+    _mock_ok(monkeypatch)
+    monkeypatch.setattr("backend.app.tasks.translation._MAX_POSTS_PER_RUN", 1)
+    calls = []
+    monkeypatch.setattr(
+        "backend.app.tasks.translation.translate_posts.apply_async",
+        lambda *a, **k: calls.append((a, k)),
+    )
+    a = _artist(db_sync)
+    _post(db_sync, a.id, title="ねこ", ext="p1")
+    _post(db_sync, a.id, title="いぬ", ext="p2")
+    _enable(db_sync)
+    db_sync.commit()
+
+    translate_posts()                            # drain=False
+    assert calls == []                           # no self-re-enqueue
+
+
 def _mock_new_model(monkeypatch, *, ver="v2"):
     """Interpreter is up and translates with a NEW engine version."""
     monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)