fix(matrix): claim the newest voice sent before the bare mention, not after it

The r3 fold made park refuse an older voice once a newer one from the same
sender had parked. With one /sync batch holding [v1 (slow gate), m1 bare
mention, v2 (fast)], v2 parked first, v1 was then refused, and m1 claimed
v2 -- a voice sent after the mention. v1 was lost and v2's own mention was
answered as empty text.

The bare mention now takes an arrival limit (ParkedVoices.mark) before it
settles, and claim pops the newest parked voice that began before that
limit, dropping older ones. park no longer refuses by arrival; parked
voices are kept per sender ordered by seq (bounded to 4). A claim made
while gates are still in flight records a floor, so a late older voice
(seq <= claimed) cannot park and outlive the claim; the floor clears when
the sender's in-flight list empties. One answer per bare mention, newest
before the mention wins, and the r3 orphan case still leaves nothing
parked.

Also: pending() ignores entries past CLAIM_WINDOW_SECONDS so an expired
voice no longer sends every later text through the mention regexes; the
m.thread root lookup is one _thread_root helper used by both the park
decision and its pre-check; voice_gate is annotated.

The kept test gains a [voice slow, mention, voice2 fast] + mention2 row:
red on a932dc031c (['$voice2', '$text2']) and on f1ea71de7c
(['$voice', '$text2']).

Co-authored-by: miregal89 <142085869+miregal89@users.noreply.github.com>

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)

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]

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()