From f2387884a3f2e28274a8be98fa195c7124ab1e4a Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:11:41 +0800 Subject: [PATCH] =?UTF-8?q?refactor(matrix):=20=E6=95=B4=E7=90=86=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E5=AF=BC=E5=85=A5=E9=A1=BA=E5=BA=8F=E4=B8=8E=E5=86=97?= =?UTF-8?q?=E4=BD=99=E7=A9=BA=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 移除多个文件中多余的空行 2. 调整sync_store和allowlist的导入顺序 3. 重新排列monitor模块的导入分组 4. 重新排序__init__.py中的导出导入顺序，统一格式 5. 调整normalizer模块的导入顺序 --- .../yuxi/channels/adapters/matrix/__init__.py | 132 +++++++++--------- .../yuxi/channels/adapters/matrix/accounts.py | 99 +++++++++++++ .../yuxi/channels/adapters/matrix/adapter.py | 46 +++++- .../yuxi/channels/adapters/matrix/doctor.py | 3 +- .../yuxi/channels/adapters/matrix/monitor.py | 8 +- .../channels/adapters/matrix/normalizer.py | 6 +- .../yuxi/channels/adapters/matrix/sticker.py | 1 - .../channels/adapters/matrix/sync_store.py | 1 - 8 files changed, 213 insertions(+), 83 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/matrix/__init__.py b/backend/package/yuxi/channels/adapters/matrix/__init__.py index ef72d22f..c2464d9a 100644 --- a/backend/package/yuxi/channels/adapters/matrix/__init__.py +++ b/backend/package/yuxi/channels/adapters/matrix/__init__.py @@ -1,89 +1,89 @@ +from yuxi.channels.adapters.matrix.accounts import MatrixAccount, MultiAccountManager +from yuxi.channels.adapters.matrix.acp_bindings import ACPBinding, ACPBindingManager from yuxi.channels.adapters.matrix.adapter import MatrixAdapter -from yuxi.channels.adapters.matrix.reaction_utils import ( - build_reaction, - extract_reaction, - summarize_reactions, - select_reaction, - build_batch_remove, +from yuxi.channels.adapters.matrix.cli import MatrixCLI, format_cli_help, get_cli_commands +from yuxi.channels.adapters.matrix.commands import ( + detect_slash_command, + format_help_text, + get_command_list, + is_slash_command, +) +from yuxi.channels.adapters.matrix.concurrency import AccountDataWriteQueue, AsyncLock, StartupSerialLock +from yuxi.channels.adapters.matrix.config_patch import MatrixAccountPatch, apply_patch, diff_config, merge_patches +from yuxi.channels.adapters.matrix.config_schema import get_config_schema, get_default_config, validate_config +from yuxi.channels.adapters.matrix.config_ui_hints import get_config_hints, get_group_hints, get_groups, get_hint +from yuxi.channels.adapters.matrix.credentials import CredentialStore, MatrixCredentials +from yuxi.channels.adapters.matrix.directory import ( + list_public_rooms, + resolve_room_alias, + search_groups, + search_users, +) +from yuxi.channels.adapters.matrix.doctor import MatrixDoctor +from yuxi.channels.adapters.matrix.draft_stream import ( + DraftStreamSession, + send_draft_update, + send_live_marker, ) from yuxi.channels.adapters.matrix.encryption import ( - check_crypto_store_ready, - format_encryption_unavailable_error, - resolve_encryption_config_path, - decrypt_event, - verify_device, check_cross_signing, + check_crypto_store_ready, + decrypt_event, + format_encryption_unavailable_error, get_crypto_bootstrap_guide, + resolve_encryption_config_path, + verify_device, ) -from yuxi.channels.adapters.matrix.concurrency import AsyncLock, AccountDataWriteQueue, StartupSerialLock -from yuxi.channels.adapters.matrix.config_patch import MatrixAccountPatch, apply_patch, merge_patches, diff_config +from yuxi.channels.adapters.matrix.env_vars import get_env_var_descriptions, resolve_env_vars +from yuxi.channels.adapters.matrix.exec_approvals import ExecApprovalManager, ExecApprovalRequest from yuxi.channels.adapters.matrix.group_mentions import GroupMentionStrategy, build_room_mention_config -from yuxi.channels.adapters.matrix.profile_sync import ( - resolve_avatar_url, - resolve_display_name, - compare_profiles, - build_profile_update, - mxc_to_http, +from yuxi.channels.adapters.matrix.location import ( + build_outbound_location, + extract_geo_coordinates, + format_location_text, + parse_location_event, +) +from yuxi.channels.adapters.matrix.media_meta import extract_duration_bytes, extract_media_metadata +from yuxi.channels.adapters.matrix.polls import ( + create_poll, + end_poll, + extract_poll_response, + extract_poll_start, + is_poll_end, + send_poll_response, + summarize_poll_responses, ) from yuxi.channels.adapters.matrix.probe import ( describe_backup_state, is_backup_healthy, probe_matrix, ) -from yuxi.channels.adapters.matrix.polls import ( - create_poll, - send_poll_response, - end_poll, - summarize_poll_responses, - extract_poll_start, - extract_poll_response, - is_poll_end, +from yuxi.channels.adapters.matrix.profile_sync import ( + build_profile_update, + compare_profiles, + mxc_to_http, + resolve_avatar_url, + resolve_display_name, ) -from yuxi.channels.adapters.matrix.location import ( - parse_location_event, - format_location_text, - extract_geo_coordinates, - build_outbound_location, +from yuxi.channels.adapters.matrix.reaction_utils import ( + build_batch_remove, + build_reaction, + extract_reaction, + select_reaction, + summarize_reactions, ) +from yuxi.channels.adapters.matrix.sent_cache import ResponseSizeLimiter, SentMessageCache, hash_content +from yuxi.channels.adapters.matrix.setup_wizard import MatrixSetupWizard from yuxi.channels.adapters.matrix.sticker import ( + build_outbound_sticker, is_sticker_event, parse_sticker, - build_outbound_sticker, ) -from yuxi.channels.adapters.matrix.accounts import MatrixAccount, MultiAccountManager -from yuxi.channels.adapters.matrix.thread_bindings import ThreadBinding, ThreadBindingManager -from yuxi.channels.adapters.matrix.directory import ( - search_users, - search_groups, - resolve_room_alias, - list_public_rooms, -) -from yuxi.channels.adapters.matrix.env_vars import resolve_env_vars, get_env_var_descriptions -from yuxi.channels.adapters.matrix.sync_store import SyncTokenStore, SSRFGuard -from yuxi.channels.adapters.matrix.config_schema import validate_config, get_config_schema, get_default_config -from yuxi.channels.adapters.matrix.credentials import MatrixCredentials, CredentialStore -from yuxi.channels.adapters.matrix.sync_state import SyncState, SyncStateMachine -from yuxi.channels.adapters.matrix.sent_cache import SentMessageCache, ResponseSizeLimiter, hash_content -from yuxi.channels.adapters.matrix.voice import send_voice_note, send_tts_message, is_voice_message, extract_audio_info -from yuxi.channels.adapters.matrix.media_meta import extract_media_metadata, extract_duration_bytes -from yuxi.channels.adapters.matrix.setup_wizard import MatrixSetupWizard -from yuxi.channels.adapters.matrix.cli import MatrixCLI, get_cli_commands, format_cli_help -from yuxi.channels.adapters.matrix.doctor import MatrixDoctor from yuxi.channels.adapters.matrix.subagent_hooks import SubagentLifecycle, build_subagent_event -from yuxi.channels.adapters.matrix.exec_approvals import ExecApprovalManager, ExecApprovalRequest -from yuxi.channels.adapters.matrix.commands import ( - detect_slash_command, - is_slash_command, - get_command_list, - format_help_text, -) -from yuxi.channels.adapters.matrix.acp_bindings import ACPBinding, ACPBindingManager -from yuxi.channels.adapters.matrix.draft_stream import ( - DraftStreamSession, - send_draft_update, - send_live_marker, -) -from yuxi.channels.adapters.matrix.config_ui_hints import get_config_hints, get_hint, get_group_hints, get_groups +from yuxi.channels.adapters.matrix.sync_state import SyncState, SyncStateMachine +from yuxi.channels.adapters.matrix.sync_store import SSRFGuard, SyncTokenStore +from yuxi.channels.adapters.matrix.thread_bindings import ThreadBinding, ThreadBindingManager +from yuxi.channels.adapters.matrix.voice import extract_audio_info, is_voice_message, send_tts_message, send_voice_note __all__ = [ "MatrixAdapter", diff --git a/backend/package/yuxi/channels/adapters/matrix/accounts.py b/backend/package/yuxi/channels/adapters/matrix/accounts.py index a8547e3c..fe605d53 100644 --- a/backend/package/yuxi/channels/adapters/matrix/accounts.py +++ b/backend/package/yuxi/channels/adapters/matrix/accounts.py @@ -1,7 +1,106 @@ from __future__ import annotations +import asyncio +from contextlib import asynccontextmanager +from enum import Enum, auto from typing import Any +from loguru import logger + + +class ClientLifecycleState(Enum): + NONE = auto() + PREPARED = auto() + STARTED = auto() + STOPPED = auto() + ERROR = auto() + + +class ActiveClientTracker: + def __init__(self): + self._client_map: dict[str, Any] = {} + self._states: dict[str, ClientLifecycleState] = {} + self._sync_tasks: dict[str, asyncio.Task] = {} + self._lock = asyncio.Lock() + + @property + def active_count(self) -> int: + return sum(1 for s in self._states.values() if s == ClientLifecycleState.STARTED) + + def register( + self, + account_id: str, + client: Any, + state: ClientLifecycleState = ClientLifecycleState.PREPARED, + ) -> None: + self._client_map[account_id] = client + self._states[account_id] = state + + def unregister(self, account_id: str) -> None: + self._client_map.pop(account_id, None) + self._states.pop(account_id, None) + self._sync_tasks.pop(account_id, None) + + def get_client(self, account_id: str) -> Any | None: + return self._client_map.get(account_id) + + def get_state(self, account_id: str) -> ClientLifecycleState: + return self._states.get(account_id, ClientLifecycleState.NONE) + + def transition(self, account_id: str, target: ClientLifecycleState) -> bool: + current = self.get_state(account_id) + valid_transitions = { + ClientLifecycleState.NONE: {ClientLifecycleState.PREPARED}, + ClientLifecycleState.PREPARED: {ClientLifecycleState.STARTED, ClientLifecycleState.ERROR}, + ClientLifecycleState.STARTED: {ClientLifecycleState.STOPPED, ClientLifecycleState.ERROR}, + ClientLifecycleState.STOPPED: {ClientLifecycleState.PREPARED}, + ClientLifecycleState.ERROR: {ClientLifecycleState.PREPARED, ClientLifecycleState.STOPPED}, + } + if target not in valid_transitions.get(current, set()): + logger.warning(f"[Matrix] Invalid state transition for '{account_id}': {current.name} -> {target.name}") + return False + self._states[account_id] = target + return True + + def set_sync_task(self, account_id: str, task: asyncio.Task) -> None: + self._sync_tasks[account_id] = task + + def list_active(self) -> list[str]: + return [aid for aid, s in self._states.items() if s == ClientLifecycleState.STARTED] + + def list_all_ids(self) -> list[str]: + return list(self._client_map.keys()) + + +@asynccontextmanager +async def account_client_context(account_id: str, tracker: ActiveClientTracker, client_factory): + if not tracker.transition(account_id, ClientLifecycleState.PREPARED): + raise RuntimeError(f"Cannot prepare client for account '{account_id}'") + + client = None + try: + client = client_factory(account_id) + tracker.register(account_id, client, ClientLifecycleState.PREPARED) + + if not tracker.transition(account_id, ClientLifecycleState.STARTED): + raise RuntimeError(f"Cannot start client for account '{account_id}'") + + yield client + + except Exception: + tracker.transition(account_id, ClientLifecycleState.ERROR) + raise + finally: + try: + tracker.transition(account_id, ClientLifecycleState.STOPPED) + except Exception: + pass + if client is not None: + try: + await client.close() + except Exception: + pass + class MatrixAccount: def __init__(self, account_id: str, config: dict[str, Any]): diff --git a/backend/package/yuxi/channels/adapters/matrix/adapter.py b/backend/package/yuxi/channels/adapters/matrix/adapter.py index a4272e26..5b8c11dc 100644 --- a/backend/package/yuxi/channels/adapters/matrix/adapter.py +++ b/backend/package/yuxi/channels/adapters/matrix/adapter.py @@ -6,20 +6,20 @@ from typing import Any, ClassVar from nio import ( AsyncClient, DownloadResponse, + LocalProtocolError, LoginError, LoginResponse, - LocalProtocolError, SyncError, ) from yuxi.channels.base import BaseChannelAdapter from yuxi.channels.capabilities import ChannelCapabilities -from yuxi.channels.meta import ChannelMeta -from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError from yuxi.channels.exceptions import ( ChannelAuthenticationError, ChannelException, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelMessage, ChannelResponse, @@ -32,10 +32,12 @@ from yuxi.channels.registry import register_builtin_adapter from yuxi.utils.datetime_utils import utc_now_naive from yuxi.utils.logging_config import logger +from .acp_bindings import ACPBindingManager from .auto_join import should_auto_join from .config_patch import MatrixAccountPatch, apply_patch from .crypto import import_room_keys from .encryption import check_crypto_store_ready, format_encryption_unavailable_error +from .exec_approvals import ExecApprovalManager from .group_mentions import GroupMentionStrategy from .monitor import MatrixMonitor from .normalizer import MatrixNormalizer @@ -53,10 +55,8 @@ from .send import ( send_thread_message, ) from .session import MatrixSessionHelper -from .sync_store import SSRFGuard, SyncTokenStore -from .acp_bindings import ACPBindingManager from .subagent_hooks import SubagentLifecycle -from .exec_approvals import ExecApprovalManager +from .sync_store import SSRFGuard, SyncTokenStore _DIRECT_ROOM_TYPE = "m.direct" @@ -115,6 +115,7 @@ class MatrixAdapter(BaseChannelAdapter): self._exec_approvals = ExecApprovalManager(self.config) crypto_store_dir = self.config.get("crypto_store_dir", "./matrix_crypto_store") self._sync_token_store = SyncTokenStore(crypto_store_dir, self.config.get("user_id", "unknown")) + self._use_msc4357: bool = False @property def user_id(self) -> str: @@ -229,6 +230,8 @@ class MatrixAdapter(BaseChannelAdapter): self._homeserver = homeserver logger.info(f"Matrix adapter connected: {user_id} @ {homeserver}") + self._use_msc4357 = await self._detect_msc4357_support() + async def disconnect(self) -> None: self._status = ChannelStatus.DISCONNECTED self._ready_event.clear() @@ -412,6 +415,20 @@ class MatrixAdapter(BaseChannelAdapter): if not self._client: return DeliveryResult(success=False, error="Client not initialized") + if self._use_msc4357 and not finished: + from yuxi.channels.adapters.matrix.draft_stream import send_draft_update + + async def _do_send(): + return await send_draft_update( + self._client, + chat_id, + chunk, + draft_event_id=msg_id or None, + finished=finished, + ) + + return await self._send_with_circuit(_do_send) + async def _do_send(): return await send_stream_chunk( self._client, @@ -424,6 +441,23 @@ class MatrixAdapter(BaseChannelAdapter): return await self._send_with_circuit(_do_send) + async def _detect_msc4357_support(self) -> bool: + if not self._client: + return False + + try: + resp = await self._client.versions() + unstable_features = resp.unstable_features or {} + supported = "org.matrix.msc4357.live" in unstable_features + if supported: + logger.info("[Matrix] MSC4357 live marker supported by server") + else: + logger.debug("[Matrix] MSC4357 not advertised by server, using m.replace fallback") + return supported + except Exception: + logger.debug("[Matrix] Failed to probe MSC4357, using m.replace fallback") + return False + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: if not self._client: return DeliveryResult(success=False, error="Client not initialized") diff --git a/backend/package/yuxi/channels/adapters/matrix/doctor.py b/backend/package/yuxi/channels/adapters/matrix/doctor.py index 21774c13..d86789fa 100644 --- a/backend/package/yuxi/channels/adapters/matrix/doctor.py +++ b/backend/package/yuxi/channels/adapters/matrix/doctor.py @@ -3,13 +3,12 @@ from __future__ import annotations import os from typing import TYPE_CHECKING, Any - if TYPE_CHECKING: from nio import AsyncClient from .config_schema import validate_config from .encryption import check_crypto_store_ready -from .sync_store import SyncTokenStore, SSRFGuard +from .sync_store import SSRFGuard, SyncTokenStore class MatrixDoctor: diff --git a/backend/package/yuxi/channels/adapters/matrix/monitor.py b/backend/package/yuxi/channels/adapters/matrix/monitor.py index 8fa8a904..72cd3c0b 100644 --- a/backend/package/yuxi/channels/adapters/matrix/monitor.py +++ b/backend/package/yuxi/channels/adapters/matrix/monitor.py @@ -1,10 +1,9 @@ from __future__ import annotations import asyncio +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime from typing import TYPE_CHECKING, Any -from collections.abc import Callable, Awaitable - -from datetime import datetime, UTC from nio import ( RoomMemberEvent, @@ -24,12 +23,13 @@ from yuxi.channels.models import ( ) from yuxi.utils.logging_config import logger +from .allowlist import match_user_in_list from .dedupe import EventDeduplicator from .security import MatrixSecurityPolicy -from .allowlist import match_user_in_list if TYPE_CHECKING: from nio import AsyncClient + from yuxi.channels.adapters.matrix.adapter import MatrixAdapter _MESSAGE_SYNC_FILTER = { diff --git a/backend/package/yuxi/channels/adapters/matrix/normalizer.py b/backend/package/yuxi/channels/adapters/matrix/normalizer.py index 06cfb9dd..5b4530ef 100644 --- a/backend/package/yuxi/channels/adapters/matrix/normalizer.py +++ b/backend/package/yuxi/channels/adapters/matrix/normalizer.py @@ -1,7 +1,7 @@ from __future__ import annotations import re -from datetime import datetime, UTC +from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from nio import ( @@ -24,10 +24,10 @@ from yuxi.channels.models import ( MessageType, ) -from .rooms_config import get_room_require_mention -from .thread_utils import is_thread_event from .location import parse_location_event +from .rooms_config import get_room_require_mention from .sticker import is_sticker_event, parse_sticker +from .thread_utils import is_thread_event if TYPE_CHECKING: from yuxi.channels.adapters.matrix.adapter import MatrixAdapter diff --git a/backend/package/yuxi/channels/adapters/matrix/sticker.py b/backend/package/yuxi/channels/adapters/matrix/sticker.py index d13c5b95..8b34576d 100644 --- a/backend/package/yuxi/channels/adapters/matrix/sticker.py +++ b/backend/package/yuxi/channels/adapters/matrix/sticker.py @@ -2,7 +2,6 @@ from __future__ import annotations from typing import Any - _STICKER_CONTENT_KEY = "body" _STICKER_INFO_KEY = "info" diff --git a/backend/package/yuxi/channels/adapters/matrix/sync_store.py b/backend/package/yuxi/channels/adapters/matrix/sync_store.py index 125ec5c1..d2b2d63d 100644 --- a/backend/package/yuxi/channels/adapters/matrix/sync_store.py +++ b/backend/package/yuxi/channels/adapters/matrix/sync_store.py @@ -1,6 +1,5 @@ from __future__ import annotations - _SYNC_TOKEN_FILE = "matrix_sync_token.json"