fix(llm): 收紧贴纸视觉投递领取查询

Author: TogetsuDo

pallas/core/foundation/db/repository_pg.py

@@ -170,6 +170,7 @@ class BackgroundJobRow(Base):
     __table_args__ = (
         UniqueConstraint("idempotency_key", name="uq_background_job_idempotency"),
         Index("ix_background_job_claim", "status", "available_at", "leased_until", "id"),
+        Index("ix_background_job_delivery_claim", "kind", "status", "finished_at"),
     )
 
     id: Mapped[str] = mapped_column(Text, primary_key=True)
@@ -735,6 +736,18 @@ def _ensure_pg_context_answer_message_reply_index(connection) -> None:
     )
 
 
+def _ensure_pg_background_job_delivery_claim_index(connection) -> None:
+    """background_job 表补已完成视觉投递领取索引。"""
+    insp = inspect(connection)
+    if not insp.has_table("background_job"):
+        return
+    connection.execute(
+        text(
+            "CREATE INDEX IF NOT EXISTS ix_background_job_delivery_claim ON background_job (kind, status, finished_at)"
+        )
+    )
+
+
 def _ensure_pg_stat_statements_extension(connection) -> None:
     """启用 pg_stat_statements（仅应在独立 autocommit 连接中调用）。
 
@@ -840,6 +853,7 @@ async def get_session(*, read_only: bool = False):
     ("ddl.message_timeline_metadata", _ensure_pg_message_timeline_metadata),
     ("ddl.context_answer_reply_index", _ensure_pg_context_answer_reply_index),
     ("ddl.context_answer_message_reply_index", _ensure_pg_context_answer_message_reply_index),
+    ("ddl.background_job_delivery_claim_index", _ensure_pg_background_job_delivery_claim_index),
 ]
 
 

pallas/product/llm/sticker_vision.py

@@ -436,15 +436,22 @@ async def claim_sticker_vision_delivery(bot_ids: set[int]) -> dict[str, object]
     from pallas.core.foundation.db.runtime import is_postgresql_backend
 
     if is_postgresql_backend():
-        from sqlalchemy import select
+        from sqlalchemy import String, cast, select
 
         from pallas.core.foundation.db.repository_pg import BackgroundJobRow, get_session
 
         async with get_session() as session:
             rows = (
                 await session.execute(
                     select(BackgroundJobRow)
-                    .where(BackgroundJobRow.kind == "sticker_vision.select", BackgroundJobRow.status == "done")
+                    .where(
+                        BackgroundJobRow.kind == "sticker_vision.select",
+                        BackgroundJobRow.status == "done",
+                        BackgroundJobRow.payload["delivery"]["state"].astext == "pending",
+                        cast(BackgroundJobRow.payload["delivery"]["bot_id"].astext, String).in_([
+                            str(bot_id) for bot_id in bot_ids
+                        ]),
+                    )
                     .order_by(BackgroundJobRow.finished_at)
                     .limit(32)
                     .with_for_update(skip_locked=True)

tests/common/test_repository_pg.py

@@ -514,6 +514,37 @@ def execute(self, statement) -> None:
     ]
 
 
+def test_background_job_rows_have_delivery_claim_index():
+    from pallas.core.foundation.db.repository_pg import BackgroundJobRow
+
+    index_names = {idx.name for idx in BackgroundJobRow.__table__.indexes}
+
+    assert "ix_background_job_delivery_claim" in index_names
+
+
+def test_ensure_pg_background_job_delivery_claim_index(monkeypatch):
+    from pallas.core.foundation.db import repository_pg as mod
+
+    executed: list[str] = []
+
+    class FakeInspector:
+        def has_table(self, name: str) -> bool:
+            return name == "background_job"
+
+    class FakeConnection:
+        def execute(self, statement) -> None:
+            executed.append(str(statement))
+
+    monkeypatch.setattr(mod, "inspect", lambda _connection: FakeInspector())
+
+    mod._ensure_pg_background_job_delivery_claim_index(FakeConnection())
+
+    assert executed == [
+        "CREATE INDEX IF NOT EXISTS ix_background_job_delivery_claim "
+        "ON background_job (kind, status, finished_at)"
+    ]
+
+
 @pytest.mark.asyncio
 async def test_find_by_keywords_for_reply_many_answers_no_in_overflow(pg_engine, monkeypatch):
     """热词大量 Answer 时不得用超大 IN (...)，接话 find 应成功且受 reply_answers_cap 限制。"""