Compare commits

...

2 Commits

Author SHA1 Message Date
bvandeusen 9eae636047 feat(translation): pooled Interpreter session + manual sweep resumes after drain
CI / lint (push) Successful in 4s
CI / frontend-build (push) Successful in 21s
CI / backend-lint-and-test (push) Successful in 34s
CI / integration (push) Successful in 3m45s
- interpreter_client: shared requests.Session with a connect-only retry
  (connect=2, no status retries — we map 429/5xx ourselves) so a proxy reload
  is smoothed and the keep-alive connection is pooled across the sweep.
- translate_posts: on an interrupt (drain), re-enqueue after the Retry-After
  hint / default backoff instead of waiting for the daily beat; self-terminates
  via the health gate. Steady-state one-chunk-per-run on success is unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-07 22:10:05 -04:00
bvandeusen f8105046dc feat(explore): exclude WIP-tagged work from the Explore rabbit-hole
Explore's neighbour grid (/api/gallery/similar → gallery_service.similar) now
takes an Explore-only exclude_wip flag that drops `wip` system-tagged images
from the candidates, alongside the banner/editor presentation tags. The
gallery's own "similar" button is unchanged (keeps wip, #1274) — only the
Explore store passes exclude_wip=1. The anchor itself may still be a WIP; only
neighbours are filtered.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-07 22:10:05 -04:00
9 changed files with 117 additions and 24 deletions
+5 -1
View File
@@ -145,6 +145,9 @@ async def similar():
filters, _sort = _parse_filters() filters, _sort = _parse_filters()
except (KeyError, ValueError): except (KeyError, ValueError):
return jsonify({"error": "similar_to query param required"}), 400 return jsonify({"error": "similar_to query param required"}), 400
# Explore passes exclude_wip=1 to also drop work-in-progress from the
# rabbit-hole; the gallery's own "similar" button omits it (keeps wip, #1274).
exclude_wip = request.args.get("exclude_wip") in ("1", "true", "True")
# post_id is the exclusive post-detail view — not a similarity scope. # post_id is the exclusive post-detail view — not a similarity scope.
# include_hidden is a gallery-browse flag; similar() has its OWN presentation # include_hidden is a gallery-browse flag; similar() has its OWN presentation
# exclusion (a similarity-quality concern, #1274), so drop it here (#141). # exclusion (a similarity-quality concern, #1274), so drop it here (#141).
@@ -154,7 +157,8 @@ async def similar():
async with get_session() as session: async with get_session() as session:
svc = GalleryService(session) svc = GalleryService(session)
try: try:
images = await svc.similar(image_id=similar_to, limit=limit, **scope) images = await svc.similar(
image_id=similar_to, limit=limit, exclude_wip=exclude_wip, **scope)
except ValueError as exc: except ValueError as exc:
return jsonify({"error": str(exc)}), 400 return jsonify({"error": str(exc)}), 400
if images is None: if images is None:
+4
View File
@@ -48,6 +48,10 @@ class TagKind(StrEnum):
# content. `wip` is real art: only the training pipelines exclude it. # content. `wip` is real art: only the training pipelines exclude it.
SYSTEM_TAG_NAMES = ("wip", "banner", "editor screenshot") SYSTEM_TAG_NAMES = ("wip", "banner", "editor screenshot")
PRESENTATION_SYSTEM_TAGS = ("banner", "editor screenshot") PRESENTATION_SYSTEM_TAGS = ("banner", "editor screenshot")
# `wip` marks real-but-unfinished art. It's kept in the gallery's own "similar"
# results (#1274), but the Explore rabbit-hole opts to hide it (exclude_wip) so a
# browse doesn't keep surfacing work-in-progress (operator, 2026-07-08).
WIP_SYSTEM_TAG = "wip"
image_tag = Table( image_tag = Table(
"image_tag", "image_tag",
+9 -4
View File
@@ -31,7 +31,7 @@ from ..models import (
Tag, Tag,
TagPositiveConfirmation, TagPositiveConfirmation,
) )
from ..models.tag import PRESENTATION_SYSTEM_TAGS, image_tag from ..models.tag import PRESENTATION_SYSTEM_TAGS, WIP_SYSTEM_TAG, image_tag
from .pagination import decode_cursor, encode_cursor from .pagination import decode_cursor, encode_cursor
from .tag_query import ( from .tag_query import (
fandom_join_alias, fandom_join_alias,
@@ -715,6 +715,7 @@ class GalleryService:
platform: str | None = None, platform: str | None = None,
untagged: bool = False, no_artist: bool = False, untagged: bool = False, no_artist: bool = False,
date_from: datetime | None = None, date_to: datetime | None = None, date_from: datetime | None = None, date_to: datetime | None = None,
exclude_wip: bool = False,
) -> list[GalleryImage] | None: ) -> list[GalleryImage] | None:
"""Visual "more like this": images near `image_id`'s SigLIP embedding """Visual "more like this": images near `image_id`'s SigLIP embedding
(pgvector, HNSW-indexed — alembic 0036), then DIVERSIFIED so the result (pgvector, HNSW-indexed — alembic 0036), then DIVERSIFIED so the result
@@ -751,14 +752,18 @@ class GalleryService:
# Presentation images (banner / editor-screenshot system tags, #128) # Presentation images (banner / editor-screenshot system tags, #128)
# cluster on UI chrome rather than content, so near any one of them # cluster on UI chrome rather than content, so near any one of them
# they'd fill the grid. Excluded from CANDIDATES only — the anchor # they'd fill the grid. Excluded from CANDIDATES only — the anchor
# itself may be a banner, and `wip` stays surfaced (real art; only # itself may be a banner. `wip` stays surfaced here by default (real art;
# the training pipelines exclude it). # only the training pipelines exclude it), but the Explore rabbit-hole
# passes exclude_wip to also drop work-in-progress (operator, 2026-07-08).
excluded_system_tags = PRESENTATION_SYSTEM_TAGS
if exclude_wip:
excluded_system_tags = (*PRESENTATION_SYSTEM_TAGS, WIP_SYSTEM_TAG)
presentation = ( presentation = (
select(image_tag.c.image_record_id) select(image_tag.c.image_record_id)
.join(Tag, Tag.id == image_tag.c.tag_id) .join(Tag, Tag.id == image_tag.c.tag_id)
.where( .where(
Tag.is_system.is_(True), Tag.is_system.is_(True),
Tag.name.in_(PRESENTATION_SYSTEM_TAGS), Tag.name.in_(excluded_system_tags),
) )
) )
stmt = stmt.where( stmt = stmt.where(
+20 -2
View File
@@ -12,6 +12,8 @@ from datetime import UTC, datetime
from email.utils import parsedate_to_datetime from email.utils import parsedate_to_datetime
import requests import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class InterpreterUnavailable(Exception): class InterpreterUnavailable(Exception):
@@ -57,13 +59,29 @@ def _parse_retry_after(resp) -> float | None:
return max(0.0, (when - datetime.now(UTC)).total_seconds()) return max(0.0, (when - datetime.now(UTC)).total_seconds())
# A shared session pools the keep-alive connection across the per-post sweep
# calls, and retries CONNECT failures only (connect=2, short backoff) — smoothing
# the instant a reverse proxy reloads. Status codes are deliberately NOT retried
# (status=0, raise_on_status=False): translate() maps 429/5xx → InterpreterUnavailable
# itself, and letting urllib3 retry a draining 503 would defeat the Retry-After
# backoff we honour upstream.
_retry = Retry(
total=None, connect=2, read=0, redirect=0, status=0,
backoff_factor=0.3, raise_on_status=False,
)
session = requests.Session()
_adapter = HTTPAdapter(max_retries=_retry)
session.mount("http://", _adapter)
session.mount("https://", _adapter)
def health(base_url: str, *, timeout: float = 5.0) -> bool: def health(base_url: str, *, timeout: float = 5.0) -> bool:
"""True iff the Interpreter LLM engine is up. Any error (unset URL, network, """True iff the Interpreter LLM engine is up. Any error (unset URL, network,
non-200, engine down) → False, so the sweep just no-ops rather than raising.""" non-200, engine down) → False, so the sweep just no-ops rather than raising."""
if not base_url: if not base_url:
return False return False
try: try:
r = requests.get(_url(base_url, "/v1/health"), timeout=timeout) r = session.get(_url(base_url, "/v1/health"), timeout=timeout)
except requests.RequestException: except requests.RequestException:
return False return False
if r.status_code != 200: if r.status_code != 200:
@@ -94,7 +112,7 @@ def translate(
return {"translations": [], "detected_lang": None, return {"translations": [], "detected_lang": None,
"engine": None, "engine_version": None} "engine": None, "engine_version": None}
try: try:
r = requests.post( r = session.post(
_url(base_url, "/v1/translate"), _url(base_url, "/v1/translate"),
json={"q": list(texts), "source": source, json={"q": list(texts), "source": source,
"target": target, "engine": "auto"}, "target": target, "engine": "auto"},
+15 -4
View File
@@ -70,10 +70,21 @@ def translate_posts() -> str:
return ready return ready
base_url, target = ready base_url, target = ready
posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN) posts = _select_untranslated(session, None, _MAX_POSTS_PER_RUN)
# Steady-state daily sweep: one chunk per run. On interruption the status, translated, retry_after = _translate_batch(
# untranslated rows stay NULL and the next daily beat (or a fresh session, posts, base_url, target,
# "Translate now") resumes them — no self-chase needed here. )
status, translated, _retry = _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):
delay = _interrupt_backoff(retry_after)
translate_posts.apply_async((), {}, countdown=delay)
log.info(
"translate_posts: interrupted (service draining) → retry in %ss",
delay,
)
return _summary(status, translated, len(posts)) return _summary(status, translated, len(posts))
+3 -1
View File
@@ -44,7 +44,9 @@ export const useExploreStore = defineStore('explore', () => {
// empty and let the view explain why (anchor.has_embedding === false). // empty and let the view explain why (anchor.has_embedding === false).
if (detail.has_embedding) { if (detail.has_embedding) {
const body = await api.get('/api/gallery/similar', { const body = await api.get('/api/gallery/similar', {
params: { similar_to: numId, limit: NEIGHBOR_LIMIT }, // exclude_wip: keep work-in-progress out of the Explore rabbit-hole
// (the gallery's own "similar" button still shows it) — operator 2026-07-08.
params: { similar_to: numId, limit: NEIGHBOR_LIMIT, exclude_wip: 1 },
}) })
if (!t.isCurrent()) return if (!t.isCurrent()) return
neighbors.value = body.images || [] neighbors.value = body.images || []
+24
View File
@@ -98,6 +98,30 @@ async def test_similar_excludes_presentation_tagged_images(db):
assert {i.id for i in res_from_banner} == {src.id, wipped.id, plain.id} assert {i.id for i in res_from_banner} == {src.id, wipped.id, plain.id}
@pytest.mark.asyncio
async def test_similar_exclude_wip_drops_wip_neighbors(db):
"""Explore passes exclude_wip=True to also hide work-in-progress from the
rabbit-hole (banner is always hidden; wip only when asked)."""
src = await _img(db, 1, _vec(1, 0))
bannered = await _img(db, 2, _vec(1, 0.02)) # always hidden
wipped = await _img(db, 3, _vec(1, 0.3)) # hidden only with exclude_wip
plain = await _img(db, 4, _vec(1, 0.6))
banner_tag = (await db.execute(select(Tag).where(
Tag.is_system.is_(True), Tag.name == "banner"))).scalar_one()
wip_tag = (await db.execute(select(Tag).where(
Tag.is_system.is_(True), Tag.name == "wip"))).scalar_one()
await db.execute(image_tag.insert().values(
image_record_id=bannered.id, tag_id=banner_tag.id, source="manual"))
await db.execute(image_tag.insert().values(
image_record_id=wipped.id, tag_id=wip_tag.id, source="manual"))
svc = GalleryService(db)
res = await svc.similar(src.id, limit=10, exclude_wip=True)
assert [i.id for i in res] == [plain.id] # wip + banner both gone
# Default (gallery "similar" button) still keeps wip (#1274).
res_default = await svc.similar(src.id, limit=10)
assert wipped.id in {i.id for i in res_default}
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_similar_composes_with_tag_filter(db): async def test_similar_composes_with_tag_filter(db):
src = await _img(db, 1, _vec(1, 0)) src = await _img(db, 1, _vec(1, 0))
+12 -12
View File
@@ -30,7 +30,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
"interpreter": {"engine": "llm", "engine_version": "ollama:x:12b"}, "interpreter": {"engine": "llm", "engine_version": "ollama:x:12b"},
}) })
monkeypatch.setattr(ic.requests, "post", fake_post) monkeypatch.setattr(ic.session, "post", fake_post)
out = ic.translate(["ねこが可愛い", "金髪ギャル!"], base_url="http://i.lan") out = ic.translate(["ねこが可愛い", "金髪ギャル!"], base_url="http://i.lan")
assert out["translations"] == ["The cat is cute", "Blonde gal!"] assert out["translations"] == ["The cat is cute", "Blonde gal!"]
assert out["detected_lang"] == "ja" assert out["detected_lang"] == "ja"
@@ -41,7 +41,7 @@ def test_translate_maps_batch_in_order(monkeypatch):
def test_translate_passthrough_unchanged(monkeypatch): def test_translate_passthrough_unchanged(monkeypatch):
# Already-English items come back unchanged in their slot (engine "none"). # Already-English items come back unchanged in their slot (engine "none").
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(200, { monkeypatch.setattr(ic.session, "post", lambda *a, **k: _Resp(200, {
"translatedText": ["already english"], "translatedText": ["already english"],
"detectedLanguage": {"language": "en"}, "detectedLanguage": {"language": "en"},
"interpreter": {"engine": "none", "engine_version": None}, "interpreter": {"engine": "none", "engine_version": None},
@@ -53,7 +53,7 @@ def test_translate_passthrough_unchanged(monkeypatch):
def test_translate_503_raises_unavailable(monkeypatch): def test_translate_503_raises_unavailable(monkeypatch):
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(503, {})) monkeypatch.setattr(ic.session, "post", lambda *a, **k: _Resp(503, {}))
with pytest.raises(ic.InterpreterUnavailable): with pytest.raises(ic.InterpreterUnavailable):
ic.translate(["x"], base_url="http://i.lan") ic.translate(["x"], base_url="http://i.lan")
@@ -62,7 +62,7 @@ def test_translate_connection_error_raises_unavailable(monkeypatch):
def boom(*a, **k): def boom(*a, **k):
raise ic.requests.ConnectionError("refused") raise ic.requests.ConnectionError("refused")
monkeypatch.setattr(ic.requests, "post", boom) monkeypatch.setattr(ic.session, "post", boom)
with pytest.raises(ic.InterpreterUnavailable): with pytest.raises(ic.InterpreterUnavailable):
ic.translate(["x"], base_url="http://i.lan") ic.translate(["x"], base_url="http://i.lan")
@@ -71,7 +71,7 @@ def test_translate_connection_error_raises_unavailable(monkeypatch):
def test_translate_429_and_5xx_raise_unavailable(monkeypatch, code): def test_translate_429_and_5xx_raise_unavailable(monkeypatch, code):
# A gracefully-draining service behind a reverse proxy returns 429/502/503/504 # A gracefully-draining service behind a reverse proxy returns 429/502/503/504
# — all mean "retry later", not an opaque error. # — all mean "retry later", not an opaque error.
monkeypatch.setattr(ic.requests, "post", lambda *a, **k: _Resp(code, {})) monkeypatch.setattr(ic.session, "post", lambda *a, **k: _Resp(code, {}))
with pytest.raises(ic.InterpreterUnavailable) as ei: with pytest.raises(ic.InterpreterUnavailable) as ei:
ic.translate(["x"], base_url="http://i.lan") ic.translate(["x"], base_url="http://i.lan")
assert ei.value.retry_after is None assert ei.value.retry_after is None
@@ -79,7 +79,7 @@ def test_translate_429_and_5xx_raise_unavailable(monkeypatch, code):
def test_translate_honours_retry_after_seconds(monkeypatch): def test_translate_honours_retry_after_seconds(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
ic.requests, "post", ic.session, "post",
lambda *a, **k: _Resp(503, {}, headers={"Retry-After": "42"}), lambda *a, **k: _Resp(503, {}, headers={"Retry-After": "42"}),
) )
with pytest.raises(ic.InterpreterUnavailable) as ei: with pytest.raises(ic.InterpreterUnavailable) as ei:
@@ -90,7 +90,7 @@ def test_translate_honours_retry_after_seconds(monkeypatch):
def test_translate_retry_after_past_http_date_clamps_to_zero(monkeypatch): def test_translate_retry_after_past_http_date_clamps_to_zero(monkeypatch):
# HTTP-date form; a past date → non-negative clamp to 0 (deterministic). # HTTP-date form; a past date → non-negative clamp to 0 (deterministic).
monkeypatch.setattr( monkeypatch.setattr(
ic.requests, "post", ic.session, "post",
lambda *a, **k: _Resp( lambda *a, **k: _Resp(
503, {}, headers={"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}), 503, {}, headers={"Retry-After": "Wed, 21 Oct 2015 07:28:00 GMT"}),
) )
@@ -101,7 +101,7 @@ def test_translate_retry_after_past_http_date_clamps_to_zero(monkeypatch):
def test_translate_400_raises_bad_request(monkeypatch): def test_translate_400_raises_bad_request(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
ic.requests, "post", lambda *a, **k: _Resp(400, {"error": "bad target"}) ic.session, "post", lambda *a, **k: _Resp(400, {"error": "bad target"})
) )
with pytest.raises(ic.InterpreterBadRequest): with pytest.raises(ic.InterpreterBadRequest):
ic.translate(["x"], base_url="http://i.lan") ic.translate(["x"], base_url="http://i.lan")
@@ -111,20 +111,20 @@ def test_translate_empty_is_noop(monkeypatch):
def boom(*a, **k): def boom(*a, **k):
raise AssertionError("should not call the service for an empty batch") raise AssertionError("should not call the service for an empty batch")
monkeypatch.setattr(ic.requests, "post", boom) monkeypatch.setattr(ic.session, "post", boom)
assert ic.translate([], base_url="http://i.lan")["translations"] == [] assert ic.translate([], base_url="http://i.lan")["translations"] == []
def test_health_true_when_llm_up(monkeypatch): def test_health_true_when_llm_up(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
ic.requests, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": True}}) ic.session, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": True}})
) )
assert ic.health("http://i.lan") is True assert ic.health("http://i.lan") is True
def test_health_false_when_engine_down(monkeypatch): def test_health_false_when_engine_down(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
ic.requests, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": False}}) ic.session, "get", lambda *a, **k: _Resp(200, {"engines": {"llm": False}})
) )
assert ic.health("http://i.lan") is False assert ic.health("http://i.lan") is False
@@ -133,7 +133,7 @@ def test_health_false_on_error(monkeypatch):
def boom(*a, **k): def boom(*a, **k):
raise ic.requests.ConnectionError("refused") raise ic.requests.ConnectionError("refused")
monkeypatch.setattr(ic.requests, "get", boom) monkeypatch.setattr(ic.session, "get", boom)
assert ic.health("http://i.lan") is False assert ic.health("http://i.lan") is False
+25
View File
@@ -137,6 +137,31 @@ def test_translate_posts_service_down_leaves_untranslated(db_sync, monkeypatch):
assert p.translated_source_lang is None # will retry next run assert p.translated_source_lang is None # will retry next run
def test_translate_posts_interrupt_reenqueues_after_backoff(db_sync, monkeypatch):
# A drain mid-sweep (health passed, translate 503s w/ Retry-After) re-enqueues
# the daily sweep after the backoff instead of waiting for tomorrow's beat.
_patch(monkeypatch, db_sync)
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
def _drain(texts, **k):
raise ic.InterpreterUnavailable("draining", retry_after=30)
monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
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")
_enable(db_sync)
db_sync.commit()
assert "interrupted" in translate_posts()
assert len(calls) == 1
assert calls[0][1]["countdown"] == 30
def _mock_new_model(monkeypatch, *, ver="v2"): def _mock_new_model(monkeypatch, *, ver="v2"):
"""Interpreter is up and translates with a NEW engine version.""" """Interpreter is up and translates with a NEW engine version."""
monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True) monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)