fix(matrix): E2EE and migration bugfixes (#10860)
* - make buffered streaming - fix path naming to expand `~` for agent. - fix stripping of matrix ID to not remove other mentions / localports. * fix(matrix): register MembershipEventDispatcher for invite auto-join The mautrix migration (#7518) broke auto-join because InternalEventType.INVITE events are only dispatched when MembershipEventDispatcher is registered on the client. Without it, _on_invite is dead code and the bot silently ignores all room invites. Closes #10094 Closes #10725 Refs: PR #10135 (digging-airfare-4u), PR #10732 (fxfitz) * fix(matrix): preserve _joined_rooms reference for CryptoStateStore connect() reassigned self._joined_rooms = set(...) after initial sync, orphaning the reference captured by _CryptoStateStore at init time. find_shared_rooms() returned [] forever, breaking Megolm session rotation on membership changes. Mutate in place with clear() + update() so the CryptoStateStore reference stays valid. Refs #8174, PR #8215 * fix(matrix): remove dual ROOM_ENCRYPTED handler to fix dedup race mautrix auto-registers DecryptionDispatcher when client.crypto is set. The adapter also registered _on_encrypted_event for the same event type. _on_encrypted_event had zero awaits and won the race to mark event IDs in the dedup set, causing _on_room_message to drop successfully decrypted events from DecryptionDispatcher. The retry loop masked this by re-decrypting every message ~4 seconds later. Remove _on_encrypted_event entirely. DecryptionDispatcher handles decryption; genuinely undecryptable events are logged by mautrix and retried on next key exchange. Refs #8174, PR #8215 * fix(matrix): re-verify device keys after share_keys() upload Matrix homeservers treat ed25519 identity keys as immutable per device. share_keys() can return 200 but silently ignore new keys if the device already exists with different identity keys. The bot would proceed with shared=True while peers encrypt to the old (unreachable) keys. Now re-queries the server after share_keys() and fails closed if keys don't match, with an actionable error message. Refs #8174, PR #8215 * fix(matrix): encrypt outbound attachments in E2EE rooms _upload_and_send() uploaded raw bytes and used the 'url' key for all rooms. In E2EE rooms, media must be encrypted client-side with encrypt_attachment(), the ciphertext uploaded, and the 'file' key (with key/iv/hashes) used instead of 'url'. Now detects encrypted rooms via state_store.is_encrypted() and branches to the encrypted upload path. Refs: PR #9822 (charles-brooks) * fix(matrix): add stop_typing to clear typing indicator after response The adapter set a 30-second typing timeout but never cleared it. The base class stop_typing() is a no-op, so the typing indicator lingered for up to 30 seconds after each response. Closes #6016 Refs: PR #6020 (r266-tech) * fix(matrix): cache all media types locally, not just photos/voice should_cache_locally only covered PHOTO, VOICE, and encrypted media. Unencrypted audio/video/documents in plaintext rooms were passed as MXC URLs that require authentication the agent doesn't have, resulting in 401 errors. Refs #3487, #3806 * fix(matrix): detect stale OTK conflict on startup and fail closed When crypto state is wiped but the same device ID is reused, the homeserver may still hold one-time keys signed with the previous identity key. Identity key re-upload succeeds but OTK uploads fail with "already exists" and a signature mismatch. Peers cannot establish new Olm sessions, so all new messages are undecryptable. Now proactively flushes OTKs via share_keys() during connect() and catches the "already exists" error with an actionable log message telling the operator to purge the device from the homeserver or generate a fresh device ID. Also documents the crypto store recovery procedure in the Matrix setup guide. Refs #8174 * docs(matrix): improve crypto recovery docs per review - Put easy path (fresh access token) first, manual purge second - URL-encode user ID in Synapse admin API example - Note that device deletion may invalidate the access token - Add "stop Synapse first" caveat for direct SQLite approach - Mention the fail-closed startup detection behavior - Add back-reference from upgrade section to OTK warning * refactor(matrix): cleanup from code review - Extract _extract_server_ed25519() and _reverify_keys_after_upload() to deduplicate the re-verification block (was copy-pasted in two places, three copies of ed25519 key extraction total) - Remove dead code: _pending_megolm, _retry_pending_decryptions, _MAX_PENDING_EVENTS, _PENDING_EVENT_TTL — all orphaned after removing _on_encrypted_event - Remove tautological TestMediaCacheGate (tested its own predicate, not production code) - Remove dead TestMatrixMegolmEventHandling and TestMatrixRetryPendingDecryptions (tested removed methods) - Merge duplicate TestMatrixStopTyping into TestMatrixTypingIndicator - Trim comment to just the "why"
This commit is contained in:
Siddharth Balyan
@@ -1013,3 +1013,106 @@ class TestFilterAndAccumulateIntegration:
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user