fix(avatar): recover BOXIM image replies

digital-avatar-app/backend/services/takeover_service.py

@@ -49,6 +49,8 @@ BOXIM_TEXT_MESSAGE_TYPE = 0
 BOXIM_IMAGE_MESSAGE_TYPE = 1
 BOXIM_IMAGE_PROMPT = "请看看这张图片。"
 BOXIM_IMAGE_UNAVAILABLE_REPLY = "这张图片我暂时没看清，麻烦重新发送一张清晰的原图。"
+IMAGE_CONTEXT_LOOKBACK_SECONDS = 1800
+MAX_RECENT_IMAGE_CONTEXTS = 3
 
 
 def _utcnow() -> datetime:
@@ -149,6 +151,7 @@ class TakeoverService:
         self._sessions: dict[str, dict] = {}
         self._poll_lock = asyncio.Lock()
         self._process_lock = asyncio.Lock()
+        self._persist_lock = asyncio.Lock()
 
     async def poll_and_process_messages(self):
         """Run one complete cycle for callers that do not use the split scheduler."""
@@ -409,13 +412,6 @@ class TakeoverService:
             max_message_id = _numeric_id(cursor.last_message_id)
             read_receipts: dict[str, int] = {}
             for message in messages:
-                self._record_message(
-                    db,
-                    avatar,
-                    cursor.boxim_owner_id,
-                    message,
-                    schedule_reply=not priming,
-                )
                 message_id = _numeric_id(message.get("id"))
                 max_message_id = max(max_message_id, message_id)
                 send_id = str(message.get("sendId") or "")
@@ -430,11 +426,22 @@ class TakeoverService:
                     session["access_token"], peer_id, message_id
                 )
 
-            cursor.last_message_id = str(max_message_id)
-            cursor.initialized = True
-            cursor.last_polled_at = self.now()
-            cursor.last_error = ""
-            db.commit()
+            # Keep SQLite write transactions short. The read-receipt request above
+            # can block on the network and must not hold the database write lock.
+            async with self._persist_lock:
+                for message in messages:
+                    self._record_message(
+                        db,
+                        avatar,
+                        cursor.boxim_owner_id,
+                        message,
+                        schedule_reply=not priming,
+                    )
+                cursor.last_message_id = str(max_message_id)
+                cursor.initialized = True
+                cursor.last_polled_at = self.now()
+                cursor.last_error = ""
+                db.commit()
             return True
         except Exception:
             db.rollback()
@@ -645,6 +652,15 @@ class TakeoverService:
                 task.status = "cancelled"
                 task.cancel_reason = "newer_incoming_message"
                 task.locked_at = None
+        if event.message_type == BOXIM_TEXT_MESSAGE_TYPE:
+            for image_event in self._recent_unhandled_images(
+                db,
+                avatar,
+                event,
+                source_ids,
+            ):
+                prompt_parts.append(_event_prompt(image_event))
+                source_ids.append(image_event.boxim_message_id)
         prompt_parts.append(_event_prompt(event))
         source_ids.append(event.boxim_message_id)
         prompt = "\n".join(part for part in prompt_parts if part).strip()[-MAX_PROMPT_LENGTH:]
@@ -670,6 +686,54 @@ class TakeoverService:
             )
         )
 
+    @staticmethod
+    def _recent_unhandled_images(
+        db: Session,
+        avatar: Avatar,
+        event: TakeoverMessage,
+        current_source_ids: list[str],
+    ) -> list[TakeoverMessage]:
+        """Recover a recent image that an older deployment recorded without a task."""
+        threshold = event.send_time - timedelta(seconds=IMAGE_CONTEXT_LOOKBACK_SECONDS)
+        candidates = (
+            db.query(TakeoverMessage)
+            .filter(
+                TakeoverMessage.avatar_id == avatar.id,
+                TakeoverMessage.owner_id == avatar.owner_id,
+                TakeoverMessage.peer_id == event.peer_id,
+                TakeoverMessage.direction == "incoming",
+                TakeoverMessage.message_type == BOXIM_IMAGE_MESSAGE_TYPE,
+                TakeoverMessage.is_avatar.is_(False),
+                TakeoverMessage.send_time >= threshold,
+                TakeoverMessage.send_time <= event.send_time,
+            )
+            .order_by(TakeoverMessage.send_time.desc())
+            .limit(MAX_RECENT_IMAGE_CONTEXTS)
+            .all()
+        )
+        if not candidates:
+            return []
+
+        handled_ids = set(current_source_ids)
+        task_sources = (
+            db.query(TakeoverReplyTask.source_message_ids)
+            .filter(
+                TakeoverReplyTask.avatar_id == avatar.id,
+                TakeoverReplyTask.owner_id == avatar.owner_id,
+                TakeoverReplyTask.peer_id == event.peer_id,
+                TakeoverReplyTask.created_at >= threshold,
+            )
+            .all()
+        )
+        for (source_message_ids,) in task_sources:
+            handled_ids.update(source_message_ids or [])
+
+        return [
+            image
+            for image in reversed(candidates)
+            if image.boxim_message_id not in handled_ids
+        ]
+
     async def _prepare_replies(self) -> int:
         db = self.session_factory()
         try:
@@ -766,6 +830,21 @@ class TakeoverService:
             db.commit()
 
             excluded_ids = set(task.source_message_ids or [])
+            source_events = {
+                event.boxim_message_id: event
+                for event in (
+                    db.query(TakeoverMessage)
+                    .filter(
+                        TakeoverMessage.owner_id == task.owner_id,
+                        TakeoverMessage.peer_id == task.peer_id,
+                        TakeoverMessage.avatar_id == task.avatar_id,
+                        TakeoverMessage.boxim_message_id.in_(excluded_ids),
+                    )
+                    .all()
+                    if excluded_ids
+                    else []
+                )
+            }
             events = (
                 db.query(TakeoverMessage)
                 .filter(
@@ -777,11 +856,6 @@ class TakeoverService:
                 .limit(30)
                 .all()
             )
-            source_events = {
-                event.boxim_message_id: event
-                for event in events
-                if event.boxim_message_id in excluded_ids
-            }
             image_attachments = []
             image_failed = False
             for message_id in (task.source_message_ids or [])[-3:]:
@@ -821,11 +895,7 @@ class TakeoverService:
             from routers.chat import _attachment_contexts, _resolve_reply
 
             image_contexts = _attachment_contexts(image_attachments)
-            has_source_text = any(
-                event.message_type == BOXIM_TEXT_MESSAGE_TYPE and event.content.strip()
-                for event in source_events.values()
-            )
-            if image_failed and not image_contexts and not has_source_text:
+            if image_failed and not image_contexts:
                 answer = BOXIM_IMAGE_UNAVAILABLE_REPLY
             else:
                 result = _resolve_reply(