|
|
@@ -30,6 +30,7 @@ import mimetypes
|
|
|
import os
|
|
|
import re
|
|
|
import tempfile
|
|
|
+from collections import deque
|
|
|
from datetime import UTC, datetime
|
|
|
from difflib import SequenceMatcher
|
|
|
from enum import StrEnum
|
|
|
@@ -302,12 +303,13 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
self._room_kinds: dict[str, RoomKind] = {}
|
|
|
# Event IDs already processed — chattolib may redeliver events across
|
|
|
# reconnects, so every inbound event is checked against this list.
|
|
|
- self._seen: list[str] = []
|
|
|
+ # Bounded deques: appending past the cap drops the oldest ID on its own.
|
|
|
+ self._seen: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
|
|
|
# Message IDs this adapter has handed to the gateway (posted or
|
|
|
# edit-re-dispatched). Edits of anything on this list never start a
|
|
|
# fresh turn — that is the lock against re-answering settled
|
|
|
# conversations by editing old messages.
|
|
|
- self._dispatched_ids: list[str] = []
|
|
|
+ self._dispatched_ids: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
|
|
|
# session_key -> message ID currently being processed there. Written
|
|
|
# by on_processing_start, cleared by on_processing_complete; an edit
|
|
|
# landing on the recorded ID is a mid-run correction.
|
|
|
@@ -398,16 +400,6 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
raise RuntimeError("Chatto client unavailable")
|
|
|
return client
|
|
|
|
|
|
- async def _ensure_token(self) -> bool:
|
|
|
- """Ensure we have a logged-in Chatto client and token."""
|
|
|
- if self.chatto_config.token.value and isinstance(
|
|
|
- self._chatto_client, ChattoClient
|
|
|
- ):
|
|
|
- return True
|
|
|
-
|
|
|
- client = await self._get_chatto_client()
|
|
|
- return client is not None
|
|
|
-
|
|
|
# ------------------------------------------------------------------ #
|
|
|
# Connection
|
|
|
# ------------------------------------------------------------------ #
|
|
|
@@ -439,9 +431,6 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
"""
|
|
|
logger.info("Chatto: connecting...")
|
|
|
|
|
|
- if not await self._ensure_token():
|
|
|
- return False
|
|
|
-
|
|
|
client = await self._get_chatto_client()
|
|
|
if client is None:
|
|
|
self._set_fatal_error(
|
|
|
@@ -598,9 +587,8 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
# ------------------------------------------------------------------ #
|
|
|
|
|
|
def _mark_seen(self, event_id: str) -> None:
|
|
|
+ # The deque's maxlen evicts the oldest ID — no manual trimming.
|
|
|
self._seen.append(event_id)
|
|
|
- while len(self._seen) > ChattoConstants.SEEN_CAP:
|
|
|
- del self._seen[0]
|
|
|
|
|
|
def _is_seen(self, event_id: str) -> bool:
|
|
|
return event_id in self._seen
|
|
|
@@ -769,7 +757,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
client: ChattoClient,
|
|
|
argument: str,
|
|
|
) -> tuple[str | None, RoomWithViewerState | None]:
|
|
|
- """Resolve a /join//leave argument to a room.
|
|
|
+ """Resolve a ``/join`` or ``/leave`` argument to a room.
|
|
|
|
|
|
``#name`` is looked up case-insensitively in a fresh directory scan
|
|
|
(which also refreshes our name/kind caches); anything else is treated
|
|
|
@@ -1110,8 +1098,6 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
if not message_id:
|
|
|
return
|
|
|
self._dispatched_ids.append(message_id)
|
|
|
- while len(self._dispatched_ids) > ChattoConstants.SEEN_CAP:
|
|
|
- del self._dispatched_ids[0]
|
|
|
|
|
|
def _edit_is_fresh(self, message: Message) -> bool:
|
|
|
"""Whether this edit is young enough to still be processed.
|
|
|
@@ -1224,7 +1210,7 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
await self.cancel_session_processing(
|
|
|
session_key, release_guard=True, discard_pending=False
|
|
|
)
|
|
|
- elif payload.message_event_id not in self._dispatched_ids:
|
|
|
+ elif not was_dispatched:
|
|
|
logger.info(
|
|
|
"Chatto: msg %s was never dispatched - edit starts a fresh turn",
|
|
|
payload.message_event_id,
|
|
|
@@ -1844,8 +1830,10 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
except Exception as e:
|
|
|
# ChattoError included: both read as "this chunk did not go
|
|
|
# out" and stop the batch — the SendResult carries the reason.
|
|
|
+ # Only server-side failures count as retryable; a bug in our
|
|
|
+ # own code must not read as a transient network blip.
|
|
|
last_error = str(e)
|
|
|
- retryable = True
|
|
|
+ retryable = isinstance(e, ChattoError)
|
|
|
break
|
|
|
|
|
|
self._mark_seen(msg_obj.id)
|
|
|
@@ -2435,6 +2423,8 @@ class ChattoAdapter(BasePlatformAdapter):
|
|
|
self._mark_seen(msg.id)
|
|
|
return SendResult(success=True, message_id=msg.id)
|
|
|
except ChattoError as e:
|
|
|
+ # Same classification as the chunked text path: server-side
|
|
|
+ # failures retry, anything else is ours and must not loop.
|
|
|
return SendResult(success=False, error=str(e), retryable=True)
|
|
|
except Exception as e:
|
|
|
return SendResult(success=False, error=str(e), retryable=False)
|
|
|
@@ -2811,7 +2801,13 @@ async def hermes_standalone_sender_fn(
|
|
|
chat_id,
|
|
|
exc,
|
|
|
)
|
|
|
- return SendResult(success=False, error=str(exc))
|
|
|
+ # Same classification as the adapter's send paths: only
|
|
|
+ # server-side failures read as retryable.
|
|
|
+ return SendResult(
|
|
|
+ success=False,
|
|
|
+ error=str(exc),
|
|
|
+ retryable=isinstance(exc, ChattoError),
|
|
|
+ )
|
|
|
return SendResult(success=True, message_id=message_ids[0])
|
|
|
finally:
|
|
|
try:
|
