# NousResearch/hermes-agent · Matrix gateway sync, auth, and messaging bug fixes
## Open now
- Nothing open on GitHub.
A collection of bugs and fixes related to the Matrix gateway integration, covering issues with the client syncer not starting, inbound messages not being received, E2EE support, auto-join on DM invites, and various session/authentication problems.
## TL;DR
- 32 member(s): 0 open issue(s), 10 closed issue(s), 0 open PR(s), 7 merged PR(s), 15 PR(s) closed without merging.
- A fix was merged to main in PR #7881. Verify it addresses your case.
- **Recommended:** read PR #7881 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.
## Maintainer resolution verdict
- Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-7981,pr-10860,pr-27330,pr-54031,pr-56345`. This is distinct from the fix status above. Weigh it yourself.
## How this cluster was grouped
- Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text.
- Members outside that structural core may belong to a different underlying problem.
- Pairwise link density `0.06`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked.
## Closed / duplicate issues
- #5819: Bot connects and syncs old messages, but silently ignores all new messages (no log output)
- #7914: Matrix gateway can send messages but doesn't receive/respond to incoming messages
- #7952: \[Bug\]: Switch to mautrix-python breaks matrix gateway integration #2
- #7966: gateway startup session suspension compares datetime to float
- #8174: \[Bug\]: Switch to mautrix-python breaks matrix gateway integration for encrypted communication #3
- #10094: \[Bug\]: Matrix bot dm cannot auto join
- #10725: fix(matrix): register MembershipEventDispatcher for invite auto-join
- #12614: \[Bug\]: Bot does not receive messages on fresh Matrix setup (no inbound events, sync stalls)
- #46142: \[Bug\]: Matrix gateway — mautrix migration (PR #18505) breaks inbound message dispatch on Tuwunel homeserver
- #46310: \[Bug\]: send_message Matrix (media) path reconnects + re-inits E2EE per message — exhausts recipient one-time keys and silently drops messages under burst
## Prior & competing PRs
- **#7881: fix: resolve three high-impact community bugs (#5819, #6893, #3388)**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `gateway/platforms/matrix.py`, `run_agent.py`, `tests/gateway/test_matrix.py`, `tests/run_agent/test_run_agent_codex_responses.py`, `website/docs/user-guide/messaging/feishu.md`
- fix hunks:
- `gateway/platforms/matrix.py`
```diff
@@ -418,6 +418,11 @@ async def connect(self) -> bool:
if isinstance(sync_data, dict):
rooms_join = sync_data.get("rooms", {}).get("join", {})
self._joined_rooms = set(rooms_join.keys())
+ # Store the next_batch token so incremental syncs start
+ # from where the initial sync left off.
+ nb = sync_data.get("next_batch")
+ if nb:
+ await client.sync_store.put_next_batch(nb)
logger.info(
"Matrix: initial sync complete, joined %d rooms",
len(self._joined_rooms),
```
```diff
@@ -809,19 +814,40 @@ async def _send_local_file(
async def _sync_loop(self) -> None:
"""Continuously sync with the homeserver."""
+ client = self._client
+ # Resume from the token stored during the initial sync.
+ next_batch = await client.sync_store.get_next_batch()
while not self._closing:
try:
- sync_data = await self._client.sync(timeout=30000)
+ sync_data = await client.sync(
+ since=next_batch, timeout=30000,
+ )
if isinstance(sync_data, dict):
# Update joined rooms from sync response.
rooms_join = sync_data.get("rooms", {}).get("join", {})
if rooms_join:
self._joined_rooms.update(rooms_join.keys())
+ # Advance the sync token so the next request is
+ # incremental instead of a full initial sync.
+ nb = sync_data.get("next_batch")
+ if nb:
+ next_batch = nb
+ await client.sync_store.put_next_batch(nb)
+
+ # Dispatch events to registered handlers so that
+ # _on_room_message / _on_reaction / _on_invite fire.
+ try:
+ tasks = client.handle_sync(sync_data)
+ if tasks:
+ await asyncio.gather(*tasks)
+ except Exception as exc:
+ logger.warning("Matrix: sync event dispatch error: %s", exc)
+
# Share keys periodically if E2EE is enabled.
- if self._encryption and getattr(self._client, "crypto", None):
+ if self._encryption and getattr(client, "crypto", None):
try:
- await self._client.crypto.share_keys()
+ await client.crypto.share_keys()
except Exception as exc:
logger.warning("Matrix: E2EE key share failed: %s", exc)
```
- `run_agent.py`
```diff
@@ -700,10 +700,14 @@ def __init__(
except Exception:
pass
- # Direct OpenAI sessions use the Responses API path. GPT-5.x tool
- # calls with reasoning are rejected on /v1/chat/completions, and
- # Hermes is a tool-using client by default.
- if self.api_mode == "chat_completions" and self._is_direct_openai_url():
+ # GPT-5.x models require the Responses API path — they are rejected
+ # on /v1/chat/completions by both OpenAI and OpenRouter. Also
+ # auto-upgrade for direct OpenAI URLs (api.openai.com) since all
+ # newer tool-calling models prefer Responses there.
+ if self.api_mode == "chat_completions" and (
+ self._is_direct_openai_url()
+ or self._model_requires_responses_api(self.model)
+ ):
self.api_mode = "codex_responses"
# Pre-warm OpenRouter model metadata cache in a background thread.
```
```diff
@@ -1702,6 +1706,21 @@ def _is_openrouter_url(self) -> bool:
"""Return True when the base URL targets OpenRouter."""
return "openrouter" in self._base_url_lower
+ @staticmethod
+ def _model_requires_responses_api(model: str) -> bool:
+ """Return True for models that require the Responses API path.
+
+ GPT-5.x models are rejected on /v1/chat/completions by both
+ OpenAI and OpenRouter (error: ``unsupported_api_for_model``).
+ Detect these so the correct api_mode is set regardless of
+ which provider is serving the model.
+ """
+ m = model.lower()
+ # Strip vendor prefix (e.g. "openai/gpt-5.4" → "gpt-5.4")
+ if "/" in m:
+ m = m.rsplit("/", 1)[-1]
+ return m.startswith("gpt-5")
+
def _max_tokens_param(self, value: int) -> dict:
"""Return the correct max tokens kwarg for the current provider.
```
```diff
@@ -5251,7 +5270,7 @@ def _try_activate_fallback(self) -> bool:
except Exception:
pass
- # Determine api_mode from provider / base URL
+ # Determine api_mode from provider / base URL / model
fb_api_mode = "chat_completions"
fb_base_url = str(fb_client.base_url)
if fb_provider == "openai-codex":
```
```diff
@@ -5260,6 +5279,10 @@ def _try_activate_fallback(self) -> bool:
fb_api_mode = "anthropic_messages"
elif self._is_direct_openai_url(fb_base_url):
fb_api_mode = "codex_responses"
+ elif self._model_requires_responses_api(fb_model):
+ # GPT-5.x models need Responses API on every provider
+ # (OpenRouter, Copilot, direct OpenAI, etc.)
+ fb_api_mode = "codex_responses"
old_model = self.model
self.model = fb_model
```
```diff
@@ -5348,8 +5371,8 @@ def _restore_primary_runtime(self) -> bool:
to the fallback provider for every subsequent turn. Calling this at
the top of ``run_conversation()`` makes fallback turn-scoped.
- The gateway creates a fresh agent per message so this is a no-op
- there (``_fallback_activated`` is always False at turn start).
+ The gateway caches agents across messages (``_agent_cache`` in
+ ``gateway/run.py``), so this restoration IS needed there too.
"""
if not self._fallback_activated:
return False
```
- `tests/gateway/test_matrix.py`
```diff
@@ -1041,20 +1041,28 @@ async def _sync_once(**kwargs):
call_count += 1
if call_count >= 1:
adapter._closing = True
- return {"rooms": {"join": {"!room:example.org": {}}}}
+ return {"rooms": {"join": {"!room:example.org": {}}}, "next_batch": "s1234"}
mock_crypto = MagicMock()
mock_crypto.share_keys = AsyncMock()
+ mock_sync_store = MagicMock()
+ mock_sync_store.get_next_batch = AsyncMock(return_value=None)
+ mock_sync_store.put_next_batch = AsyncMock()
+
fake_client = MagicMock()
fake_client.sync = AsyncMock(side_effect=_sync_once)
fake_client.crypto = mock_crypto
+ fake_client.sync_store = mock_sync_store
+ fake_client.handle_sync = MagicMock(return_value=[])
adapter._client = fake_client
await adapter._sync_loop()
fake_client.sync.assert_awaited_once()
mock_crypto.share_keys.assert_awaited_once()
+ fake_client.handle_sync.assert_called_once()
+ mock_sync_store.put_next_batch.assert_awaited_once_with("s1234")
class TestMatrixEncryptedSendFallback:
```
- `tests/run_agent/test_run_agent_codex_responses.py`
```diff
@@ -222,6 +222,12 @@ def test_api_mode_normalizes_provider_case(monkeypatch):
def test_api_mode_respects_explicit_openrouter_provider_over_codex_url(monkeypatch):
+ """GPT-5.x models need codex_responses even on OpenRouter.
+
+ OpenRouter rejects GPT-5 models on /v1/chat/completions with
+ ``unsupported_api_for_model``. The model-level check overrides
+ the provider default.
+ """
_patch_agent_bootstrap(monkeypatch)
agent = run_agent.AIAgent(
model="gpt-5-codex",
```
```diff
@@ -233,7 +239,7 @@ def test_api_mode_respects_explicit_openrouter_provider_over_codex_url(monkeypat
skip_context_files=True,
skip_memory=True,
)
- assert agent.api_mode == "chat_completions"
+ assert agent.api_mode == "codex_responses"
assert agent.provider == "openrouter"
```
- `website/docs/user-guide/messaging/feishu.md`
```diff
@@ -212,7 +212,24 @@ When users click buttons or interact with interactive cards sent by the bot, the
Card action events are dispatched with `MessageType.COMMAND`, so they flow through the normal command processing pipeline.
-To use this feature, enable the **Interactive Card** event in your Feishu app's event subscriptions (`card.action.trigger`).
+This is also how **command approval** works — when the agent needs to run a dangerous command, it sends an interactive card with Allow Once / Session / Always / Deny buttons. The user clicks a button, and the card action callback delivers the approval decision back to the agent.
+
+### Required Feishu App Configuration
+
+Interactive cards require **three** configuration steps in the Feishu Developer Console. Missing any of them causes error **200340** when users click card buttons.
+
+1. **Subscribe to the card action event:**
+ In **Event Subscriptions**, add `card.action.trigger` to your subscribed events.
+
+2. **Enable the Interactive Card capability:**
+ In **App Features > Bot**, ensure the **Interactive Card** toggle is enabled. This tells Feishu that your app can receive card action callbacks.
+
+3. **Configure the Card Request URL (webhook mode only):**
+ In **App Features > Bot > Message Card Request URL**, set the URL to the same endpoint as your event webhook (e.g. `https://your-server:8765/feishu/webhook`). In WebSocket mode this is handled automatically by the SDK.
+
+:::warning
+Without all three steps, Feishu will successfully *send* interactive cards (sending only requires `im:message:send` permission), but clicking any button will return error 200340. The card appears to work — the error only surfaces when a user interacts with it.
+:::
## Media Support
```
```diff
@@ -412,6 +429,7 @@ WebSocket and per-group ACL settings are configured via `config.yaml` under `pla
| Post messages show as plain text | The Feishu API rejected the post payload; this is normal fallback behavior. Check logs for details. |
| Images/files not received by bot | Grant `im:message` and `im:resource` permission scopes to your Feishu app |
| Bot identity not auto-detected | Grant `admin:app.info:readonly` scope, or set `FEISHU_BOT_OPEN_ID` / `FEISHU_BOT_NAME` manually |
+| Error 200340 when clicking approval buttons | Enable **Interactive Card** capability and configure **Card Request URL** in the Feishu Developer Console. See [Required Feishu App Configuration](#required-feishu-app-configuration) above. |
| `Webhook rate limit exceeded` | More than 120 requests/minute from the same IP. This is usually a misconfiguration or loop. |
## Toolset
```
- **#7981: fix(matrix): replace pickle crypto store with SQLite, fix E2EE decryption**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `gateway/platforms/matrix.py`, `gateway/session.py`, `pyproject.toml`, `tests/gateway/test_matrix.py`, `website/docs/user-guide/messaging/matrix.md`
- fix hunks:
- `gateway/platforms/matrix.py`
```diff
@@ -104,7 +104,7 @@ class _TrustStateStub: # type: ignore[no-redef]
# Uses get_hermes_home() so each profile gets its own Matrix store.
from hermes_constants import get_hermes_dir as _get_hermes_dir
_STORE_DIR = _get_hermes_dir("platforms/matrix/store", "matrix/store")
-_CRYPTO_PICKLE_PATH = _STORE_DIR / "crypto_store.pickle"
+_CRYPTO_DB_PATH = _STORE_DIR / "crypto.db"
# Grace period: ignore messages older than this many seconds before startup.
_STARTUP_GRACE_SECONDS = 5
```
```diff
@@ -165,6 +165,33 @@ def check_matrix_requirements() -> bool:
return True
+class _CryptoStateStore:
+ """Adapter that satisfies the mautrix crypto StateStore interface.
+
+ OlmMachine requires a StateStore with ``is_encrypted``,
+ ``get_encryption_info``, and ``find_shared_rooms``. The basic
+ ``MemoryStateStore`` from ``mautrix.client`` doesn't implement these,
+ so we provide simple implementations that consult the client's room
+ state.
+ """
+
+ def __init__(self, client_state_store: Any, joined_rooms: set):
+ self._ss = client_state_store
+ self._joined_rooms = joined_rooms
+
+ async def is_encrypted(self, room_id: str) -> bool:
+ return (await self.get_encryption_info(room_id)) is not None
+
+ async def get_encryption_info(self, room_id: str):
+ if hasattr(self._ss, "get_encryption_info"):
+ return await self._ss.get_encryption_info(room_id)
+ return None
+
+ async def find_shared_rooms(self, user_id: str) -> list:
+ # Return all joined rooms — simple but correct for a single-user bot.
+ return list(self._joined_rooms)
+
+
class MatrixAdapter(BasePlatformAdapter):
"""Gateway adapter for Matrix (any homeserver)."""
```
```diff
@@ -199,6 +226,7 @@ def __init__(self, config: PlatformConfig):
)
self._client: Any = None # mautrix.client.Client
+ self._crypto_db: Any = None # mautrix.util.async_db.Database
self._sync_task: Optional[asyncio.Task] = None
self._closing = False
self._startup_ts: float = 0.0
```
```diff
@@ -252,6 +280,92 @@ def _is_duplicate_event(self, event_id) -> bool:
self._processed_events_set.add(event_id)
return False
+ # ------------------------------------------------------------------
+ # E2EE helpers
+ # ------------------------------------------------------------------
+
+ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
+ """Verify our device keys are on the homeserver after loading crypto state.
+
+ Returns True if keys are valid or were successfully re-uploaded.
+ Returns False if verification fails (caller should refuse E2EE).
+ """
+ try:
+ resp = await client.query_keys({client.mxid: [client.device_id]})
+ except Exception as exc:
+ logger.error(
+ "Matrix: cannot verify device keys on server: %s — refusing E2EE", exc,
+ )
+ return False
+
+ # query_keys returns typed objects (QueryKeysResponse, DeviceKeys
+ # with KeyID keys). Normalise to plain strings for comparison.
+ device_keys_map = getattr(resp, "device_keys", {}) or {}
+ our_user_devices = device_keys_map.get(str(client.mxid)) or {}
+ our_keys = our_user_devices.get(str(client.device_id))
+
+ if not our_keys:
+ logger.warning("Matrix: device keys missing from server — re-uploading")
+ olm.account.shared = False
+ try:
+ await olm.share_keys()
+ except Exception as exc:
+ logger.error("Matrix: failed to re-upload device keys: %s", exc)
+ return False
+ return True
+
+ # DeviceKeys.keys is a dict[KeyID, str]. Iterate to find the
+ # ed25519 key rather than constructing a KeyID for lookup.
+ server_ed25519 = None
+ keys_dict = getattr(our_keys, "keys", {}) or {}
+ for key_id, key_value in keys_dict.items():
+ if str(key_id).startswith("ed25519:"):
+ server_ed25519 = str(key_value)
+ break
+ local_ed25519 = olm.account.identity_keys.get("ed25519")
+
+ if server_ed25519 != local_ed25519:
+ if olm.account.shared:
+ # Restored account from DB but server has different keys — corrupted state.
+ logger.error(
+ "Matrix: server has different identity keys for device %s — "
+ "local crypto state is stale. Delete %s and restart.",
+ client.device_id,
+ _CRYPTO_DB_PATH,
+ )
+ return False
+
+ # Fresh account (never uploaded). Server has stale keys from a
+ # previous installation. Try to delete the old device and re-upload.
+ logger.warning(
+ "Matrix: server has stale keys for device %s — attempting re-upload",
+ client.device_id,
+ )
+ try:
+ await client.api.request(
+ client.api.Method.DELETE
+ if hasattr(client.api, "Method")
+ else "DELETE",
+ f"/_matrix/client/v3/devices/{client.device_id}",
+ )
+ logger.info("Matrix: deleted stale device %s from server", client.device_id)
+ except Exception:
+ # Device deletion often requires UIA or may simply not be
+ # permitted — that's fine, share_keys will try to overwrite.
+ pass
+ try:
+ await olm.share_keys()
+ except Exception as exc:
+ logger.error(
+ "Matrix: cannot upload device keys for %s: %s. "
+ "Try generating a new access token to get a fresh device.",
+ client.device_id,
+ exc,
+ )
+ return False
+
+ return True
+
# ------------------------------------------------------------------
# Required overrides
# ------------------------------------------------------------------
```
```diff
@@ -350,54 +464,54 @@ async def connect(self) -> bool:
return False
try:
from mautrix.crypto import OlmMachine
- from mautrix.crypto.store import MemoryCryptoStore
+ from mautrix.crypto.store.asyncpg import PgCryptoStore
+ from mautrix.util.async_db import Database
+
+ _STORE_DIR.mkdir(parents=True, exist_ok=True)
+
+ # Remove legacy pickle file from pre-SQLite era.
+ legacy_pickle = _STORE_DIR / "crypto_store.pickle"
+ if legacy_pickle.exists():
+ logger.info("Matrix: removing legacy crypto_store.pickle (migrated to SQLite)")
+ legacy_pickle.unlink()
+
+ # Open SQLite-backed crypto store.
+ crypto_db = Database.create(
+ f"sqlite:///{_CRYPTO_DB_PATH}",
+ upgrade_table=PgCryptoStore.upgrade_table,
+ )
+ await crypto_db.start()
+ self._crypto_db = crypto_db
- # account_id and pickle_key are required by mautrix ≥0.21.
- # Use the Matrix user ID as account_id for stable identity.
- # pickle_key secures in-memory serialisation; derive from
- # the same user_id:device_id pair used for the on-disk HMAC.
_acct_id = self._user_id or "hermes"
- _pickle_key = f"{_acct_id}:{self._device_id}"
- crypto_store = MemoryCryptoStore(
+ _pickle_key = f"{_acct_id}:{self._device_id or 'default'}"
+ crypto_store = PgCryptoStore(
account_id=_acct_id,
pickle_key=_pickle_key,
+ db=crypto_db,
)
+ await crypto_store.open()
- # Restore persisted crypto state from a previous run.
- # Uses HMAC to verify integrity before unpickling.
- pickle_path = _CRYPTO_PICKLE_PATH
- if pickle_path.exists():
- try:
- import hashlib, hmac, pickle
- raw = pickle_path.read_bytes()
- # Format: 32-byte HMAC-SHA256 signature + pickle data.
- if len(raw) > 32:
- sig, payload = raw[:32], raw[32:]
- # Key is derived from the device_id + user_id (stable per install).
- hmac_key = f"{self._user_id}:{self._device_id}".encode()
- expected = hmac.new(hmac_key, payload, hashlib.sha256).digest()
- if hmac.compare_digest(sig, expected):
- saved = pickle.loads(payload) # noqa: S301
- if isinstance(saved, MemoryCryptoStore):
- crypto_store = saved
- logger.info("Matrix: restored E2EE crypto store from %s", pickle_path)
- else:
- logger.warning("Matrix: crypto store HMAC mismatch — ignoring stale/tampered file")
- except Exception as exc:
- logger.warning("Matrix: could not restore crypto store: %s", exc)
-
- olm = OlmMachine(client, crypto_store, state_store)
+ crypto_state = _CryptoStateStore(state_store, self._joined_rooms)
+ olm = OlmMachine(client, crypto_store, crypto_state)
- # Set trust policy: accept unverified devices so senders
- # share Megolm session keys with us automatically.
+ # Accept unverified devices so senders share Megolm
+ # session keys with us automatically.
olm.share_keys_min_trust = TrustState.UNVERIFIED
olm.send_keys_min_trust = TrustState.UNVERIFIED
await olm.load()
+
+ # Verify our device keys are still on the homeserver.
+ if not await self._verify_device_keys_on_server(client, olm):
+ await crypto_db.stop()
+ await api.session.close()
+ return False
+
client.crypto = olm
logger.info(
"Matrix: E2EE enabled (store: %s%s)",
- str(_STORE_DIR),
+ str(_CRYPTO_DB_PATH),
f", device_id={client.device_id}" if client.device_id else "",
)
except Exception as exc:
```
```diff
@@ -438,6 +552,15 @@ async def connect(self) -> bool:
)
# Build DM room cache from m.direct account data.
await self._refresh_dm_cache()
+
+ # Dispatch events from the initial sync so the OlmMachine
+ # receives to-device key shares queued while we were offline.
+ try:
+ tasks = client.handle_sync(sync_data)
+ if tasks:
+ await asyncio.gather(*tasks)
+ except Exception as exc:
+ logger.warning("Matrix: initial sync event dispatch error: %s", exc)
else:
logger.warning("Matrix: initial sync returned unexpected type %s", type(sync_data).__name__)
except Exception as exc:
```
```diff
@@ -466,21 +589,12 @@ async def disconnect(self) -> None:
except (asyncio.CancelledError, Exception):
pass
- # Persist E2EE crypto store before closing so the next restart
- # can decrypt events using sessions from this run.
- if self._client and self._encryption and getattr(self._client, "crypto", None):
+ # Close the SQLite crypto store database.
+ if hasattr(self, "_crypto_db") and self._crypto_db:
try:
- import hashlib, hmac, pickle
- crypto_store = self._client.crypto.crypto_store
- _STORE_DIR.mkdir(parents=True, exist_ok=True)
- pickle_path = _CRYPTO_PICKLE_PATH
- payload = pickle.dumps(crypto_store)
- hmac_key = f"{self._user_id}:{self._device_id}".encode()
- sig = hmac.new(hmac_key, payload, hashlib.sha256).digest()
- pickle_path.write_bytes(sig + payload)
- logger.info("Matrix: persisted E2EE crypto store to %s", pickle_path)
+ await self._crypto_db.stop()
except Exception as exc:
- logger.debug("Matrix: could not persist crypto store on disconnect: %s", exc)
+ logger.debug("Matrix: could not close crypto DB on disconnect: %s", exc)
if self._client:
try:
```
```diff
@@ -853,13 +967,6 @@ async def _sync_loop(self) -> None:
except Exception as exc:
logger.warning("Matrix: sync event dispatch error: %s", exc)
- # Share keys periodically if E2EE is enabled.
- if self._encryption and getattr(client, "crypto", None):
- try:
- await client.crypto.share_keys()
- except Exception as exc:
- logger.warning("Matrix: E2EE key share failed: %s", exc)
-
# Retry any buffered undecrypted events.
if self._pending_megolm:
await self._retry_pending_decryptions()
```
- `gateway/session.py`
```diff
@@ -807,9 +807,9 @@ def suspend_recently_active(self, max_age_seconds: int = 120) -> int:
to avoid resetting long-idle sessions that are harmless to resume.
Returns the number of sessions that were suspended.
"""
- import time as _time
+ from datetime import timedelta
- cutoff = _time.time() - max_age_seconds
+ cutoff = _now() - timedelta(seconds=max_age_seconds)
count = 0
with self._lock:
self._ensure_loaded_locked()
```
- `pyproject.toml`
```diff
@@ -43,7 +43,7 @@ dev = ["debugpy>=1.8.0,<2", "pytest>=9.0.2,<10", "pytest-asyncio>=1.3.0,<2", "py
messaging = ["python-telegram-bot[webhooks]>=22.6,<23", "discord.py[voice]>=2.7.1,<3", "aiohttp>=3.13.3,<4", "slack-bolt>=1.18.0,<2", "slack-sdk>=3.27.0,<4"]
cron = ["croniter>=6.0.0,<7"]
slack = ["slack-bolt>=1.18.0,<2", "slack-sdk>=3.27.0,<4"]
-matrix = ["mautrix[encryption]>=0.20,<1", "Markdown>=3.6,<4"]
+matrix = ["mautrix[encryption]>=0.20,<1", "Markdown>=3.6,<4", "aiosqlite>=0.20", "asyncpg>=0.29"]
cli = ["simple-term-menu>=1.0,<2"]
tts-premium = ["elevenlabs>=1.0,<2"]
voice = [
```
- `tests/gateway/test_matrix.py`
```diff
@@ -157,12 +157,44 @@ async def decrypt_megolm_event(self, event):
mautrix_crypto_store = types.ModuleType("mautrix.crypto.store")
class MemoryCryptoStore:
- def __init__(self, account_id="", pickle_key=""):
+ def __init__(self, account_id="", pickle_key=""): # noqa: S301
self.account_id = account_id
self.pickle_key = pickle_key
mautrix_crypto_store.MemoryCryptoStore = MemoryCryptoStore
+ # --- mautrix.crypto.store.asyncpg ---
+ mautrix_crypto_store_asyncpg = types.ModuleType("mautrix.crypto.store.asyncpg")
+
+ class PgCryptoStore:
+ upgrade_table = MagicMock()
+
+ def __init__(self, account_id="", pickle_key="", db=None): # noqa: S301
+ self.account_id = account_id
+ self.pickle_key = pickle_key
+ self.db = db
+
+ async def open(self):
+ pass
+
+ mautrix_crypto_store_asyncpg.PgCryptoStore = PgCryptoStore
+
+ # --- mautrix.util ---
+ mautrix_util = types.ModuleType("mautrix.util")
+
+ # --- mautrix.util.async_db ---
+ mautrix_util_async_db = types.ModuleType("mautrix.util.async_db")
+
+ class Database:
+ @classmethod
+ def create(cls, url, upgrade_table=None):
+ db = MagicMock()
+ db.start = AsyncMock()
+ db.stop = AsyncMock()
+ return db
+
+ mautrix_util_async_db.Database = Database
+
return {
"mautrix": mautrix,
"mautrix.api": mautrix_api,
```
```diff
@@ -171,6 +203,9 @@ def __init__(self, account_id="", pickle_key=""):
"mautrix.client.state_store": mautrix_client_state_store,
"mautrix.crypto": mautrix_crypto,
"mautrix.crypto.store": mautrix_crypto_store,
+ "mautrix.crypto.store.asyncpg": mautrix_crypto_store_asyncpg,
+ "mautrix.util": mautrix_util,
+ "mautrix.util.async_db": mautrix_util_async_db,
}
```
```diff
@@ -740,6 +775,12 @@ def __init__(self, user_id, device_id):
mock_client.whoami = AsyncMock(return_value=FakeWhoamiResponse("@bot:example.org", "DEV123"))
mock_client.sync = AsyncMock(return_value={"rooms": {"join": {"!room:server": {}}}})
mock_client.add_event_handler = MagicMock()
+ mock_client.handle_sync = MagicMock(return_value=[])
+ mock_client.query_keys = AsyncMock(return_value={
+ "device_keys": {"@bot:example.org": {"DEV123": {
+ "keys": {"ed25519:DEV123": "fake_ed25519_key"},
+ }}},
+ })
mock_client.api = MagicMock()
mock_client.api.token = "[redacted:secret-assignment]"
mock_client.api.session = MagicMock()
```
```diff
@@ -751,6 +792,8 @@ def __init__(self, user_id, device_id):
mock_olm.share_keys = AsyncMock()
mock_olm.share_keys_min_trust = None
mock_olm.send_keys_min_trust = None
+ mock_olm.account = MagicMock()
+ mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"}
# Patch Client constructor to return our mock
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
```
```diff
@@ -924,6 +967,12 @@ async def test_connect_uses_configured_device_id_over_whoami(self):
mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="WHOAMI_DEV"))
mock_client.sync = AsyncMock(return_value={"rooms": {"join": {"!room:server": {}}}})
mock_client.add_event_handler = MagicMock()
+ mock_client.handle_sync = MagicMock(return_value=[])
+ mock_client.query_keys = AsyncMock(return_value={
+ "device_keys": {"@bot:example.org": {"MY_STABLE_DEVICE": {
+ "keys": {"ed25519:MY_STABLE_DEVICE": "fake_ed25519_key"},
+ }}},
+ })
mock_client.api = MagicMock()
mock_client.api.token = "[redacted:secret-assignment]"
mock_client.api.session = MagicMock()
```
```diff
@@ -934,6 +983,8 @@ async def test_connect_uses_configured_device_id_over_whoami(self):
mock_olm.share_keys = AsyncMock()
mock_olm.share_keys_min_trust = None
mock_olm.send_keys_min_trust = None
+ mock_olm.account = MagicMock()
+ mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"}
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
```
```diff
@@ -1030,8 +1081,8 @@ def test_device_id_not_set_when_env_empty(self, monkeypatch):
class TestMatrixSyncLoop:
@pytest.mark.asyncio
- async def test_sync_loop_shares_keys_when_encryption_enabled(self):
- """_sync_loop should call crypto.share_keys() after each sync."""
+ async def test_sync_loop_dispatches_events_and_stores_token(self):
+ """_sync_loop should call handle_sync() and persist next_batch."""
adapter = _make_adapter()
adapter._encryption = True
adapter._closing = False
```
```diff
@@ -1046,7 +1097,6 @@ async def _sync_once(**kwargs):
return {"rooms": {"join": {"!room:example.org": {}}}, "next_batch": "s1234"}
mock_crypto = MagicMock()
- mock_crypto.share_keys = AsyncMock()
mock_sync_store = MagicMock()
mock_sync_store.get_next_batch = AsyncMock(return_value=None)
```
```diff
@@ -1062,7 +1112,6 @@ async def _sync_once(**kwargs):
await adapter._sync_loop()
fake_client.sync.assert_awaited_once()
- mock_crypto.share_keys.assert_awaited_once()
fake_client.handle_sync.assert_called_once()
mock_sync_store.put_next_batch.assert_awaited_once_with("s1234")
```
```diff
@@ -1248,6 +1297,12 @@ async def test_connect_registers_encrypted_event_handler_when_encryption_on(self
mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123"))
mock_client.sync = AsyncMock(return_value={"rooms": {"join": {"!room:server": {}}}})
mock_client.add_event_handler = MagicMock()
+ mock_client.handle_sync = MagicMock(return_value=[])
+ mock_client.query_keys = AsyncMock(return_value={
+ "device_keys": {"@bot:example.org": {"DEV123": {
+ "keys": {"ed25519:DEV123": "fake_ed25519_key"},
+ }}},
+ })
mock_client.api = MagicMock()
mock_client.api.token = "[redacted:secret-assignment]"
mock_client.api.session = MagicMock()
```
```diff
@@ -1258,6 +1313,8 @@ async def test_connect_registers_encrypted_event_handler_when_encryption_on(self
mock_olm.share_keys = AsyncMock()
mock_olm.share_keys_min_trust = None
mock_olm.send_keys_min_trust = None
+ mock_olm.account = MagicMock()
+ mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"}
fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
```
- `website/docs/user-guide/messaging/matrix.md`
```diff
@@ -344,9 +344,79 @@ pip install 'hermes-agent[matrix]'
**Fix**:
1. Verify `libolm` is installed on your system (see the E2EE section above).
2. Make sure `MATRIX_ENCRYPTION=true` is set in your `.env`.
-3. In your Matrix client (Element), go to the bot's profile → **Sessions** → verify/trust the bot's device.
+3. In your Matrix client (Element), go to the bot's profile -> Sessions -> verify/trust the bot's device.
4. If the bot just joined an encrypted room, it can only decrypt messages sent *after* it joined. Older messages are inaccessible.
+### Upgrading from a previous version with E2EE
+
+If you previously used Hermes with `MATRIX_ENCRYPTION=true` and are upgrading to
+a version that uses the new SQLite-based crypto store, the bot's encryption
+identity has changed. Your Matrix client (Element) may cache the old device keys
+and refuse to share encryption sessions with the bot.
+
+**Symptoms**: The bot connects and shows "E2EE enabled" in the logs, but all
+messages show "could not decrypt event" and the bot never responds.
+
+**What's happening**: The old encryption state (from the previous `matrix-nio` or
+serialization-based `mautrix` backend) is incompatible with the new SQLite crypto
+store. The bot creates a fresh encryption identity, but your Matrix client still
+has the old keys cached and won't share the room's encryption session with a
+device whose keys changed. This is a Matrix security feature -- clients treat
+changed identity keys for the same device as suspicious.
+
+**Fix** (one-time migration):
+
+1. **Generate a new access token** to get a fresh device ID. The simplest way:
+
+ ```bash
+ curl -X POST https://your-server/_matrix/client/v3/login \
+ -H "Content-Type: application/json" \
+ -d '{
+ "type": "m.login.password",
+ "identifier": {"type": "m.id.user", "user": "@hermes:your-server.org"},
+ "password": "your-password",
+ "initial_device_display_name": "Hermes Agent"
+ }'
+ ```
+
+ Copy the new `access_token` and update `MATRIX_ACCESS_TOKEN` in `~/.hermes/.env`.
+
+2. **Delete old encryption state**:
+
+ ```bash
+ rm -f ~/.hermes/platforms/matrix/store/crypto.db
+ rm -f ~/.hermes/platforms/matrix/store/crypto_store.*
+ ```
+
+3. **Force your Matrix client to rotate the encryption session**. In Element,
+ open the DM room with the bot and type `/discardsession`. This forces Element
+ to create a new encryption session and share it with the bot's new device.
+
+4. **Restart the gateway**:
+
+ ```bash
+ hermes gateway run
+ ```
+
+5. **Send a new message**. The bot should decrypt and respond normally.
+
+:::note
+After migration, messages sent *before* the upgrade cannot be decrypted -- the old
+encryption keys are gone. This only affects the transition; new messages work
+normally.
+:::
+
+:::tip
+**New installations are not affected.** This migration is only needed if you had
+a working E2EE setup with a previous version of Hermes and are upgrading.
+
+**Why a new access token?** Each Matrix access token is bound to a specific device
+ID. Reusing the same device ID with new encryption keys causes other Matrix
+clients to distrust the device (they see changed identity keys as a potential
+security breach). A new access token gets a new device ID with no stale key
+history, so other clients trust it immediately.
+:::
+
### Sync issues / bot falls behind
**Cause**: Long-running tool executions can delay the sync loop, or the homeserver is slow.
```
- **#10860: fix(matrix): E2EE and migration bugfixes**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `gateway/platforms/base.py`, `gateway/platforms/matrix.py`, `gateway/run.py`, `gateway/stream_consumer.py`, `tests/gateway/test_matrix.py`, `tests/gateway/test_matrix_mention.py`, `tests/gateway/test_stream_consumer.py`, `website/docs/user-guide/messaging/matrix.md`
- fix hunks:
- `gateway/platforms/base.py`
```diff
@@ -1260,7 +1260,7 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
path = path[1:-1].strip()
path = path.lstrip("`\"'").rstrip("`\"',.;:)}]")
if path:
- media.append((path, has_voice_tag))
+ media.append((os.path.expanduser(path), has_voice_tag))
# Remove MEDIA tags from content (including surrounding quote/backtick wrappers)
if media:
```
- `gateway/platforms/matrix.py`
```diff
@@ -30,11 +30,10 @@
import os
import re
import time
+from html import escape as _html_escape
from pathlib import Path
from typing import Any, Dict, Optional, Set
-from html import escape as _html_escape
-
try:
from mautrix.types import (
ContentURI,
```
```diff
@@ -60,28 +59,33 @@ class _EventTypeStub: # type: ignore[no-redef]
REACTION = "m.reaction"
ROOM_ENCRYPTED = "m.room.encrypted"
ROOM_NAME = "m.room.name"
+
EventType = _EventTypeStub # type: ignore[misc,assignment]
class _PaginationDirectionStub: # type: ignore[no-redef]
BACKWARD = "b"
FORWARD = "f"
+
PaginationDirection = _PaginationDirectionStub # type: ignore[misc,assignment]
class _PresenceStateStub: # type: ignore[no-redef]
ONLINE = "online"
OFFLINE = "offline"
UNAVAILABLE = "unavailable"
+
PresenceState = _PresenceStateStub # type: ignore[misc,assignment]
class _RoomCreatePresetStub: # type: ignore[no-redef]
PRIVATE = "private_chat"
PUBLIC = "public_chat"
TRUSTED_PRIVATE = "trusted_private_chat"
+
RoomCreatePreset = _RoomCreatePresetStub # type: ignore[misc,assignment]
class _TrustStateStub: # type: ignore[no-redef]
UNVERIFIED = 0
VERIFIED = 1
+
TrustState = _TrustStateStub # type: ignore[misc,assignment]
from gateway.config import Platform, PlatformConfig
```
```diff
@@ -103,20 +107,16 @@ class _TrustStateStub: # type: ignore[no-redef]
# Store directory for E2EE keys and sync state.
# Uses get_hermes_home() so each profile gets its own Matrix store.
from hermes_constants import get_hermes_dir as _get_hermes_dir
+
_STORE_DIR = _get_hermes_dir("platforms/matrix/store", "matrix/store")
_CRYPTO_DB_PATH = _STORE_DIR / "crypto.db"
# Grace period: ignore messages older than this many seconds before startup.
_STARTUP_GRACE_SECONDS = 5
-# Pending undecrypted events: cap and TTL for retry buffer.
-_MAX_PENDING_EVENTS = 100
-_PENDING_EVENT_TTL = 300 # seconds — stop retrying after 5 min
-
_E2EE_INSTALL_HINT = (
- "Install with: pip install 'mautrix[encryption]' "
- "(requires libolm C library)"
+ "Install with: pip install 'mautrix[encryption]' (requires libolm C library)"
)
```
```diff
@@ -124,6 +124,7 @@ def _check_e2ee_deps() -> bool:
"""Return True if mautrix E2EE dependencies (python-olm) are available."""
try:
from mautrix.crypto import OlmMachine # noqa: F401
+
return True
except (ImportError, AttributeError):
return False
```
```diff
@@ -145,14 +146,17 @@ def check_matrix_requirements() -> bool:
import mautrix # noqa: F401
except ImportError:
logger.warning(
- "Matrix: mautrix not installed. "
- "Run: pip install 'mautrix[encryption]'"
+ "Matrix: mautrix not installed. Run: pip install 'mautrix[encryption]'"
)
return False
# If encryption is requested, verify E2EE deps are available at startup
# rather than silently degrading to plaintext-only at connect time.
- encryption_requested = os.getenv("MATRIX_ENCRYPTION", "").lower() in ("true", "1", "yes")
+ encryption_requested = os.getenv("MATRIX_ENCRYPTION", "").lower() in (
+ "true",
+ "1",
+ "yes",
+ )
if encryption_requested and not _check_e2ee_deps():
logger.error(
"Matrix: MATRIX_ENCRYPTION=true but E2EE dependencies are missing. %s. "
```
```diff
@@ -204,25 +208,21 @@ def __init__(self, config: PlatformConfig):
super().__init__(config, Platform.MATRIX)
self._homeserver: str = (
- config.extra.get("homeserver", "")
- or os.getenv("MATRIX_HOMESERVER", "")
+ config.extra.get("homeserver", "") or os.getenv("MATRIX_HOMESERVER", "")
).rstrip("/")
self._access_token: str = config.token or os.getenv("MATRIX_ACCESS_TOKEN", "")
- self._user_id: str = (
- config.extra.get("user_id", "")
- or os.getenv("MATRIX_USER_ID", "")
+ self._user_id: str = config.extra.get("user_id", "") or os.getenv(
+ "MATRIX_USER_ID", ""
)
- self._password: str = (
- config.extra.get("password", "")
- or os.getenv("MATRIX_PASSWORD", "")
+ self._password: str = config.extra.get("password", "") or os.getenv(
+ "MATRIX_PASSWORD", ""
)
self._encryption: bool = config.extra.get(
"encryption",
os.getenv("MATRIX_ENCRYPTION", "").lower() in ("true", "1", "yes"),
)
- self._device_id: str = (
- config.extra.get("device_id", "")
- or os.getenv("MATRIX_DEVICE_ID", "")
+ self._device_id: str = config.extra.get("device_id", "") or os.getenv(
+ "MATRIX_DEVICE_ID", ""
)
self._client: Any = None # mautrix.client.Client
```
```diff
@@ -237,22 +237,32 @@ def __init__(self, config: PlatformConfig):
self._joined_rooms: Set[str] = set()
# Event deduplication (bounded deque keeps newest entries)
from collections import deque
+
self._processed_events: deque = deque(maxlen=1000)
self._processed_events_set: set = set()
# Buffer for undecrypted events pending key receipt.
# Each entry: (room_id, event, timestamp)
- self._pending_megolm: list = []
# Thread participation tracking (for require_mention bypass)
self._threads = ThreadParticipationTracker("matrix")
# Mention/thread gating — parsed once from env vars.
- self._require_mention: bool = os.getenv("MATRIX_REQUIRE_MENTION", "true").lower() not in ("false", "0", "no")
+ self._require_mention: bool = os.getenv(
+ "MATRIX_REQUIRE_MENTION", "true"
+ ).lower() not in ("false", "0", "no")
free_rooms_raw = os.getenv("MATRIX_FREE_RESPONSE_ROOMS", "")
- self._free_rooms: Set[str] = {r.strip() for r in free_rooms_raw.split(",") if r.strip()}
- self._auto_thread: bool = os.getenv("MATRIX_AUTO_THREAD", "true").lower() in ("true", "1", "yes")
- self._dm_mention_threads: bool = os.getenv("MATRIX_DM_MENTION_THREADS", "false").lower() in ("true", "1", "yes")
+ self._free_rooms: Set[str] = {
+ r.strip() for r in free_rooms_raw.split(",") if r.strip()
+ }
+ self._auto_thread: bool = os.getenv("MATRIX_AUTO_THREAD", "true").lower() in (
+ "true",
+ "1",
+ "yes",
+ )
+ self._dm_mention_threads: bool = os.getenv(
+ "MATRIX_DM_MENTION_THREADS", "false"
+ ).lower() in ("true", "1", "yes")
# Reactions: configurable via MATRIX_REACTIONS (default: true).
self._reactions_enabled: bool = os.getenv(
```
```diff
@@ -262,8 +272,12 @@ def __init__(self, config: PlatformConfig):
# Text batching: merge rapid successive messages (Telegram-style).
# Matrix clients split long messages around 4000 chars.
- self._text_batch_delay_seconds = float(os.getenv("HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS", "0.6"))
- self._text_batch_split_delay_seconds = float(os.getenv("HERMES_MATRIX_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0"))
+ self._text_batch_delay_seconds = float(
+ os.getenv("HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS", "0.6")
+ )
+ self._text_batch_split_delay_seconds = float(
+ os.getenv("HERMES_MATRIX_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0")
+ )
self._pending_text_batches: Dict[str, MessageEvent] = {}
self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
```
```diff
@@ -284,6 +298,38 @@ def _is_duplicate_event(self, event_id) -> bool:
# E2EE helpers
# ------------------------------------------------------------------
+ @staticmethod
+ def _extract_server_ed25519(device_keys_obj: Any) -> Optional[str]:
+ """Extract the ed25519 identity key from a DeviceKeys object."""
+ for kid, kval in (getattr(device_keys_obj, "keys", {}) or {}).items():
+ if str(kid).startswith("ed25519:"):
+ return str(kval)
+ return None
+
+ async def _reverify_keys_after_upload(
+ self, client: Any, local_ed25519: str
+ ) -> bool:
+ """Re-query the server after share_keys() and verify our ed25519 key matches."""
+ try:
+ resp = await client.query_keys({client.mxid: [client.device_id]})
+ dk = getattr(resp, "device_keys", {}) or {}
+ ud = dk.get(str(client.mxid)) or {}
+ dev = ud.get(str(client.device_id))
+ if dev:
+ server_ed = self._extract_server_ed25519(dev)
+ if server_ed != local_ed25519:
+ logger.error(
+ "Matrix: device %s has immutable identity keys that "
+ "don't match this installation. Generate a new access "
+ "token with a fresh device.",
+ client.device_id,
+ )
+ return False
+ except Exception as exc:
+ logger.error("Matrix: post-upload key verification failed: %s", exc)
+ return False
+ return True
+
async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
"""Verify our device keys are on the homeserver after loading crypto state.
```
```diff
@@ -294,15 +340,15 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
resp = await client.query_keys({client.mxid: [client.device_id]})
except Exception as exc:
logger.error(
- "Matrix: cannot verify device keys on server: %s — refusing E2EE", exc,
+ "Matrix: cannot verify device keys on server: %s — refusing E2EE",
+ exc,
)
return False
- # query_keys returns typed objects (QueryKeysResponse, DeviceKeys
- # with KeyID keys). Normalise to plain strings for comparison.
device_keys_map = getattr(resp, "device_keys", {}) or {}
our_user_devices = device_keys_map.get(str(client.mxid)) or {}
our_keys = our_user_devices.get(str(client.device_id))
+ local_ed25519 = olm.account.identity_keys.get("ed25519")
if not our_keys:
logger.warning("Matrix: device keys missing from server — re-uploading")
```
```diff
@@ -312,21 +358,12 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
except Exception as exc:
logger.error("Matrix: failed to re-upload device keys: %s", exc)
return False
- return True
+ return await self._reverify_keys_after_upload(client, local_ed25519)
- # DeviceKeys.keys is a dict[KeyID, str]. Iterate to find the
- # ed25519 key rather than constructing a KeyID for lookup.
- server_ed25519 = None
- keys_dict = getattr(our_keys, "keys", {}) or {}
- for key_id, key_value in keys_dict.items():
- if str(key_id).startswith("ed25519:"):
- server_ed25519 = str(key_value)
- break
- local_ed25519 = olm.account.identity_keys.get("ed25519")
+ server_ed25519 = self._extract_server_ed25519(our_keys)
if server_ed25519 != local_ed25519:
if olm.account.shared:
- # Restored account from DB but server has different keys — corrupted state.
logger.error(
"Matrix: server has different identity keys for device %s — "
"local crypto state is stale. Delete %s and restart.",
```
```diff
@@ -335,8 +372,6 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
)
return False
- # Fresh account (never uploaded). Server has stale keys from a
- # previous installation. Try to delete the old device and re-upload.
logger.warning(
"Matrix: server has stale keys for device %s — attempting re-upload",
client.device_id,
```
```diff
@@ -348,10 +383,10 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
else "DELETE",
f"/_matrix/client/v3/devices/{client.device_id}",
)
- logger.info("Matrix: deleted stale device %s from server", client.device_id)
+ logger.info(
+ "Matrix: deleted stale device %s from server", client.device_id
+ )
except Exception:
- # Device deletion often requires UIA or may simply not be
- # permitted — that's fine, share_keys will try to overwrite.
pass
try:
await olm.share_keys()
```
```diff
@@ -363,6 +398,7 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
exc,
)
return False
+ return await self._reverify_keys_after_upload(client, local_ed25519)
return True
```
```diff
@@ -448,7 +484,9 @@ async def connect(self) -> bool:
await api.session.close()
return False
else:
- logger.error("Matrix: need MATRIX_ACCESS_TOKEN or MATRIX_USER_ID + MATRIX_PASSWORD")
+ logger.error(
+ "Matrix: need MATRIX_ACCESS_TOKEN or MATRIX_USER_ID + MATRIX_PASSWORD"
+ )
await api.session.close()
return False
```
```diff
@@ -472,7 +510,9 @@ async def connect(self) -> bool:
# Remove legacy pickle file from pre-SQLite era.
legacy_pickle = _STORE_DIR / "crypto_store.pickle"
if legacy_pickle.exists():
- logger.info("Matrix: removing legacy crypto_store.pickle (migrated to SQLite)")
+ logger.info(
+ "Matrix: removing legacy crypto_store.pickle (migrated to SQLite)"
+ )
legacy_pickle.unlink()
# Open SQLite-backed crypto store.
```
```diff
@@ -508,6 +548,37 @@ async def connect(self) -> bool:
await api.session.close()
return False
+ # Proactively flush one-time keys to detect stale OTK
+ # conflicts early. When crypto state is wiped but the
+ # same device ID is reused, the server may still hold OTKs
+ # signed with the old ed25519 key. Identity key re-upload
+ # succeeds but OTK uploads fail ("already exists" with
+ # mismatched signature). Peers then cannot establish Olm
+ # sessions and all new messages are undecryptable.
+ try:
+ await olm.share_keys()
+ except Exception as exc:
+ exc_str = str(exc)
+ if "already exists" in exc_str:
+ logger.error(
+ "Matrix: device %s has stale one-time keys on the "
+ "server signed with a previous identity key. "
+ "Peers cannot establish new Olm sessions with "
+ "this device. Delete the device from the "
+ "homeserver and restart, or generate a new "
+ "access token to get a fresh device ID.",
+ client.device_id,
+ )
+ await crypto_db.stop()
+ await api.session.close()
+ return False
+ # Non-OTK errors are transient (network, etc.) — log
+ # but allow startup to continue.
+ logger.warning(
+ "Matrix: share_keys() warning during startup: %s",
+ exc,
+ )
+
# Import cross-signing private keys from SSSS and self-sign
# the current device. Required after any device-key rotation
# (fresh crypto.db, share_keys re-upload) — otherwise the
```
```diff
@@ -519,7 +590,9 @@ async def connect(self) -> bool:
await olm.verify_with_recovery_key(recovery_key)
logger.info("Matrix: cross-signing verified via recovery key")
except Exception as exc:
- logger.warning("Matrix: recovery key verification failed: %s", exc)
+ logger.warning(
+ "Matrix: recovery key verification failed: %s", exc
+ )
client.crypto = olm
logger.info(
```
```diff
@@ -530,21 +603,23 @@ async def connect(self) -> bool:
except Exception as exc:
logger.error(
"Matrix: failed to create E2EE client: %s. %s",
- exc, _E2EE_INSTALL_HINT,
+ exc,
+ _E2EE_INSTALL_HINT,
)
await api.session.close()
return False
# Register event handlers.
from mautrix.client import InternalEventType as IntEvt
+ from mautrix.client.dispatcher import MembershipEventDispatcher
+
+ # Without this the INVITE handler below never fires.
+ client.add_dispatcher(MembershipEventDispatcher)
client.add_event_handler(EventType.ROOM_MESSAGE, self._on_room_message)
client.add_event_handler(EventType.REACTION, self._on_reaction)
client.add_event_handler(IntEvt.INVITE, self._on_invite)
- if self._encryption and getattr(client, "crypto", None):
- client.add_event_handler(EventType.ROOM_ENCRYPTED, self._on_encrypted_event)
-
# Initial sync to catch up, then start background sync.
self._startup_ts = time.time()
self._closing = False
```
```diff
@@ -553,7 +628,8 @@ async def connect(self) -> bool:
sync_data = await client.sync(timeout=10000, full_state=True)
if isinstance(sync_data, dict):
rooms_join = sync_data.get("rooms", {}).get("join", {})
- self._joined_rooms = set(rooms_join.keys())
+ self._joined_rooms.clear()
+ self._joined_rooms.update(rooms_join.keys())
# Store the next_batch token so incremental syncs start
# from where the initial sync left off.
nb = sync_data.get("next_batch")
```
```diff
@@ -575,7 +651,10 @@ async def connect(self) -> bool:
except Exception as exc:
logger.warning("Matrix: initial sync event dispatch error: %s", exc)
else:
- logger.warning("Matrix: initial sync returned unexpected type %s", type(sync_data).__name__)
+ logger.warning(
+ "Matrix: initial sync returned unexpected type %s",
+ type(sync_data).__name__,
+ )
except Exception as exc:
logger.warning("Matrix: initial sync error: %s", exc)
```
```diff
@@ -648,9 +727,7 @@ async def send(
# Reply-to support.
if reply_to:
- msg_content["m.relates_to"] = {
- "m.in_reply_to": {"event_id": reply_to}
- }
+ msg_content["m.relates_to"] = {"m.in_reply_to": {"event_id": reply_to}}
# Thread support: if metadata has thread_id, send as threaded reply.
thread_id = (metadata or {}).get("thread_id")
```
```diff
@@ -688,10 +765,18 @@ async def send(
timeout=45,
)
last_event_id = str(event_id)
- logger.info("Matrix: sent event %s to %s (after key share)", last_event_id, chat_id)
+ logger.info(
+ "Matrix: sent event %s to %s (after key share)",
+ last_event_id,
+ chat_id,
+ )
continue
except Exception as retry_exc:
- logger.error("Matrix: failed to send to %s after retry: %s", chat_id, retry_exc)
+ logger.error(
+ "Matrix: failed to send to %s after retry: %s",
+ chat_id,
+ retry_exc,
+ )
return SendResult(success=False, error=str(retry_exc))
logger.error("Matrix: failed to send to %s: %s", chat_id, exc)
return SendResult(success=False, error=str(exc))
```
```diff
@@ -706,7 +791,8 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
if self._client:
try:
name_evt = await self._client.get_state_event(
- RoomID(chat_id), EventType.ROOM_NAME,
+ RoomID(chat_id),
+ EventType.ROOM_NAME,
)
if name_evt and hasattr(name_evt, "name") and name_evt.name:
name = name_evt.name
```
```diff
@@ -730,13 +816,14 @@ async def send_typing(
pass
async def stop_typing(self, chat_id: str) -> None:
- """Stop the Matrix typing indicator."""
+ """Clear the typing indicator."""
if self._client:
try:
await self._client.set_typing(RoomID(chat_id), timeout=0)
except Exception:
pass
+
async def edit_message(
self, chat_id: str, message_id: str, content: str
) -> SendResult:
```
```diff
@@ -765,7 +852,9 @@ async def edit_message(
try:
event_id = await self._client.send_message_event(
- RoomID(chat_id), EventType.ROOM_MESSAGE, msg_content,
+ RoomID(chat_id),
+ EventType.ROOM_MESSAGE,
+ msg_content,
)
return SendResult(success=True, message_id=str(event_id))
except Exception as exc:
```
```diff
@@ -781,22 +870,31 @@ async def send_image(
) -> SendResult:
"""Download an image URL and upload it to Matrix."""
from tools.url_safety import is_safe_url
+
if not is_safe_url(image_url):
logger.warning("Matrix: blocked unsafe image URL (SSRF protection)")
- return await super().send_image(chat_id, image_url, caption, reply_to, metadata=metadata)
+ return await super().send_image(
+ chat_id, image_url, caption, reply_to, metadata=metadata
+ )
try:
# Try aiohttp first (always available), fall back to httpx
try:
import aiohttp as _aiohttp
+
async with _aiohttp.ClientSession(trust_env=True) as http:
- async with http.get(image_url, timeout=_aiohttp.ClientTimeout(total=30)) as resp:
+ async with http.get(
+ image_url, timeout=_aiohttp.ClientTimeout(total=30)
+ ) as resp:
resp.raise_for_status()
data = await resp.read()
ct = resp.content_type or "image/png"
- fname = image_url.rsplit("/", 1)[-1].split("?")[0] or "image.png"
+ fname = (
+ image_url.rsplit("/", 1)[-1].split("?")[0] or "image.png"
+ )
except ImportError:
import httpx
+
async with httpx.AsyncClient() as http:
resp = await http.get(image_url, follow_redirects=True, timeout=30)
resp.raise_for_status()
```
```diff
@@ -805,9 +903,13 @@ async def send_image(
fname = image_url.rsplit("/", 1)[-1].split("?")[0] or "image.png"
except Exception as exc:
logger.warning("Matrix: failed to download image %s: %s", image_url, exc)
- return await self.send(chat_id, f"{caption or ''}\n{image_url}".strip(), reply_to)
+ return await self.send(
+ chat_id, f"{caption or ''}\n{image_url}".strip(), reply_to
+ )
- return await self._upload_and_send(chat_id, data, fname, ct, "m.image", caption, reply_to, metadata)
+ return await self._upload_and_send(
+ chat_id, data, fname, ct, "m.image", caption, reply_to, metadata
+ )
async def send_image_file(
self,
```
```diff
@@ -818,7 +920,9 @@ async def send_image_file(
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Upload a local image file to Matrix."""
- return await self._send_local_file(chat_id, image_path, "m.image", caption, reply_to, metadata=metadata)
+ return await self._send_local_file(
+ chat_id, image_path, "m.image", caption, reply_to, metadata=metadata
+ )
async def send_document(
self,
```
```diff
@@ -830,7 +934,9 @@ async def send_document(
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Upload a local file as a document."""
- return await self._send_local_file(chat_id, file_path, "m.file", caption, reply_to, file_name, metadata)
+ return await self._send_local_file(
+ chat_id, file_path, "m.file", caption, reply_to, file_name, metadata
+ )
async def send_voice(
self,
```
```diff
@@ -842,8 +948,13 @@ async def send_voice(
) -> SendResult:
"""Upload an audio file as a voice message (MSC3245 native voice)."""
return await self._send_local_file(
- chat_id, audio_path, "m.audio", caption, reply_to,
- metadata=metadata, is_voice=True
+ chat_id,
+ audio_path,
+ "m.audio",
+ caption,
+ reply_to,
+ metadata=metadata,
+ is_voice=True,
)
async def send_video(
```
```diff
@@ -855,7 +966,9 @@ async def send_video(
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Upload a video file."""
- return await self._send_local_file(chat_id, video_path, "m.video", caption, reply_to, metadata=metadata)
+ return await self._send_local_file(
+ chat_id, video_path, "m.video", caption, reply_to, metadata=metadata
+ )
def format_message(self, content: str) -> str:
"""Pass-through — Matrix supports standard Markdown natively."""
```
```diff
@@ -881,12 +994,30 @@ async def _upload_and_send(
) -> SendResult:
"""Upload bytes to Matrix and send as a media message."""
+ upload_data = data
+ encrypted_file = None
+ if self._encryption and getattr(self._client, "crypto", None):
+ state_store = getattr(self._client, "state_store", None)
+ if state_store:
+ try:
+ room_encrypted = bool(await state_store.is_encrypted(RoomID(room_id)))
+ except Exception:
+ room_encrypted = False
+ if room_encrypted:
+ try:
+ from mautrix.crypto.attachments import encrypt_attachment
+ upload_data, encrypted_file = encrypt_attachment(data)
+ except Exception as exc:
+ logger.error("Matrix: attachment encryption failed: %s", exc)
+ return SendResult(success=False, error=str(exc))
+
# Upload to homeserver.
try:
mxc_url = await self._client.upload_media(
- data,
+ upload_data,
mime_type=content_type,
filename=filename,
+ size=len(upload_data),
)
except Exception as exc:
logger.error("Matrix: upload failed: %s", exc)
```
```diff
@@ -896,21 +1027,24 @@ async def _upload_and_send(
msg_content: Dict[str, Any] = {
"msgtype": msgtype,
"body": caption or filename,
- "url": str(mxc_url),
"info": {
"mimetype": content_type,
"size": len(data),
},
}
+ if encrypted_file is not None:
+ file_payload = encrypted_file.serialize()
+ file_payload["url"] = str(mxc_url)
+ msg_content["file"] = file_payload
+ else:
+ msg_content["url"] = str(mxc_url)
# Add MSC3245 voice flag for native voice messages.
if is_voice:
msg_content["org.matrix.msc3245.voice"] = {}
if reply_to:
- msg_content["m.relates_to"] = {
- "m.in_reply_to": {"event_id": reply_to}
- }
+ msg_content["m.relates_to"] = {"m.in_reply_to": {"event_id": reply_to}}
thread_id = (metadata or {}).get("thread_id")
if thread_id:
```
```diff
@@ -922,7 +1056,9 @@ async def _upload_and_send(
try:
event_id = await self._client.send_message_event(
- RoomID(room_id), EventType.ROOM_MESSAGE, msg_content,
+ RoomID(room_id),
+ EventType.ROOM_MESSAGE,
+ msg_content,
)
return SendResult(success=True, message_id=str(event_id))
except Exception as exc:
```
```diff
@@ -940,7 +1076,7 @@ async def _send_local_file(
is_voice: bool = False,
) -> SendResult:
"""Read a local file and upload it."""
- p = Path(file_path)
+ p = Path(file_path).expanduser()
if not p.exists():
return await self.send(
room_id, f"{caption or ''}\n(file not found: {file_path})", reply_to
```
```diff
@@ -950,7 +1086,9 @@ async def _send_local_file(
ct = mimetypes.guess_type(fname)[0] or "application/octet-stream"
data = p.read_bytes()
- return await self._upload_and_send(room_id, data, fname, ct, msgtype, caption, reply_to, metadata, is_voice)
+ return await self._upload_and_send(
+ room_id, data, fname, ct, msgtype, caption, reply_to, metadata, is_voice
+ )
# ------------------------------------------------------------------
# Sync loop
```
```diff
@@ -964,7 +1102,8 @@ async def _sync_loop(self) -> None:
while not self._closing:
try:
sync_data = await client.sync(
- since=next_batch, timeout=30000,
+ since=next_batch,
+ timeout=30000,
)
# nio returns SyncError objects (not exceptions) for auth
```
```diff
@@ -973,7 +1112,10 @@ async def _sync_loop(self) -> None:
if _sync_msg and isinstance(_sync_msg, str):
_lower = _sync_msg.lower()
if "m_unknown_token" in _lower or "unknown_token" in _lower:
- logger.error("Matrix: permanent auth error from sync: %s — stopping", _sync_msg)
+ logger.error(
+ "Matrix: permanent auth error from sync: %s — stopping",
+ _sync_msg,
+ )
return
if isinstance(sync_data, dict):
```
```diff
@@ -998,10 +1140,6 @@ async def _sync_loop(self) -> None:
except Exception as exc:
logger.warning("Matrix: sync event dispatch error: %s", exc)
- # Retry any buffered undecrypted events.
- if self._pending_megolm:
- await self._retry_pending_decryptions()
-
except asyncio.CancelledError:
return
except Exception as exc:
```
```diff
@@ -1009,64 +1147,19 @@ async def _sync_loop(self) -> None:
return
# Detect permanent auth/permission failures.
err_str = str(exc).lower()
- if "401" in err_str or "403" in err_str or "unauthorized" in err_str or "forbidden" in err_str:
- logger.error("Matrix: permanent auth error: %s — stopping sync", exc)
+ if (
+ "401" in err_str
+ or "403" in err_str
+ or "unauthorized" in err_str
+ or "forbidden" in err_str
+ ):
+ logger.error(
+ "Matrix: permanent auth error: %s — stopping sync", exc
+ )
return
logger.warning("Matrix: sync error: %s — retrying in 5s", exc)
await asyncio.sleep(5)
- async def _retry_pending_decryptions(self) -> None:
- """Retry decrypting buffered encrypted events after new keys arrive."""
- client = self._client
- if not client or not self._pending_megolm:
- return
- crypto = getattr(client, "crypto", None)
- if not crypto:
- return
-
- now = time.time()
- still_pending: list = []
-
- for room_id, event, ts in self._pending_megolm:
- # Drop events that have aged past the TTL.
- if now - ts > _PENDING_EVENT_TTL:
- logger.debug(
- "Matrix: dropping expired pending event %s (age %.0fs)",
- getattr(event, "event_id", "?"), now - ts,
- )
- continue
-
- try:
- decrypted = await crypto.decrypt_megolm_event(event)
- except Exception:
- still_pending.append((room_id, event, ts))
- continue
-
- if decrypted is None or decrypted is event:
- still_pending.append((room_id, event, ts))
- continue
-
- logger.info(
- "Matrix: decrypted buffered event %s",
- getattr(event, "event_id", "?"),
- )
-
- # Route to the appropriate handler.
- # Remove from dedup set so _on_room_message doesn't drop it
- # (the encrypted event ID was already registered by _on_encrypted_event).
- decrypted_id = str(getattr(decrypted, "event_id", getattr(event, "event_id", "")))
- if decrypted_id:
- self._processed_events_set.discard(decrypted_id)
- try:
- await self._on_room_message(decrypted)
- except Exception as exc:
- logger.warning(
- "Matrix: error processing decrypted event %s: %s",
- getattr(event, "event_id", "?"), exc,
- )
-
- self._pending_megolm = still_pending
-
# ------------------------------------------------------------------
# Event callbacks
# ------------------------------------------------------------------
```
```diff
@@ -1086,7 +1179,11 @@ async def _on_room_message(self, event: Any) -> None:
return
# Startup grace: ignore old messages from initial sync.
- raw_ts = getattr(event, "timestamp", None) or getattr(event, "server_timestamp", None) or 0
+ raw_ts = (
+ getattr(event, "timestamp", None)
+ or getattr(event, "server_timestamp", None)
+ or 0
+ )
event_ts = raw_ts / 1000.0 if raw_ts else 0.0
if event_ts and event_ts < self._startup_ts - _STARTUP_GRACE_SECONDS:
return
```
```diff
@@ -1126,9 +1223,13 @@ async def _on_room_message(self, event: Any) -> None:
# Dispatch by msgtype.
media_msgtypes = ("m.image", "m.audio", "m.video", "m.file")
if msgtype in media_msgtypes:
- await self._handle_media_message(room_id, sender, event_id, event_ts, source_content, relates_to, msgtype)
+ await self._handle_media_message(
+ room_id, sender, event_id, event_ts, source_content, relates_to, msgtype
+ )
elif msgtype == "m.text":
- await self._handle_text_message(room_id, sender, event_id, event_ts, source_content, relates_to)
+ await self._handle_text_message(
+ room_id, sender, event_id, event_ts, source_content, relates_to
+ )
async def _resolve_message_context(
self,
```
```diff
@@ -1154,7 +1255,9 @@ async def _resolve_message_context(
formatted_body = source_content.get("formatted_body")
# m.mentions.user_ids (MSC3952 / Matrix v1.7) — authoritative mention signal.
mentions_block = source_content.get("m.mentions") or {}
- mention_user_ids = mentions_block.get("user_ids") if isinstance(mentions_block, dict) else None
+ mention_user_ids = (
+ mentions_block.get("user_ids") if isinstance(mentions_block, dict) else None
+ )
is_mentioned = self._is_bot_mentioned(body, formatted_body, mention_user_ids)
# Require-mention gating.
```
```diff
@@ -1170,8 +1273,8 @@ async def _resolve_message_context(
thread_id = event_id
self._threads.mark(thread_id)
- # Strip mention from body.
- if is_mentioned:
+ # Strip mention from body (only when mention-gating is active).
+ if is_mentioned and self._require_mention:
body = self._strip_mention(body)
# Auto-thread.
```
```diff
@@ -1210,7 +1313,12 @@ async def _handle_text_message(
return
ctx = await self._resolve_message_context(
- room_id, sender, event_id, body, source_content, relates_to,
+ room_id,
+ sender,
+ event_id,
+ body,
+ source_content,
+ relates_to,
)
if ctx is None:
return
```
```diff
@@ -1288,7 +1396,9 @@ async def _handle_media_message(
if url and url.startswith("mxc://"):
http_url = self._mxc_to_http(url)
- is_encrypted_media = bool(file_content and isinstance(file_content, dict) and file_content.get("url"))
+ is_encrypted_media = bool(
+ file_content and isinstance(file_content, dict) and file_content.get("url")
+ )
media_type = "application/octet-stream"
msg_type = MessageType.DOCUMENT
```
```diff
@@ -1312,9 +1422,9 @@ async def _handle_media_message(
# Cache media locally when downstream tools need a real file path.
cached_path = None
- should_cache_locally = (
- msg_type == MessageType.PHOTO or is_voice_message or is_encrypted_media
- )
+ should_cache_locally = msg_type in (
+ MessageType.PHOTO, MessageType.AUDIO, MessageType.VIDEO, MessageType.DOCUMENT,
+ ) or is_voice_message or is_encrypted_media
if should_cache_locally and url:
try:
file_bytes = await self._client.download_media(ContentURI(url))
```
```diff
@@ -1322,17 +1432,35 @@ async def _handle_media_message(
if is_encrypted_media:
from mautrix.crypto.attachments import decrypt_attachment
- hashes_value = file_content.get("hashes") if isinstance(file_content, dict) else None
- hash_value = hashes_value.get("sha256") if isinstance(hashes_value, dict) else None
+ hashes_value = (
+ file_content.get("hashes")
+ if isinstance(file_content, dict)
+ else None
+ )
+ hash_value = (
+ hashes_value.get("sha256")
+ if isinstance(hashes_value, dict)
+ else None
+ )
- key_value = file_content.get("key") if isinstance(file_content, dict) else None
+ key_value = (
+ file_content.get("key")
+ if isinstance(file_content, dict)
+ else None
+ )
if isinstance(key_value, dict):
key_value = key_value.get("k")
- iv_value = file_content.get("iv") if isinstance(file_content, dict) else None
+ iv_value = (
+ file_content.get("iv")
+ if isinstance(file_content, dict)
+ else None
+ )
if key_value and hash_value and iv_value:
- file_bytes = decrypt_attachment(file_bytes, key_value, hash_value, iv_value)
+ file_bytes = decrypt_attachment(
+ file_bytes, key_value, hash_value, iv_value
+ )
else:
logger.warning(
"[Matrix] Encrypted media event missing decryption metadata for %s",
```
```diff
@@ -1358,25 +1486,46 @@ async def _handle_media_message(
cached_path = cache_image_from_bytes(file_bytes, ext=ext)
logger.info("[Matrix] Cached user image at %s", cached_path)
elif msg_type in (MessageType.AUDIO, MessageType.VOICE):
- ext = Path(body or ("voice.ogg" if is_voice_message else "audio.ogg")).suffix or ".ogg"
+ ext = (
+ Path(
+ body
+ or (
+ "voice.ogg" if is_voice_message else "audio.ogg"
+ )
+ ).suffix
+ or ".ogg"
+ )
cached_path = cache_audio_from_bytes(file_bytes, ext=ext)
else:
filename = body or (
- "video.mp4" if msg_type == MessageType.VIDEO else "document"
+ "video.mp4"
+ if msg_type == MessageType.VIDEO
+ else "document"
+ )
+ cached_path = cache_document_from_bytes(
+ file_bytes, filename
)
- cached_path = cache_document_from_bytes(file_bytes, filename)
except Exception as e:
logger.warning("[Matrix] Failed to cache media: %s", e)
ctx = await self._resolve_message_context(
- room_id, sender, event_id, body, source_content, relates_to,
+ room_id,
+ sender,
+ event_id,
+ body,
+ source_content,
+ relates_to,
)
if ctx is None:
return
body, is_dm, chat_type, thread_id, display_name, source = ctx
allow_http_fallback = bool(http_url) and not is_encrypted_media
- media_urls = [cached_path] if cached_path else ([http_url] if allow_http_fallback else None)
+ media_urls = (
+ [cached_path]
+ if cached_path
+ else ([http_url] if allow_http_fallback else None)
+ )
media_types = [media_type] if media_urls else None
msg_event = MessageEvent(
```
```diff
@@ -1391,23 +1540,6 @@ async def _handle_media_message(
await self.handle_message(msg_event)
- async def _on_encrypted_event(self, event: Any) -> None:
- """Handle encrypted events that could not be auto-decrypted."""
- room_id = str(getattr(event, "room_id", ""))
- event_id = str(getattr(event, "event_id", ""))
-
- if self._is_duplicate_event(event_id):
- return
-
- logger.warning(
- "Matrix: could not decrypt event %s in %s — buffering for retry",
- event_id, room_id,
- )
-
- self._pending_megolm.append((room_id, event, time.time()))
- if len(self._pending_megolm) > _MAX_PENDING_EVENTS:
- self._pending_megolm = self._pending_megolm[-_MAX_PENDING_EVENTS:]
-
async def _on_invite(self, event: Any) -> None:
"""Auto-join rooms when invited."""
```
```diff
@@ -1430,7 +1562,10 @@ async def _on_invite(self, event: Any) -> None:
# ------------------------------------------------------------------
async def _send_reaction(
- self, room_id: str, event_id: str, emoji: str,
+ self,
+ room_id: str,
+ event_id: str,
+ emoji: str,
) -> Optional[str]:
"""Send an emoji reaction to a message in a room.
Returns the reaction event_id on success, None on failure.
```
```diff
@@ -1447,7 +1582,9 @@ async def _send_reaction(
}
try:
resp_event_id = await self._client.send_message_event(
- RoomID(room_id), EventType.REACTION, content,
+ RoomID(room_id),
+ EventType.REACTION,
+ content,
)
logger.debug("Matrix: sent reaction %s to %s", emoji, event_id)
return str(resp_event_id)
```
```diff
@@ -1456,7 +1593,10 @@ async def _send_reaction(
return None
async def _redact_reaction(
- self, room_id: str, reaction_event_id: str, reason: str = "",
+ self,
+ room_id: str,
+ reaction_event_id: str,
+ reason: str = "",
) -> bool:
"""Remove a reaction by redacting its event."""
return await self.redact_message(room_id, reaction_event_id, reason)
```
```diff
@@ -1473,7 +1613,9 @@ async def on_processing_start(self, event: MessageEvent) -> None:
self._pending_reactions[(room_id, msg_id)] = reaction_event_id
async def on_processing_complete(
- self, event: MessageEvent, outcome: ProcessingOutcome,
+ self,
+ event: MessageEvent,
+ outcome: ProcessingOutcome,
) -> None:
"""Replace eyes with checkmark (success) or cross (failure)."""
if not self._reactions_enabled:
```
```diff
@@ -1507,7 +1649,11 @@ async def _on_reaction(self, event: Any) -> None:
room_id = str(getattr(event, "room_id", ""))
content = getattr(event, "content", None)
if content:
- relates_to = content.get("m.relates_to", {}) if isinstance(content, dict) else getattr(content, "relates_to", {})
+ relates_to = (
+ content.get("m.relates_to", {})
+ if isinstance(content, dict)
+ else getattr(content, "relates_to", {})
+ )
reacts_to = ""
key = ""
if isinstance(relates_to, dict):
```
```diff
@@ -1518,7 +1664,10 @@ async def _on_reaction(self, event: Any) -> None:
key = str(getattr(relates_to, "key", ""))
logger.info(
"Matrix: reaction %s from %s on %s in %s",
- key, sender, reacts_to, room_id,
+ key,
+ sender,
+ reacts_to,
+ room_id,
)
# ------------------------------------------------------------------
```
```diff
@@ -1528,10 +1677,15 @@ async def _on_reaction(self, event: Any) -> None:
def _text_batch_key(self, event: MessageEvent) -> str:
"""Session-scoped key for text message batching."""
from gateway.session import build_session_key
+
return build_session_key(
event.source,
- group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
- thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
+ group_sessions_per_user=self.config.extra.get(
+ "group_sessions_per_user", True
+ ),
+ thread_sessions_per_user=self.config.extra.get(
+ "thread_sessions_per_user", False
+ ),
)
def _enqueue_text_event(self, event: MessageEvent) -> None:
```
```diff
@@ -1544,7 +1698,9 @@ def _enqueue_text_event(self, event: MessageEvent) -> None:
self._pending_text_batches[key] = event
else:
if event.text:
- existing.text = f"{existing.text}\n{event.text}" if existing.text else event.text
+ existing.text = (
+ f"{existing.text}\n{event.text}" if existing.text else event.text
+ )
existing._last_chunk_len = chunk_len # type: ignore[attr-defined]
if event.media_urls:
existing.media_urls.extend(event.media_urls)
```
```diff
@@ -1573,7 +1729,8 @@ async def _flush_text_batch(self, key: str) -> None:
return
logger.info(
"[Matrix] Flushing text batch %s (%d chars)",
- key, len(event.text or ""),
+ key,
+ len(event.text or ""),
)
await self.handle_message(event)
finally:
```
```diff
@@ -1586,11 +1743,13 @@ async def _flush_text_batch(self, key: str) -> None:
def _background_read_receipt(self, room_id: str, event_id: str) -> None:
"""Fire-and-forget read receipt with error logging."""
+
async def _send() -> None:
try:
await self.send_read_receipt(room_id, event_id)
except Exception as exc: # pragma: no cover — defensive
logger.debug("Matrix: background read receipt failed: %s", exc)
+
asyncio.ensure_future(_send())
async def send_read_receipt(self, room_id: str, event_id: str) -> bool:
```
```diff
@@ -1614,14 +1773,19 @@ async def send_read_receipt(self, room_id: str, event_id: str) -> bool:
# ------------------------------------------------------------------
async def redact_message(
- self, room_id: str, event_id: str, reason: str = "",
+ self,
+ room_id: str,
+ event_id: str,
+ reason: str = "",
) -> bool:
"""Redact (delete) a message or event from a room."""
if not self._client:
return False
try:
await self._client.redact(
- RoomID(room_id), EventID(event_id), reason=reason or None,
+ RoomID(room_id),
+ EventID(event_id),
+ reason=reason or None,
)
logger.info("Matrix: redacted %s in %s", event_id, room_id)
return True
```
```diff
@@ -1712,7 +1876,10 @@ async def set_presence(self, state: str = "online", status_msg: str = "") -> boo
# ------------------------------------------------------------------
async def _send_simple_message(
- self, chat_id: str, text: str, msgtype: str,
+ self,
+ chat_id: str,
+ text: str,
+ msgtype: str,
) -> SendResult:
"""Send a simple message (emote, notice) with optional HTML formatting."""
if not self._client or not text:
```
```diff
@@ -1726,7 +1893,9 @@ async def _send_simple_message(
try:
event_id = await self._client.send_message_event(
- RoomID(chat_id), EventType.ROOM_MESSAGE, msg_content,
+ RoomID(chat_id),
+ EventType.ROOM_MESSAGE,
+ msg_content,
)
return SendResult(success=True, message_id=str(event_id))
except Exception as exc:
```
```diff
@@ -1741,7 +1910,9 @@ async def _is_dm_room(self, room_id: str) -> bool:
if self._dm_rooms.get(room_id, False):
return True
# Fallback: check member count via state store.
- state_store = getattr(self._client, "state_store", None) if self._client else None
+ state_store = (
+ getattr(self._client, "state_store", None) if self._client else None
+ )
if state_store:
try:
members = await state_store.get_members(room_id)
```
```diff
@@ -1775,10 +1946,7 @@ async def _refresh_dm_cache(self) -> None:
if isinstance(rooms, list):
dm_room_ids.update(str(r) for r in rooms)
- self._dm_rooms = {
- rid: (rid in dm_room_ids)
- for rid in self._joined_rooms
- }
+ self._dm_rooms = {rid: (rid in dm_room_ids) for rid in self._joined_rooms}
# ------------------------------------------------------------------
# Mention detection helpers
```
```diff
@@ -1808,7 +1976,9 @@ def _is_bot_mentioned(
return True
if self._user_id and ":" in self._user_id:
localpart = self._user_id.split(":")[0].lstrip("@")
- if localpart and re.search(r'\b' + re.escape(localpart) + r'\b', body, re.IGNORECASE):
+ if localpart and re.search(
+ r"\b" + re.escape(localpart) + r"\b", body, re.IGNORECASE
+ ):
return True
if formatted_body and self._user_id:
if f"matrix.to/#/{self._user_id}" in formatted_body:
```
```diff
@@ -1816,18 +1986,20 @@ def _is_bot_mentioned(
return False
def _strip_mention(self, body: str) -> str:
- """Remove bot mention from message body."""
+ """Strip the bot's full MXID (``@user:server``) from *body*.
+
+ The bare localpart is intentionally *not* stripped — it would
+ mangle file paths like ``/home/hermes/media/file.png``.
+ """
if self._user_id:
body = body.replace(self._user_id, "")
- if self._user_id and ":" in self._user_id:
- localpart = self._user_id.split(":")[0].lstrip("@")
- if localpart:
- body = re.sub(r'\b' + re.escape(localpart) + r'\b', '', body, flags=re.IGNORECASE)
return body.strip()
async def _get_display_name(self, room_id: str, user_id: str) -> str:
"""Get a user's display name in a room, falling back to user_id."""
- state_store = getattr(self._client, "state_store", None) if self._client else None
+ state_store = (
+ getattr(self._client, "state_store", None) if self._client else None
+ )
if state_store:
try:
member = await state_store.get_member(room_id, user_id)
```
```diff
@@ -1915,9 +2087,7 @@ def _protect_html(html_fragment: str) -> str:
# Inline code: `code`
result = re.sub(
r"`([^`\n]+)`",
- lambda m: _protect_html(
- f"{_html_escape(m.group(1))}"
- ),
+ lambda m: _protect_html(f"{_html_escape(m.group(1))}"),
result,
)
```
```diff
@@ -1962,11 +2132,18 @@ def _protect_html(html_fragment: str) -> str:
continue
# Blockquote
- if line.startswith("> ") or line == ">" or line.startswith("> ") or line == ">":
+ if (
+ line.startswith("> ")
+ or line == ">"
+ or line.startswith("> ")
+ or line == ">"
+ ):
bq_lines = []
while i < len(lines) and (
- lines[i].startswith("> ") or lines[i] == ">"
- or lines[i].startswith("> ") or lines[i] == ">"
+ lines[i].startswith("> ")
+ or lines[i] == ">"
+ or lines[i].startswith("> ")
+ or lines[i] == ">"
):
ln = lines[i]
if ln.startswith("> "):
```
```diff
@@ -2007,13 +2184,19 @@ def _protect_html(html_fragment: str) -> str:
result = "\n".join(out_lines)
# Inline transforms.
- result = re.sub(r"\*\*(.+?)\*\*", r"\1", result, flags=re.DOTALL)
+ result = re.sub(
+ r"\*\*(.+?)\*\*", r"\1", result, flags=re.DOTALL
+ )
result = re.sub(r"__(.+?)__", r"\1", result, flags=re.DOTALL)
result = re.sub(r"\*(.+?)\*", r"\1", result, flags=re.DOTALL)
- result = re.sub(r"(?\1", result, flags=re.DOTALL)
+ result = re.sub(
+ r"(?\1", result, flags=re.DOTALL
+ )
result = re.sub(r"~~(.+?)~~", r"\1", result, flags=re.DOTALL)
result = re.sub(r"\n", "
\n", result)
- result = re.sub(r"
\n(\n()
", r"\1", result)
# Restore protected regions.
```
- `gateway/run.py`
```diff
@@ -7929,12 +7929,15 @@ async def _run_agent_via_proxy(
if _adapter:
_adapter_supports_edit = getattr(_adapter, "SUPPORTS_MESSAGE_EDITING", True)
_effective_cursor = _scfg.cursor if _adapter_supports_edit else ""
+ _buffer_only = False
if source.platform == Platform.MATRIX:
_effective_cursor = ""
+ _buffer_only = True
_consumer_cfg = StreamConsumerConfig(
edit_interval=_scfg.edit_interval,
buffer_threshold=_scfg.buffer_threshold,
cursor=_effective_cursor,
+ buffer_only=_buffer_only,
)
_stream_consumer = GatewayStreamConsumer(
adapter=_adapter,
```
```diff
@@ -8500,12 +8503,15 @@ def run_sync():
# Some Matrix clients render the streaming cursor
# as a visible tofu/white-box artifact. Keep
# streaming text on Matrix, but suppress the cursor.
+ _buffer_only = False
if source.platform == Platform.MATRIX:
_effective_cursor = ""
+ _buffer_only = True
_consumer_cfg = StreamConsumerConfig(
edit_interval=_scfg.edit_interval,
buffer_threshold=_scfg.buffer_threshold,
cursor=_effective_cursor,
+ buffer_only=_buffer_only,
)
_stream_consumer = GatewayStreamConsumer(
adapter=_adapter,
```
- `gateway/stream_consumer.py`
```diff
@@ -43,6 +43,7 @@ class StreamConsumerConfig:
edit_interval: float = 1.0
buffer_threshold: int = 40
cursor: str = " ▉"
+ buffer_only: bool = False
class GatewayStreamConsumer:
```
```diff
@@ -295,10 +296,13 @@ async def run(self) -> None:
got_done
or got_segment_break
or commentary_text is not None
- or (elapsed >= self._current_edit_interval
- and self._accumulated)
- or len(self._accumulated) >= self.cfg.buffer_threshold
)
+ if not self.cfg.buffer_only:
+ should_edit = should_edit or (
+ (elapsed >= self._current_edit_interval
+ and self._accumulated)
+ or len(self._accumulated) >= self.cfg.buffer_threshold
+ )
current_update_visible = False
if should_edit and self._accumulated:
```
- `tests/gateway/test_matrix.py`
```diff
@@ -108,6 +108,9 @@ def __init__(self, mxid=None, device_id=None, api=None,
def add_event_handler(self, event_type, handler):
self._event_handlers.setdefault(event_type, []).append(handler)
+ def add_dispatcher(self, dispatcher_type):
+ pass
+
class InternalEventType:
INVITE = "internal.invite"
```
```diff
@@ -115,6 +118,14 @@ class InternalEventType:
mautrix_client.InternalEventType = InternalEventType
mautrix.client = mautrix_client
+ # --- mautrix.client.dispatcher ---
+ mautrix_client_dispatcher = types.ModuleType("mautrix.client.dispatcher")
+
+ class MembershipEventDispatcher:
+ pass
+
+ mautrix_client_dispatcher.MembershipEventDispatcher = MembershipEventDispatcher
+
# --- mautrix.client.state_store ---
mautrix_client_state_store = types.ModuleType("mautrix.client.state_store")
```
```diff
@@ -163,6 +174,19 @@ def __init__(self, account_id="", pickle_key=""): # noqa: S301
mautrix_crypto_store.MemoryCryptoStore = MemoryCryptoStore
+ # --- mautrix.crypto.attachments ---
+ mautrix_crypto_attachments = types.ModuleType("mautrix.crypto.attachments")
+
+ def encrypt_attachment(data):
+ encrypted_file = MagicMock()
+ encrypted_file.serialize.return_value = {
+ "key": {"k": "testkey"}, "iv": "testiv",
+ "hashes": {"sha256": "testhash"}, "v": "v2",
+ }
+ return (b"ciphertext_" + data, encrypted_file)
+
+ mautrix_crypto_attachments.encrypt_attachment = encrypt_attachment
+
# --- mautrix.crypto.store.asyncpg ---
mautrix_crypto_store_asyncpg = types.ModuleType("mautrix.crypto.store.asyncpg")
```
```diff
@@ -200,8 +224,10 @@ def create(cls, url, upgrade_table=None):
"mautrix.api": mautrix_api,
"mautrix.types": mautrix_types,
"mautrix.client": mautrix_client,
+ "mautrix.client.dispatcher": mautrix_client_dispatcher,
"mautrix.client.state_store": mautrix_client_state_store,
"mautrix.crypto": mautrix_crypto,
+ "mautrix.crypto.attachments": mautrix_crypto_attachments,
"mautrix.crypto.store": mautrix_crypto_store,
"mautrix.crypto.store.asyncpg": mautrix_crypto_store_asyncpg,
"mautrix.util": mautrix_util,
```
```diff
@@ -357,6 +383,16 @@ async def test_stop_typing_clears_matrix_typing_state(self):
timeout=0,
)
+ @pytest.mark.asyncio
+ async def test_stop_typing_no_client_is_noop(self):
+ self.adapter._client = None
+ await self.adapter.stop_typing("!room:example.org") # should not raise
+
+ @pytest.mark.asyncio
+ async def test_stop_typing_suppresses_exceptions(self):
+ self.adapter._client.set_typing = AsyncMock(side_effect=Exception("network"))
+ await self.adapter.stop_typing("!room:example.org") # should not raise
+
# ---------------------------------------------------------------------------
# mxc:// URL conversion
```
```diff
@@ -835,6 +871,41 @@ def __init__(self, user_id, device_id):
await adapter.disconnect()
+class TestDeviceKeyReVerification:
+ @pytest.mark.asyncio
+ async def test_verify_fails_when_server_keys_mismatch_after_upload(self):
+ """share_keys() succeeds but server still has old keys -> should return False."""
+ adapter = _make_adapter()
+
+ mock_client = MagicMock()
+ mock_client.mxid = "@bot:example.org"
+ mock_client.device_id = "TESTDEVICE"
+
+ # First query: keys missing -> triggers share_keys
+ # Second query: keys still don't match -> should fail
+ mock_keys_missing = MagicMock()
+ mock_keys_missing.device_keys = {"@bot:example.org": {}}
+
+ mock_keys_mismatch = MagicMock()
+ mock_device = MagicMock()
+ mock_device.keys = {"ed25519:TESTDEVICE": "server_old_key"}
+ mock_keys_mismatch.device_keys = {"@bot:example.org": {"TESTDEVICE": mock_device}}
+
+ mock_client.query_keys = AsyncMock(side_effect=[mock_keys_missing, mock_keys_mismatch])
+
+ mock_olm = MagicMock()
+ mock_olm.account = MagicMock()
+ mock_olm.account.shared = False
+ mock_olm.account.identity_keys = {"ed25519": "local_new_key"}
+ mock_olm.share_keys = AsyncMock()
+
+ from gateway.platforms.matrix import MatrixAdapter
+ result = await adapter._verify_device_keys_on_server(mock_client, mock_olm)
+
+ assert result is False
+ mock_olm.share_keys.assert_awaited_once()
+
+
class TestMatrixE2EEHardFail:
"""connect() must refuse to start when E2EE is requested but deps are missing."""
```
```diff
@@ -1139,6 +1210,56 @@ async def _sync_once(**kwargs):
mock_sync_store.put_next_batch.assert_awaited_once_with("s1234")
+class TestMatrixUploadAndSend:
+ @pytest.mark.asyncio
+ async def test_upload_unencrypted_room_uses_plain_url(self):
+ """Unencrypted rooms should use plain 'url' key."""
+ adapter = _make_adapter()
+ adapter._encryption = True
+ mock_client = MagicMock()
+ mock_client.crypto = object()
+ mock_client.state_store = MagicMock()
+ mock_client.state_store.is_encrypted = AsyncMock(return_value=False)
+ mock_client.upload_media = AsyncMock(return_value="mxc://example.org/plain")
+ mock_client.send_message_event = AsyncMock(return_value="$event")
+ adapter._client = mock_client
+
+ result = await adapter._upload_and_send(
+ "!room:example.org", b"hello", "test.txt", "text/plain", "m.file",
+ )
+
+ assert result.success is True
+ sent = mock_client.send_message_event.await_args.args[2]
+ assert sent["url"] == "mxc://example.org/plain"
+ assert "file" not in sent
+
+ @pytest.mark.asyncio
+ async def test_upload_encrypted_room_uses_file_payload(self):
+ """Encrypted rooms should use 'file' key with crypto metadata."""
+ adapter = _make_adapter()
+ adapter._encryption = True
+ mock_client = MagicMock()
+ mock_client.crypto = object()
+ mock_client.state_store = MagicMock()
+ mock_client.state_store.is_encrypted = AsyncMock(return_value=True)
+ mock_client.upload_media = AsyncMock(return_value="mxc://example.org/enc")
+ mock_client.send_message_event = AsyncMock(return_value="$event")
+ adapter._client = mock_client
+
+ result = await adapter._upload_and_send(
+ "!room:example.org", b"secret", "secret.txt", "text/plain", "m.file",
+ )
+
+ assert result.success is True
+ # Should have uploaded ciphertext, not plaintext
+ uploaded_data = mock_client.upload_media.await_args.args[0]
+ assert uploaded_data != b"secret"
+ sent = mock_client.send_message_event.await_args.args[2]
+ assert "url" not in sent
+ assert "file" in sent
+ assert sent["file"]["url"] == "mxc://example.org/enc"
+
+
class TestMatrixEncryptedSendFallback:
@pytest.mark.asyncio
async def test_send_retries_after_e2ee_error(self):
```
```diff
@@ -1165,128 +1286,24 @@ async def test_send_retries_after_e2ee_error(self):
# ---------------------------------------------------------------------------
-# E2EE: MegolmEvent key request + buffering via _on_encrypted_event
+# E2EE: _joined_rooms reference preservation for CryptoStateStore
# ---------------------------------------------------------------------------
-class TestMatrixMegolmEventHandling:
- @pytest.mark.asyncio
- async def test_encrypted_event_buffers_for_retry(self):
- """_on_encrypted_event should buffer undecrypted events for retry."""
- adapter = _make_adapter()
- adapter._user_id = "@bot:example.org"
- adapter._startup_ts = 0.0
- adapter._dm_rooms = {}
-
- fake_event = MagicMock()
- fake_event.room_id = "!room:example.org"
- fake_event.event_id = "$encrypted_event"
- fake_event.sender = "@alice:example.org"
-
- await adapter._on_encrypted_event(fake_event)
-
- # Should have buffered the event
- assert len(adapter._pending_megolm) == 1
- room_id, event, ts = adapter._pending_megolm[0]
- assert room_id == "!room:example.org"
- assert event is fake_event
-
- @pytest.mark.asyncio
- async def test_encrypted_event_buffer_capped(self):
- """Buffer should not grow past _MAX_PENDING_EVENTS."""
- adapter = _make_adapter()
- adapter._user_id = "@bot:example.org"
- adapter._startup_ts = 0.0
- adapter._dm_rooms = {}
-
- from gateway.platforms.matrix import _MAX_PENDING_EVENTS
-
- for i in range(_MAX_PENDING_EVENTS + 10):
- evt = MagicMock()
- evt.room_id = "!room:example.org"
- evt.event_id = f"$event_{i}"
- evt.sender = "@alice:example.org"
- await adapter._on_encrypted_event(evt)
-
- assert len(adapter._pending_megolm) == _MAX_PENDING_EVENTS
-
-
-# ---------------------------------------------------------------------------
-# E2EE: Retry pending decryptions
-# ---------------------------------------------------------------------------
-
-class TestMatrixRetryPendingDecryptions:
- @pytest.mark.asyncio
- async def test_successful_decryption_routes_to_handler(self):
- adapter = _make_adapter()
- adapter._user_id = "@bot:example.org"
- adapter._startup_ts = 0.0
- adapter._dm_rooms = {}
-
- fake_encrypted = MagicMock()
- fake_encrypted.event_id = "$encrypted"
-
- decrypted_event = MagicMock()
-
- mock_crypto = MagicMock()
- mock_crypto.decrypt_megolm_event = AsyncMock(return_value=decrypted_event)
-
- fake_client = MagicMock()
- fake_client.crypto = mock_crypto
- adapter._client = fake_client
-
- now = time.time()
- adapter._pending_megolm = [("!room:ex.org", fake_encrypted, now)]
-
- with patch.object(adapter, "_on_room_message", AsyncMock()) as mock_handler:
- await adapter._retry_pending_decryptions()
- mock_handler.assert_awaited_once_with(decrypted_event)
-
- # Buffer should be empty now
- assert len(adapter._pending_megolm) == 0
-
- @pytest.mark.asyncio
- async def test_still_undecryptable_stays_in_buffer(self):
- adapter = _make_adapter()
-
- fake_encrypted = MagicMock()
- fake_encrypted.event_id = "$still_encrypted"
-
- mock_crypto = MagicMock()
- mock_crypto.decrypt_megolm_event = AsyncMock(side_effect=Exception("missing key"))
-
- fake_client = MagicMock()
- fake_client.crypto = mock_crypto
- adapter._client = fake_client
-
- now = time.time()
- adapter._pending_megolm = [("!room:ex.org", fake_encrypted, now)]
-
- await adapter._retry_pending_decryptions()
-
- assert len(adapter._pending_megolm) == 1
-
- @pytest.mark.asyncio
- async def test_expired_events_dropped(self):
- adapter = _make_adapter()
-
- from gateway.platforms.matrix import _PENDING_EVENT_TTL
-
- fake_event = MagicMock()
- fake_event.event_id = "$old_event"
-
- mock_crypto = MagicMock()
- fake_client = MagicMock()
- fake_client.crypto = mock_crypto
- adapter._client = fake_client
+class TestJoinedRoomsReference:
+ def test_joined_rooms_reference_preserved_after_reassignment(self):
+ """_CryptoStateStore must see updates after initial sync populates rooms."""
+ from gateway.platforms.matrix import _CryptoStateStore
- # Timestamp well past TTL
- old_ts = time.time() - _PENDING_EVENT_TTL - 60
- adapter._pending_megolm = [("!room:ex.org", fake_event, old_ts)]
+ joined = set()
+ store = _CryptoStateStore(MagicMock(), joined)
- await adapter._retry_pending_decryptions()
+ # Simulate what connect() should do: mutate in place, not reassign.
+ joined.clear()
+ joined.update(["!room1:example.org", "!room2:example.org"])
- # Should have been dropped
- assert len(adapter._pending_megolm) == 0
+ import asyncio
+ rooms = asyncio.get_event_loop().run_until_complete(store.find_shared_rooms("@user:ex"))
+ assert set(rooms) == {"!room1:example.org", "!room2:example.org"}
# ---------------------------------------------------------------------------
```
```diff
@@ -1354,11 +1371,70 @@ async def test_connect_registers_encrypted_event_handler_when_encryption_on(self
handler_calls = mock_client.add_event_handler.call_args_list
registered_types = [call.args[0] for call in handler_calls]
- # Should have registered handlers for ROOM_MESSAGE, REACTION, INVITE, and ROOM_ENCRYPTED
- assert len(handler_calls) >= 4 # At minimum these four
+ # Should have registered handlers for ROOM_MESSAGE, REACTION, INVITE
+ assert len(handler_calls) >= 3
await adapter.disconnect()
+ @pytest.mark.asyncio
+ async def test_connect_fails_on_stale_otk_conflict(self):
+ """connect() must refuse E2EE when OTK upload hits 'already exists'."""
+ from gateway.platforms.matrix import MatrixAdapter
+
+ config = PlatformConfig(
+ enabled=True,
+ token="[redacted:secret-assignment]",
+ extra={
+ "homeserver": "https://matrix.example.org",
+ "user_id": "@bot:example.org",
+ "encryption": True,
+ },
+ )
+ adapter = MatrixAdapter(config)
+
+ fake_mautrix_mods = _make_fake_mautrix()
+
+ mock_client = MagicMock()
+ mock_client.mxid = "@bot:example.org"
+ mock_client.device_id = None
+ mock_client.state_store = MagicMock()
+ mock_client.sync_store = MagicMock()
+ mock_client.crypto = None
+ mock_client.whoami = AsyncMock(return_value=MagicMock(user_id="@bot:example.org", device_id="DEV123"))
+ mock_client.add_event_handler = MagicMock()
+ mock_client.add_dispatcher = MagicMock()
+ mock_client.query_keys = AsyncMock(return_value={
+ "device_keys": {"@bot:example.org": {"DEV123": {
+ "keys": {"ed25519:DEV123": "fake_ed25519_key"},
+ }}},
+ })
+ mock_client.api = MagicMock()
+ mock_client.api.token = "[redacted:secret-assignment]"
+ mock_client.api.session = MagicMock()
+ mock_client.api.session.close = AsyncMock()
+
+ # share_keys succeeds on first call (from _verify_device_keys_on_server),
+ # then raises "already exists" on the proactive OTK flush in connect().
+ mock_olm = MagicMock()
+ mock_olm.load = AsyncMock()
+ mock_olm.share_keys = AsyncMock(
+ side_effect=[None, Exception("One time key signed_curve25519:AAAAAQ already exists")]
+ )
+ mock_olm.share_keys_min_trust = None
+ mock_olm.send_keys_min_trust = None
+ mock_olm.account = MagicMock()
+ mock_olm.account.identity_keys = {"ed25519": "fake_ed25519_key"}
+
+ fake_mautrix_mods["mautrix.client"].Client = MagicMock(return_value=mock_client)
+ fake_mautrix_mods["mautrix.crypto"].OlmMachine = MagicMock(return_value=mock_olm)
+
+ from gateway.platforms import matrix as matrix_mod
+ with patch.object(matrix_mod, "_check_e2ee_deps", return_value=True):
+ with patch.dict("sys.modules", fake_mautrix_mods):
+ result = await adapter.connect()
+
+ assert result is False
+
# ---------------------------------------------------------------------------
# Disconnect
```
- `tests/gateway/test_matrix_mention.py`
```diff
@@ -10,7 +10,6 @@
from gateway.config import PlatformConfig
-
# The matrix adapter module is importable without mautrix installed
# (module-level imports use try/except with stubs). No need for
# module-level mock installation — tests that call adapter methods
```
```diff
@@ -159,9 +158,15 @@ def test_strip_full_user_id(self):
result = self.adapter._strip_mention("@hermes:example.org help me")
assert result == "help me"
- def test_strip_localpart(self):
+ def test_localpart_preserved(self):
+ """Localpart-only text is no longer stripped — avoids false positives in paths."""
result = self.adapter._strip_mention("hermes help me")
- assert result == "help me"
+ assert result == "hermes help me"
+
+ def test_localpart_in_path_preserved(self):
+ """Localpart inside a file path must not be damaged."""
+ result = self.adapter._strip_mention("read /home/hermes/config.yaml")
+ assert result == "read /home/hermes/config.yaml"
def test_strip_returns_empty_for_mention_only(self):
result = self.adapter._strip_mention("@hermes:example.org")
```
```diff
@@ -273,8 +278,8 @@ async def test_require_mention_dm_always_responds(monkeypatch):
@pytest.mark.asyncio
-async def test_dm_strips_mention(monkeypatch):
- """DMs strip mention from body, matching Discord behavior."""
+async def test_dm_strips_full_mxid(monkeypatch):
+ """DMs strip the full MXID from body when require_mention is on (default)."""
monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
```
```diff
@@ -289,6 +294,23 @@ async def test_dm_strips_mention(monkeypatch):
assert msg.text == "help me"
[redacted:email]
+async def test_dm_preserves_localpart_in_body(monkeypatch):
+ """DMs no longer strip bare localpart — only the full MXID is removed."""
+ monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
+ monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+ monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+ adapter = _make_adapter()
+ _set_dm(adapter)
+ event = _make_event("hermes help me")
+
+ await adapter._on_room_message(event)
+ adapter.handle_message.assert_awaited_once()
+ msg = adapter.handle_message.await_args.args[0]
+ assert msg.text == "hermes help me"
+
+
@pytest.mark.asyncio
async def test_bare_mention_passes_empty_string(monkeypatch):
"""A message that is only a mention should pass through as empty, not be dropped."""
```
```diff
@@ -309,7 +331,9 @@ async def test_bare_mention_passes_empty_string(monkeypatch):
async def test_require_mention_free_response_room(monkeypatch):
"""Free-response rooms bypass mention requirement."""
monkeypatch.delenv("MATRIX_REQUIRE_MENTION", raising=False)
- monkeypatch.setenv("MATRIX_FREE_RESPONSE_ROOMS", "!room1:example.org,!room2:example.org")
+ monkeypatch.setenv(
+ "MATRIX_FREE_RESPONSE_ROOMS", "!room1:example.org,!room2:example.org"
+ )
monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
adapter = _make_adapter()
```
```diff
@@ -351,6 +375,22 @@ async def test_require_mention_disabled(monkeypatch):
assert msg.text == "hello without mention"
[redacted:email]
+async def test_require_mention_disabled_skips_stripping(monkeypatch):
+ """MATRIX_REQUIRE_MENTION=false: mention text is NOT stripped from body."""
+ monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+ monkeypatch.delenv("MATRIX_FREE_RESPONSE_ROOMS", raising=False)
+ monkeypatch.setenv("MATRIX_AUTO_THREAD", "false")
+
+ adapter = _make_adapter()
+ event = _make_event("@hermes:example.org help me")
+
+ await adapter._on_room_message(event)
+ adapter.handle_message.assert_awaited_once()
+ msg = adapter.handle_message.await_args.args[0]
+ assert msg.text == "@hermes:example.org help me"
+
+
# ---------------------------------------------------------------------------
# Auto-thread in _on_room_message
# ---------------------------------------------------------------------------
```
```diff
@@ -442,8 +482,10 @@ class TestThreadPersistence:
def test_empty_state_file(self, tmp_path, monkeypatch):
"""No state file → empty set."""
from gateway.platforms.helpers import ThreadParticipationTracker
+
monkeypatch.setattr(
- ThreadParticipationTracker, "_state_path",
+ ThreadParticipationTracker,
+ "_state_path",
lambda self: tmp_path / "matrix_threads.json",
)
adapter = _make_adapter()
```
```diff
@@ -452,9 +494,11 @@ def test_empty_state_file(self, tmp_path, monkeypatch):
def test_track_thread_persists(self, tmp_path, monkeypatch):
"""mark() writes to disk."""
from gateway.platforms.helpers import ThreadParticipationTracker
+
state_path = tmp_path / "matrix_threads.json"
monkeypatch.setattr(
- ThreadParticipationTracker, "_state_path",
+ ThreadParticipationTracker,
+ "_state_path",
lambda self: state_path,
)
adapter = _make_adapter()
```
```diff
@@ -466,10 +510,12 @@ def test_track_thread_persists(self, tmp_path, monkeypatch):
def test_threads_survive_reload(self, tmp_path, monkeypatch):
"""Persisted threads are loaded by a new adapter instance."""
from gateway.platforms.helpers import ThreadParticipationTracker
+
state_path = tmp_path / "matrix_threads.json"
state_path.write_text(json.dumps(["$t1", "$t2"]))
monkeypatch.setattr(
- ThreadParticipationTracker, "_state_path",
+ ThreadParticipationTracker,
+ "_state_path",
lambda self: state_path,
)
adapter = _make_adapter()
```
```diff
@@ -479,9 +525,11 @@ def test_threads_survive_reload(self, tmp_path, monkeypatch):
def test_cap_max_tracked_threads(self, tmp_path, monkeypatch):
"""Thread set is trimmed to max_tracked."""
from gateway.platforms.helpers import ThreadParticipationTracker
+
state_path = tmp_path / "matrix_threads.json"
monkeypatch.setattr(
- ThreadParticipationTracker, "_state_path",
+ ThreadParticipationTracker,
+ "_state_path",
lambda self: state_path,
)
adapter = _make_adapter()
```
```diff
@@ -604,6 +652,7 @@ def test_yaml_bridge_sets_env_vars(self, monkeypatch, tmp_path):
}
import os
+
import yaml
config_file = tmp_path / "config.yaml"
```
```diff
@@ -613,18 +662,27 @@ def test_yaml_bridge_sets_env_vars(self, monkeypatch, tmp_path):
yaml_cfg = yaml.safe_load(config_file.read_text())
matrix_cfg = yaml_cfg.get("matrix", {})
if isinstance(matrix_cfg, dict):
- if "require_mention" in matrix_cfg and not os.getenv("MATRIX_REQUIRE_MENTION"):
- monkeypatch.setenv("MATRIX_REQUIRE_MENTION", str(matrix_cfg["require_mention"]).lower())
+ if "require_mention" in matrix_cfg and not os.getenv(
+ "MATRIX_REQUIRE_MENTION"
+ ):
+ monkeypatch.setenv(
+ "MATRIX_REQUIRE_MENTION", str(matrix_cfg["require_mention"]).lower()
+ )
frc = matrix_cfg.get("free_response_rooms")
if frc is not None and not os.getenv("MATRIX_FREE_RESPONSE_ROOMS"):
if isinstance(frc, list):
frc = ",".join(str(v) for v in frc)
monkeypatch.setenv("MATRIX_FREE_RESPONSE_ROOMS", str(frc))
if "auto_thread" in matrix_cfg and not os.getenv("MATRIX_AUTO_THREAD"):
- monkeypatch.setenv("MATRIX_AUTO_THREAD", str(matrix_cfg["auto_thread"]).lower())
+ monkeypatch.setenv(
+ "MATRIX_AUTO_THREAD", str(matrix_cfg["auto_thread"]).lower()
+ )
assert os.getenv("MATRIX_REQUIRE_MENTION") == "false"
- assert os.getenv("MATRIX_FREE_RESPONSE_ROOMS") == "!room1:example.org,!room2:example.org"
+ assert (
+ os.getenv("MATRIX_FREE_RESPONSE_ROOMS")
+ == "!room1:example.org,!room2:example.org"
+ )
assert os.getenv("MATRIX_AUTO_THREAD") == "false"
def test_yaml_bridge_sets_dm_mention_threads(self, monkeypatch, tmp_path):
```
```diff
@@ -632,6 +690,7 @@ def test_yaml_bridge_sets_dm_mention_threads(self, monkeypatch, tmp_path):
monkeypatch.delenv("MATRIX_DM_MENTION_THREADS", raising=False)
import os
+
import yaml
yaml_content = {"matrix": {"dm_mention_threads": True}}
```
```diff
@@ -641,8 +700,13 @@ def test_yaml_bridge_sets_dm_mention_threads(self, monkeypatch, tmp_path):
yaml_cfg = yaml.safe_load(config_file.read_text())
matrix_cfg = yaml_cfg.get("matrix", {})
if isinstance(matrix_cfg, dict):
- if "dm_mention_threads" in matrix_cfg and not os.getenv("MATRIX_DM_MENTION_THREADS"):
- monkeypatch.setenv("MATRIX_DM_MENTION_THREADS", str(matrix_cfg["dm_mention_threads"]).lower())
+ if "dm_mention_threads" in matrix_cfg and not os.getenv(
+ "MATRIX_DM_MENTION_THREADS"
+ ):
+ monkeypatch.setenv(
+ "MATRIX_DM_MENTION_THREADS",
+ str(matrix_cfg["dm_mention_threads"]).lower(),
+ )
assert os.getenv("MATRIX_DM_MENTION_THREADS") == "true"
```
```diff
@@ -651,9 +715,12 @@ def test_env_vars_take_precedence_over_yaml(self, monkeypatch):
monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "true")
import os
+
yaml_cfg = {"matrix": {"require_mention": False}}
matrix_cfg = yaml_cfg.get("matrix", {})
if "require_mention" in matrix_cfg and not os.getenv("MATRIX_REQUIRE_MENTION"):
- monkeypatch.setenv("MATRIX_REQUIRE_MENTION", str(matrix_cfg["require_mention"]).lower())
+ monkeypatch.setenv(
+ "MATRIX_REQUIRE_MENTION", str(matrix_cfg["require_mention"]).lower()
+ )
assert os.getenv("MATRIX_REQUIRE_MENTION") == "true"
```
- `tests/gateway/test_stream_consumer.py`
```diff
@@ -963,3 +963,106 @@ async def test_think_block_not_sent_to_platform(self):
await task
except asyncio.CancelledError:
pass
+
+
+# ── buffer_only mode tests ─────────────────────────────────────────────
+
+
+class TestBufferOnlyMode:
+ """Verify buffer_only mode suppresses intermediate edits and only
+ flushes on structural boundaries (done, segment break, commentary)."""
+
+ @pytest.mark.asyncio
+ async def test_suppresses_intermediate_edits(self):
+ """Time-based and size-based edits are skipped; only got_done flushes."""
+ adapter = MagicMock()
+ adapter.MAX_MESSAGE_LENGTH = 4096
+ adapter.send = AsyncMock(return_value=SimpleNamespace(success=True, message_id="msg1"))
+ adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True))
+
+ cfg = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor="", buffer_only=True)
+ consumer = GatewayStreamConsumer(adapter, "!room:server", config=cfg)
+
+ for word in ["Hello", " world", ", this", " is", " a", " test"]:
+ consumer.on_delta(word)
+ consumer.finish()
+
+ await consumer.run()
+
+ adapter.send.assert_called_once()
+ adapter.edit_message.assert_not_called()
+ assert "Hello world, this is a test" in adapter.send.call_args_list[0][1]["content"]
+
+ @pytest.mark.asyncio
+ async def test_flushes_on_segment_break(self):
+ """A segment break (tool call boundary) flushes accumulated text."""
+ adapter = MagicMock()
+ adapter.MAX_MESSAGE_LENGTH = 4096
+ adapter.send = AsyncMock(side_effect=[
+ SimpleNamespace(success=True, message_id="msg1"),
+ SimpleNamespace(success=True, message_id="msg2"),
+ ])
+ adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True))
+
+ cfg = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor="", buffer_only=True)
+ consumer = GatewayStreamConsumer(adapter, "!room:server", config=cfg)
+
+ consumer.on_delta("Before tool call")
+ consumer.on_delta(None)
+ consumer.on_delta("After tool call")
+ consumer.finish()
+
+ await consumer.run()
+
+ assert adapter.send.call_count == 2
+ assert "Before tool call" in adapter.send.call_args_list[0][1]["content"]
+ assert "After tool call" in adapter.send.call_args_list[1][1]["content"]
+ adapter.edit_message.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_flushes_on_commentary(self):
+ """An interim commentary message flushes in buffer_only mode."""
+ adapter = MagicMock()
+ adapter.MAX_MESSAGE_LENGTH = 4096
+ adapter.send = AsyncMock(side_effect=[
+ SimpleNamespace(success=True, message_id="msg1"),
+ SimpleNamespace(success=True, message_id="msg2"),
+ SimpleNamespace(success=True, message_id="msg3"),
+ ])
+ adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True))
+
+ cfg = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor="", buffer_only=True)
+ consumer = GatewayStreamConsumer(adapter, "!room:server", config=cfg)
+
+ consumer.on_delta("Working on it...")
+ consumer.on_commentary("I'll search for that first.")
+ consumer.on_delta("Here are the results.")
+ consumer.finish()
+
+ await consumer.run()
+
+ # Three sends: accumulated text, commentary, final text
+ assert adapter.send.call_count >= 2
+ adapter.edit_message.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_default_mode_still_triggers_intermediate_edits(self):
+ """Regression: buffer_only=False (default) still does progressive edits."""
+ adapter = MagicMock()
+ adapter.MAX_MESSAGE_LENGTH = 4096
+ adapter.send = AsyncMock(return_value=SimpleNamespace(success=True, message_id="msg1"))
+ adapter.edit_message = AsyncMock(return_value=SimpleNamespace(success=True))
+
+ # buffer_threshold=5 means any 5+ chars triggers an early edit
+ cfg = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5, cursor="")
+ consumer = GatewayStreamConsumer(adapter, "!room:server", config=cfg)
+
+ consumer.on_delta("Hello world, this is long enough to trigger edits")
+ consumer.finish()
+
+ await consumer.run()
+
+ # Should have at least one send. With buffer_threshold=5 and this much
+ # text, the consumer may send then edit, or just send once at got_done.
+ # The key assertion: this doesn't break.
+ assert adapter.send.call_count >= 1
```
- `website/docs/user-guide/messaging/matrix.md`
```diff
@@ -284,8 +284,40 @@ MATRIX_RECOVERY_KEY=EsT... your recovery key here
On each startup, if `MATRIX_RECOVERY_KEY` is set, Hermes imports cross-signing keys from the homeserver's secure secret storage and signs the current device. This is idempotent and safe to leave enabled permanently.
-:::warning
-If you delete the `~/.hermes/platforms/matrix/store/` directory, the bot loses its encryption keys. You'll need to verify the device again in your Matrix client. Back up this directory if you want to preserve encrypted sessions.
+:::warning[Deleting the crypto store]
+If you delete `~/.hermes/platforms/matrix/store/crypto.db`, the bot loses its encryption identity. Simply restarting with the same device ID will **not** fully recover — the homeserver still holds one-time keys signed with the old identity key, and peers cannot establish new Olm sessions.
+
+Hermes detects this condition on startup and refuses to enable E2EE, logging: `device XXXX has stale one-time keys on the server signed with a previous identity key`.
+
+**Easiest recovery: generate a new access token** (which gets a fresh device ID with no stale key history). See the "Upgrading from a previous version with E2EE" section below. This is the most reliable path and avoids touching the homeserver database.
+
+**Manual recovery** (advanced — keeps the same device ID):
+
+1. Stop Synapse and delete the old device from its database:
+ ```bash
+ sudo systemctl stop matrix-synapse
+ sudo sqlite3 /var/lib/matrix-synapse/homeserver.db "
+ DELETE FROM e2e_device_keys_json WHERE device_id = 'DEVICE_ID' AND user_id = '@hermes:your-server';
+ DELETE FROM e2e_one_time_keys_json WHERE device_id = 'DEVICE_ID' AND user_id = '@hermes:your-server';
+ DELETE FROM e2e_fallback_keys_json WHERE device_id = 'DEVICE_ID' AND user_id = '@hermes:your-server';
+ DELETE FROM devices WHERE device_id = 'DEVICE_ID' AND user_id = '@hermes:your-server';
+ "
+ sudo systemctl start matrix-synapse
+ ```
+ Or via the Synapse admin API (note the URL-encoded user ID):
+ ```bash
+ curl -X DELETE -H "Authorization: Bearer ADMIN_TOKEN" \
+ 'https://your-server/_synapse/admin/v2/users/%40hermes%3Ayour-server/devices/DEVICE_ID'
+ ```
+ Note: deleting a device via the admin API may also invalidate the associated access token. You may need to generate a new token afterward.
+
+2. Delete the local crypto store and restart Hermes:
+ ```bash
+ rm -f ~/.hermes/platforms/matrix/store/crypto.db*
+ # restart hermes
+ ```
+
+Other Matrix clients (Element, matrix-commander) may cache the old device keys. After recovery, type `/discardsession` in Element to force a new encryption session with the bot.
:::
:::info
```
```diff
@@ -361,6 +393,10 @@ pip install 'hermes-agent[matrix]'
### Upgrading from a previous version with E2EE
+:::tip
+If you also manually deleted `crypto.db`, see the "Deleting the crypto store" warning in the E2EE section above — there are additional steps to clear stale one-time keys from the homeserver.
+:::
+
If you previously used Hermes with `MATRIX_ENCRYPTION=true` and are upgrading to
a version that uses the new SQLite-based crypto store, the bot's encryption
identity has changed. Your Matrix client (Element) may cache the old device keys
```
- **#16821: matrix: parity salvage batch (10 contributor PRs)**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `gateway/platforms/base.py`, `gateway/platforms/matrix.py`, `gateway/run.py`, `hermes_cli/config.py`, `pyproject.toml`, `scripts/release.py`, `tests/e2e/matrix_xsign_bootstrap/README.md`, `tests/e2e/matrix_xsign_bootstrap/docker-compose.yml`, `tests/e2e/matrix_xsign_bootstrap/test_bootstrap.py`, `tests/gateway/test_matrix.py`, `tests/gateway/test_matrix_exec_approval.py`, `tests/gateway/test_matrix_mention.py`, `tests/gateway/test_platform_base.py`
- **#27330: fix(matrix): warn on clock-skew silent message drops (#12614)**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `gateway/platforms/matrix.py`, `tests/gateway/test_matrix.py`, `website/docs/user-guide/messaging/matrix.md`
- **#54031: fix(send_message): reuse live gateway adapter for Matrix media sends (#46310)**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `tests/tools/test_send_message_tool.py`, `tools/send_message_tool.py`
- **#56345: fix(matrix): await inbound sync handlers**. merged to main · fix tier `verified`
- maintainer direction: category `merged`; implication `superseded`
- files: `plugins/platforms/matrix/adapter.py`, `scripts/release.py`, `tests/gateway/test_matrix.py`
- fix hunks:
- `plugins/platforms/matrix/adapter.py`
```diff
@@ -1459,9 +1459,21 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
# Without this the INVITE handler below never fires.
client.add_dispatcher(MembershipEventDispatcher)
- client.add_event_handler(EventType.ROOM_MESSAGE, self._on_room_message)
- client.add_event_handler(EventType.REACTION, self._on_reaction)
- client.add_event_handler(IntEvt.INVITE, self._on_invite)
+ client.add_event_handler(
+ EventType.ROOM_MESSAGE,
+ self._on_room_message,
+ wait_sync=True,
+ )
+ client.add_event_handler(
+ EventType.REACTION,
+ self._on_reaction,
+ wait_sync=True,
+ )
+ client.add_event_handler(
+ IntEvt.INVITE,
+ self._on_invite,
+ wait_sync=True,
+ )
# Initial sync to catch up, then start background sync.
self._startup_ts = time.time()
```
- `scripts/release.py`
```diff
@@ -232,6 +232,7 @@
"[redacted:email]": "synapsesx",
"[redacted:email]": "itsflownium",
"[redacted:email]": "dirtyren",
+ "[redacted:email]": "yyzquwu", # PR #47567 salvage (Matrix: register inbound handlers with wait_sync=True so _dispatch_sync's gather awaits them; without it mautrix fire-and-forgets and inbound intake has no completion point)
"[redacted:email]": "tgmerritt", # PR #43553 salvage (parse vLLM's token-based output-cap error format so over-cap max_tokens 400s reduce the output cap instead of death-looping into compression)
"[redacted:email]": "justin-cyhuang",
"[redacted:email]": "Tranquil-Flow",
```
- `tests/gateway/test_matrix.py`
```diff
@@ -106,7 +106,7 @@ def __init__(self, mxid=None, device_id=None, api=None,
self.crypto = None
self._event_handlers = {}
- def add_event_handler(self, event_type, handler):
+ def add_event_handler(self, event_type, handler, **kwargs):
self._event_handlers.setdefault(event_type, []).append(handler)
def add_dispatcher(self, dispatcher_type):
```
```diff
@@ -2727,13 +2727,20 @@ async def test_connect_registers_encrypted_event_handler_when_encryption_on(self
with patch.object(adapter, "_sync_loop", AsyncMock(return_value=None)):
assert await adapter.connect() is True
- # Verify event handlers were registered.
- # In mautrix the order is: add_event_handler(EventType, callback)
+ # Verify inbound event handlers were registered as sync-awaited
+ # callbacks. mautrix only returns waited handler tasks from
+ # handle_sync(), so background-only handlers leave _dispatch_sync()
+ # without a completion point for Hermes' Matrix intake.
handler_calls = mock_client.add_event_handler.call_args_list
- registered_types = [call.args[0] for call in handler_calls]
+ waited_types = {
+ str(call.args[0])
+ for call in handler_calls
+ if call.kwargs.get("wait_sync") is True
+ }
- # Should have registered handlers for ROOM_MESSAGE, REACTION, INVITE
- assert len(handler_calls) >= 3
+ assert "m.room.message" in waited_types
+ assert "m.reaction" in waited_types
+ assert "internal.invite" in waited_types
await adapter.disconnect()
```
- **#5829: fix: apply login credentials to client after Matrix password auth**. closed without merge
- maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `6101be6db40e9e9d1aa77a92293934db5987ee66`, `#5819`, `#7881`, `#7881`, `#5819`
- **#5830: fix(matrix): add sync timeout, callback diagnostics, and mention-drop logging**. closed without merge
- maintainer direction: category `merged_via`; implication `superseded`; refs `f223346eb`, `#16821`, `#5819`
- **#8026: fix(gateway): use datetime cutoff in suspend_recently_active**. closed without merge
- maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `50d86b3c7`, `50d86b3c71f274ed6ff489cdb68a7393377f3d1c`, `#7981`, `#7966`, `#8002`
- **#8100: fix(gateway): use datetime cutoff in suspend_recently_active**. closed without merge
- maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `#7966`
- **#8215: fix(gateway): fix Matrix E2EE silent failures after mautrix migration**. closed without merge
- maintainer direction: category `silent_close`; implication `active`
- **#10135: fix(matrix): register MembershipEventDispatcher to fix auto-join on DM invites**. closed without merge
- maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `d38b73fa5`, `d38b73fa57b19b11efaeaf6256c05cc2d4f7387f`, `#10860`, `#10860`, `#10135`, `#10094`
- **#10732: fix(matrix): register MembershipEventDispatcher for invite auto-join**. closed without merge
- maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `d38b73fa5`, `d38b73fa5`, `#10860`, `#10860`, `#10732`
- **#10803: fix(matrix): register MembershipEventDispatcher for invite auto-join**. closed without merge
- maintainer direction: category `implemented_elsewhere`; implication `superseded`; refs `d38b73fa5`, `d38b73fa5`, `#10860`, `#10860`, `#10725`
- **#12803: fix(matrix): improve fresh-room detection and harden sync/timestamp handling**. closed without merge
- maintainer direction: category `uncategorized`; implication `needs_llm`; refs `#12614`, `#27330`
- **#37807: fix(gateway): start Matrix mautrix syncer**. closed without merge
- maintainer direction: category `stale_abandoned`; implication `active`; refs `#7914`, `#18505`, `#18505`
- **#37991: fix(gateway): start Matrix client syncer**. closed without merge
- maintainer direction: category `stale_abandoned`; implication `active`; refs `#37807`, `#7914`, `#18505`
- **#40313: fix(gateway): start matrix client syncer**. closed without merge
- maintainer direction: category `volume_quota`; implication `active`
- **#46365: fix(send_message): reuse live gateway adapter for Matrix media sends (#46310)**. closed without merge
- maintainer direction: category `merged_via`; implication `superseded`; refs `a7fd62d82`, `#54031`, `#41112`, `#46310`
- **#46407: fix(matrix): reuse single adapter for multi-chunk media sends (#46310)**. closed without merge
- maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#54031`, `#46310`, `#46365`
- **#47567: fix(matrix): await inbound sync handlers**. closed without merge
- maintainer direction: category `merged_via`; implication `superseded`; refs `4c2c54c78`, `#56345`
## Fix status
- Merged to main in PR #7881 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-07-01`.
- Subsystem: `gateway`