diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py index 3eb83abb28..365d9b27e7 100644 --- a/plugins/platforms/matrix/adapter.py +++ b/plugins/platforms/matrix/adapter.py @@ -77,7 +77,7 @@ from gateway.platforms.base import ( from gateway.platforms.base import transcode_to_ogg_opus from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome from gateway.platforms.helpers import ThreadParticipationTracker -from plugins.platforms.matrix.voice_mention import ParkedVoices, has_voice_marker, is_voice_event +from plugins.platforms.matrix.voice_mention import ParkedVoices, VoiceGate, has_voice_marker, is_voice_event logger = logging.getLogger(__name__) @@ -535,6 +535,11 @@ def _csv_set(raw: Any) -> Set[str]: return {r.strip() for r in str(raw).split(",") if r.strip()} +def _thread_root(relates_to: dict) -> Optional[str]: + """The m.thread root event_id an event belongs to, else None.""" + return relates_to.get("event_id") if relates_to.get("rel_type") == "m.thread" else None + + def _extra_csv_set(config, key: str, env_name: str) -> Set[str]: """Resolve a room/user list: scoped env var → config.extra[key] → empty.""" return _csv_set(_extra_or_secret(config.extra, key, env_name, "", blank_is_unset=False)) @@ -2028,7 +2033,8 @@ class MatrixAdapter(BasePlatformAdapter): async def _resolve_message_context( self, room_id: str, sender: str, event_id: str, body: str, source_content: dict, - relates_to: dict, mention_claimed: bool = False, voice_gate=None) -> Optional[tuple]: + relates_to: dict, mention_claimed: bool = False, + voice_gate: Optional[VoiceGate] = None) -> Optional[tuple]: """Shared mention/thread/DM gating. Returns (body, is_dm, chat_type, thread_id, display_name, source) or None when the message should be dropped. ``mention_claimed`` marks a parked voice claimed by the sender's follow-up bare @mention; ``voice_gate`` is @@ -2036,7 +2042,7 @@ class MatrixAdapter(BasePlatformAdapter): identity = await self._resolve_room_identity(room_id) is_dm = await self._is_dm_room(room_id) chat_type = "dm" if is_dm else "group" - thread_id = relates_to.get("event_id") if relates_to.get("rel_type") == "m.thread" else None + thread_id = _thread_root(relates_to) is_mentioned = mention_claimed or self._content_mentions_bot(body, source_content) if not is_dm: # Whitelist first: non-listed rooms are dropped even when @mentioned (DMs exempt). @@ -2148,8 +2154,9 @@ class MatrixAdapter(BasePlatformAdapter): # (both only happen under require_mention). if (self._parked_voices.pending(room_id, sender) and not self._strip_mention(body).strip() and self._content_mentions_bot(body, source_content)): + limit = self._parked_voices.mark() # never claim a voice sent after this mention await self._parked_voices.settle(room_id, sender) # same-/sync-batch voice still gating - parked = self._parked_voices.claim(room_id, sender) + parked = self._parked_voices.claim(room_id, sender, before=limit) if parked: # answer the voice this bare mention was typed for, not an empty text voice_id, voice_content, voice_relates = parked await self._handle_media_message( @@ -2916,7 +2923,7 @@ class MatrixAdapter(BasePlatformAdapter): return False if room_id in self._free_rooms or (self._allowed_rooms and room_id not in self._allowed_rooms): return False - thread_id = relates_to.get("event_id") if relates_to.get("rel_type") == "m.thread" else None + thread_id = _thread_root(relates_to) if thread_id and thread_id in self._threads: return False return not body.startswith("/") and not self._content_mentions_bot(body, content) diff --git a/plugins/platforms/matrix/voice_mention.py b/plugins/platforms/matrix/voice_mention.py index 52f63a76c7..0ec3cf71a6 100644 --- a/plugins/platforms/matrix/voice_mention.py +++ b/plugins/platforms/matrix/voice_mention.py @@ -9,13 +9,14 @@ the window claims it. Unmentioned voices are never downloaded or transcribed whi from __future__ import annotations import asyncio -import itertools import time from typing import Dict, List, Optional, Tuple CLAIM_WINDOW_SECONDS = 120.0 # How long a bare mention waits for a voice from the same /sync batch that is still being gated. SETTLE_TIMEOUT_SECONDS = 5.0 +# Voices a sender can have parked per room at once (oldest dropped beyond this). +MAX_PARKED_PER_SENDER = 4 # (voice event_id, content, relates_to) ParkedVoice = Tuple[str, dict, dict] @@ -42,30 +43,39 @@ class VoiceGate: class ParkedVoices: def __init__(self) -> None: - # (room_id, sender) -> (parked_at, seq, parked voice) - self._parked: Dict[Tuple[str, str], Tuple[float, int, ParkedVoice]] = {} + # (room_id, sender) -> parked voices as (parked_at, seq, voice), ordered by seq, bounded. + self._parked: Dict[Tuple[str, str], List[Tuple[float, int, ParkedVoice]]] = {} # (room_id, sender) -> every voice of that sender still being gated. mautrix runs one # /sync batch's events as concurrent tasks, so a voice may still be awaiting room # identity when its bare mention is handled -- and a sender can have several in flight. self._inflight: Dict[Tuple[str, str], List[VoiceGate]] = {} - # (room_id, sender) -> seq of the newest voice parked while gates were in flight, so an - # older voice finishing late never lands over (or after the claim of) a newer one. - self._newest: Dict[Tuple[str, str], int] = {} - self._seq = itertools.count() + # (room_id, sender) -> seq of the last claimed voice while gates were in flight, so an + # older voice finishing late never parks after (and outlives) that claim. + self._floor: Dict[Tuple[str, str], int] = {} + self._next_seq = 0 def _prune(self) -> None: cutoff = time.monotonic() - CLAIM_WINDOW_SECONDS - self._parked = {k: v for k, v in self._parked.items() if v[0] >= cutoff} + kept = {k: [e for e in v if e[0] >= cutoff] for k, v in self._parked.items()} + self._parked = {k: v for k, v in kept.items() if v} def pending(self, room_id: str, sender: str) -> bool: - """Cheap pre-check: a voice is parked (maybe expired; ``claim`` prunes) or still being gated.""" + """Cheap pre-check: an unexpired voice is parked or one is still being gated.""" key = (room_id, sender) - return key in self._parked or key in self._inflight + if key in self._inflight: + return True + cutoff = time.monotonic() - CLAIM_WINDOW_SECONDS + return any(e[0] >= cutoff for e in self._parked.get(key, ())) + + def mark(self) -> int: + """Arrival limit for a bare mention: only voices that began before this may be claimed.""" + return self._next_seq def begin(self, room_id: str, sender: str) -> VoiceGate: """Mark a parkable voice as being gated. Call before the first await; always pair with ``release`` (idempotent, so it may run early and again in a ``finally``).""" - gate = VoiceGate(next(self._seq)) + gate = VoiceGate(self._next_seq) + self._next_seq += 1 # unbounded Python int: never wraps self._inflight.setdefault((room_id, sender), []).append(gate) return gate @@ -77,7 +87,7 @@ class ParkedVoices: gates.remove(gate) if not gates: # no older voice can park any more del self._inflight[key] - self._newest.pop(key, None) + self._floor.pop(key, None) async def settle(self, room_id: str, sender: str) -> None: """Wait (bounded) for every concurrently gated voice from this sender to park or drop.""" @@ -93,15 +103,28 @@ class ParkedVoices: def park(self, room_id: str, sender: str, gate: VoiceGate, event_id: str, content: dict, relates_to: dict) -> None: key = (room_id, sender) - if gate.seq < self._newest.get(key, -1): - return # a newer voice from this sender already parked (and maybe was claimed) + if gate.seq <= self._floor.get(key, -1): + return # a newer voice from this sender was already claimed self._prune() - self._parked[key] = (time.monotonic(), gate.seq, (event_id, content, relates_to)) - self._newest[key] = gate.seq + entries = self._parked.setdefault(key, []) + entries.append((time.monotonic(), gate.seq, (event_id, content, relates_to))) + entries.sort(key=lambda e: e[1]) + del entries[:-MAX_PARKED_PER_SENDER] - def claim(self, room_id: str, sender: str) -> Optional[ParkedVoice]: - """Pop the sender's parked voice for this room (the caller re-dispatches it with - ``mention_claimed=True`` so it passes the mention gate without re-parking).""" + def claim(self, room_id: str, sender: str, before: int) -> Optional[ParkedVoice]: + """Pop the sender's newest parked voice for this room that began before ``before`` + (``mark()`` taken when the bare mention arrived); older ones are dropped, later voices + stay for their own mention. The caller re-dispatches it with ``mention_claimed=True``.""" self._prune() - entry = self._parked.pop((room_id, sender), None) - return entry[2] if entry else None + key = (room_id, sender) + entries = self._parked.get(key, []) + idx = max((i for i, e in enumerate(entries) if e[1] < before), default=None) + if idx is None: + return None + claimed = entries[idx] + del entries[:idx + 1] + if not entries: + del self._parked[key] + if key in self._inflight: # a late older voice must not park after this claim + self._floor[key] = max(self._floor.get(key, -1), claimed[1]) + return claimed[2] diff --git a/tests/gateway/test_matrix_mention.py b/tests/gateway/test_matrix_mention.py index bc24959e2c..534fc7ddde 100644 --- a/tests/gateway/test_matrix_mention.py +++ b/tests/gateway/test_matrix_mention.py @@ -248,13 +248,16 @@ async def test_bare_mention_passes_empty_string(monkeypatch): ("!room2:example.org", "@hermes:example.org", False, False), ("!room1:example.org", "@hermes:example.org hi", False, False), ("!room1:example.org", "@hermes:example.org", True, True), + ("!room1:example.org", "@hermes:example.org", True, "two_voices"), ]) async def test_bare_mention_claims_parked_voice_only_in_same_room( monkeypatch, mention_room, mention_body, claims, same_sync_batch): """An unmentioned MSC3245 voice (empty m.mentions) is answered by the sender's bare @mention typed right after it in the SAME room; a bare mention in another room never pulls it across, and a mention carrying text is answered as that text. mautrix runs one /sync batch's events as - concurrent tasks, so the claim must also win while the voice still awaits a room-identity fetch.""" + concurrent tasks, so the claim must also win while the voice still awaits a room-identity fetch. + ``two_voices``: batch [voice (slow gate), mention, voice2 (fast)] then mention2 -- each mention + answers the voice sent before it, even though voice2 parks first, and nothing stays parked.""" import asyncio monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False) @@ -272,12 +275,26 @@ async def test_bare_mention_claims_parked_voice_only_in_same_room( if same_sync_batch: resolve_identity = adapter._resolve_room_identity + delays = [0.1] if same_sync_batch == "two_voices" else [] async def slow_identity(room_id): # stale 60s cache -> homeserver round-trip - await asyncio.sleep(0.01) + await asyncio.sleep(delays.pop(0) if delays else 0.01) return await resolve_identity(room_id) adapter._resolve_room_identity = slow_identity - await asyncio.gather(adapter._on_room_message(voice), adapter._on_room_message(mention)) + batch = [voice, mention] + if same_sync_batch == "two_voices": + voice2 = _make_event("voice message", event_id="$voice2") + voice2.content.update({k: voice.content[k] for k in ( + "msgtype", "url", "info", "org.matrix.msc3245.voice", "m.mentions")}) + batch.append(voice2) + await asyncio.gather(*(adapter._on_room_message(e) for e in batch)) + if same_sync_batch == "two_voices": + await adapter._on_room_message(_make_event( + "@hermes:example.org", event_id="$text2", mention_user_ids=["@hermes:example.org"])) + dispatched = [m.args[0].message_id for m in adapter.handle_message.await_args_list] + assert dispatched == ["$voice", "$voice2"] + assert not adapter._parked_voices._parked and not adapter._parked_voices._inflight + return else: await adapter._on_room_message(voice) adapter.handle_message.assert_not_awaited()