From 977724d7c30d57eb68e59ede3f5537f6cbd0cb6c Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Sat, 11 Jul 2026 05:42:26 +0800 Subject: [PATCH] =?UTF-8?q?refactor(wechat-woc,transport):=20=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E4=BB=A3=E7=A0=81=E4=BC=98=E5=8C=96=E4=B8=8E=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E5=A2=9E=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 修复代码格式与缩进问题，统一代码风格 2. 调整凭证缓存TTL为不过期，显式管理缓存失效 3. 重构wechat-woc出站请求，添加可重入锁保证同账号串行调用 4. 新增wechat-woc向导适配器配置端口支持，兼容开发环境localhost 5. 实现入站消息出站投递异步化，添加账号级并发限制 6. 新增transport启动前凭证缓存预热逻辑 7. 调整导入顺序与依赖位置，优化代码结构 --- .../channels/adapters/agent_run_adapter.py | 5 +- .../channels/adapters/conversation_adapter.py | 12 +- .../handlers/content_review_handler.py | 2 +- .../inbound/inbound_idempotency_stage.py | 4 +- .../pipeline/outbound/load_build_stage.py | 6 +- .../pipeline/outbound/stream_chunk_stage.py | 10 +- .../application/transport/base_worker.py | 4 +- .../channels/application/transport/manager.py | 86 +++++++--- .../application/transport/stream_worker.py | 4 +- .../usecase/admin_message_service.py | 2 +- .../usecase/inbound_message_service.py | 150 ++++++++++++++---- .../core/service/credential_service.py | 10 +- .../yuxi/channels/infrastructure/factory.py | 1 + .../package/yuxi/channels/plugins/README.md | 2 +- .../wechat_woc/adapters/outbound_adapter.py | 2 +- .../adapters/stream_connector_adapter.py | 6 +- .../wechat_woc/adapters/wizard_adapter.py | 55 +++++-- .../yuxi/channels/plugins/wechat_woc/entry.py | 9 +- .../plugins/wechat_woc/woc_bridge_client.py | 101 +++++++++--- 19 files changed, 356 insertions(+), 115 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/agent_run_adapter.py b/backend/package/yuxi/channels/adapters/agent_run_adapter.py index b2d594de..f1fe0652 100644 --- a/backend/package/yuxi/channels/adapters/agent_run_adapter.py +++ b/backend/package/yuxi/channels/adapters/agent_run_adapter.py @@ -26,6 +26,9 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Any +from sqlalchemy import select +from sqlalchemy.exc import SQLAlchemyError + from yuxi.channels.contract.dtos.agent_run import AgentRunCmd, AgentRunId from yuxi.channels.contract.dtos.common import MessageContent from yuxi.channels.contract.dtos.option import Nothing, Option, Some @@ -44,8 +47,6 @@ from yuxi.channels.contract.ports.driven.logger_port import LoggerPort from yuxi.channels.contract.ports.driven.service_account_port import ServiceAccountPort from yuxi.storage.postgres.manager import pg_manager from yuxi.storage.postgres.models_business import Conversation -from sqlalchemy import select -from sqlalchemy.exc import SQLAlchemyError if TYPE_CHECKING: from yuxi.channels.contract.ports.driven.transaction_port import TransactionContext diff --git a/backend/package/yuxi/channels/adapters/conversation_adapter.py b/backend/package/yuxi/channels/adapters/conversation_adapter.py index 5b16bc09..39fac220 100644 --- a/backend/package/yuxi/channels/adapters/conversation_adapter.py +++ b/backend/package/yuxi/channels/adapters/conversation_adapter.py @@ -1398,8 +1398,8 @@ class ConversationAdapter(ConversationPort): row = result.one_or_none() if row is None: return None - message = self._orm_to_message(row.MessageORM) - return self._attach_context(message, row.ConversationORM, row[2]) + message = self._orm_to_message(row.Message) + return self._attach_context(message, row.Conversation, row[2]) except IntegrityError as exc: raise self._translate_db_error(exc, "message") from exc except SQLAlchemyError as exc: @@ -1504,8 +1504,8 @@ class ConversationAdapter(ConversationPort): result = await self._db.execute(stmt) messages: list[Message] = [] for row in result.all(): - message = self._orm_to_message(row.MessageORM) - self._attach_context(message, row.ConversationORM, row[2]) + message = self._orm_to_message(row.Message) + self._attach_context(message, row.Conversation, row[2]) messages.append(message) return tuple(messages) except IntegrityError as exc: @@ -1812,8 +1812,8 @@ class ConversationAdapter(ConversationPort): result = await self._db.execute(stmt) items: list[MessageSearchItem] = [] for row in result.all(): - msg = row.MessageORM - conv = row.ConversationORM + msg = row.Message + conv = row.Conversation sess = row[2] items.append( MessageSearchItem( diff --git a/backend/package/yuxi/channels/application/pipeline/control_plane/handlers/content_review_handler.py b/backend/package/yuxi/channels/application/pipeline/control_plane/handlers/content_review_handler.py index aa8c1fba..aa732016 100644 --- a/backend/package/yuxi/channels/application/pipeline/control_plane/handlers/content_review_handler.py +++ b/backend/package/yuxi/channels/application/pipeline/control_plane/handlers/content_review_handler.py @@ -19,7 +19,6 @@ from datetime import UTC, datetime from typing import Any from uuid import uuid4 -from yuxi.utils.datetime_utils import utc_now_naive from yuxi.channels.application.context.control_plane_context import ( ControlPlaneContext, ) @@ -54,6 +53,7 @@ from yuxi.channels.contract.ports.driven.content_review_repository_port import ( from yuxi.channels.contract.ports.driven.logger_port import LoggerPort from yuxi.channels.contract.ports.driven.persistence_port import PersistencePort from yuxi.channels.contract.ports.driven.transaction_port import TransactionPort +from yuxi.utils.datetime_utils import utc_now_naive # 内容审核单条内容长度上界 MAX_CONTENT_LENGTH: int = 10000 diff --git a/backend/package/yuxi/channels/application/pipeline/inbound/inbound_idempotency_stage.py b/backend/package/yuxi/channels/application/pipeline/inbound/inbound_idempotency_stage.py index 6d6c488a..f77e3365 100644 --- a/backend/package/yuxi/channels/application/pipeline/inbound/inbound_idempotency_stage.py +++ b/backend/package/yuxi/channels/application/pipeline/inbound/inbound_idempotency_stage.py @@ -196,7 +196,9 @@ class InboundIdempotencyStage: existing_resource_id=str(record.record_id), trace_id=context.trace_id, ) - if record.status == "in_progress" and not _isStaleInProgress(record, _IDEMPOTENCY_TIMEOUT_SECONDS): + if record.status == "in_progress" and not _isStaleInProgress( + record, _IDEMPOTENCY_TIMEOUT_SECONDS + ): # 锁内重查发现已被其他请求重建为新的非 stale # in_progress 记录：抛冲突让渠道重试 raise IdempotencyConflictError( diff --git a/backend/package/yuxi/channels/application/pipeline/outbound/load_build_stage.py b/backend/package/yuxi/channels/application/pipeline/outbound/load_build_stage.py index b81179b3..32e2df9b 100644 --- a/backend/package/yuxi/channels/application/pipeline/outbound/load_build_stage.py +++ b/backend/package/yuxi/channels/application/pipeline/outbound/load_build_stage.py @@ -98,11 +98,7 @@ class LoadBuildStage: # condition 跳过）。阻塞消费事件流等待 AgentRun 完成后加载完整输出， # 由下游 outbox-persist/deliver 投递。流式模式下 stream_chunk_stage # 负责加载+投递，此处不介入。 - if ( - context.delivery_mode == "persistent" - and context.agent_run_id - and not context.stream_chunks - ): + if context.delivery_mode == "persistent" and context.agent_run_id and not context.stream_chunks: await self._loadAgentRunOutput(context) rich_message_fields = None diff --git a/backend/package/yuxi/channels/application/pipeline/outbound/stream_chunk_stage.py b/backend/package/yuxi/channels/application/pipeline/outbound/stream_chunk_stage.py index 62079723..46d64b1c 100644 --- a/backend/package/yuxi/channels/application/pipeline/outbound/stream_chunk_stage.py +++ b/backend/package/yuxi/channels/application/pipeline/outbound/stream_chunk_stage.py @@ -41,12 +41,12 @@ from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.common import MessageContent, MessageFormat from yuxi.channels.contract.dtos.config import ConfigScope from yuxi.channels.contract.dtos.outbound import FormattedMessage +from yuxi.channels.contract.dtos.stream_event import StreamEvent from yuxi.channels.contract.dtos.streaming import ( ChunkResult, StreamChunk, StreamingCompleted, ) -from yuxi.channels.contract.dtos.stream_event import StreamEvent from yuxi.channels.contract.errors.domain import ChannelDegradedError from yuxi.channels.contract.plugin.adapters.outbound_adapter import OutboundAdapter from yuxi.channels.contract.plugin.adapters.streaming_adapter import StreamingAdapter @@ -331,9 +331,7 @@ class StreamChunkStage: context.formatted_message = FormattedMessage( content=content, format=MessageFormat.TEXT, - attachments=context.formatted_message.attachments - if context.formatted_message is not None - else (), + attachments=context.formatted_message.attachments if context.formatted_message is not None else (), ) return message_content = MessageContent(text=content, format=MessageFormat.TEXT) @@ -342,7 +340,5 @@ class StreamChunkStage: content=formatted.content, format=formatted.format, rich_message=formatted.rich_message, - attachments=context.formatted_message.attachments - if context.formatted_message is not None - else (), + attachments=context.formatted_message.attachments if context.formatted_message is not None else (), ) diff --git a/backend/package/yuxi/channels/application/transport/base_worker.py b/backend/package/yuxi/channels/application/transport/base_worker.py index 4ea9d561..10a0a5a5 100644 --- a/backend/package/yuxi/channels/application/transport/base_worker.py +++ b/backend/package/yuxi/channels/application/transport/base_worker.py @@ -942,9 +942,7 @@ class BaseTransportWorker(ABC): # 仅 StreamWorker 降级有意义（PullerWorker permanent 失败时无 # 更低优先级模式可降级）。 if self.transport_mode == "stream": - await self._publishChannelDegraded( - channel_type, account_id, error.message, trace_id - ) + await self._publishChannelDegraded(channel_type, account_id, error.message, trace_id) return True if error.category == "rate_limited": diff --git a/backend/package/yuxi/channels/application/transport/manager.py b/backend/package/yuxi/channels/application/transport/manager.py index 97405c24..290b9d35 100644 --- a/backend/package/yuxi/channels/application/transport/manager.py +++ b/backend/package/yuxi/channels/application/transport/manager.py @@ -18,16 +18,20 @@ from yuxi.channels.application.transport.base_worker import ( from yuxi.channels.application.transport.puller_worker import PullerWorker from yuxi.channels.application.transport.stream_worker import StreamWorker from yuxi.channels.contract.dtos.channel import AccountFilter, AccountStatus, ChannelType +from yuxi.channels.contract.dtos.config import ConfigScope from yuxi.channels.contract.dtos.health import TransportHealthSnapshot +from yuxi.channels.contract.dtos.option import Some from yuxi.channels.contract.dtos.plugin import DomainEvent, EventHandler from yuxi.channels.contract.plugin.extension_point import EventSubscription from yuxi.channels.contract.plugin.manifest import FailurePolicy from yuxi.channels.contract.ports.driven import ( + CachePort, ConfigPort, LoggerPort, PersistencePort, ) from yuxi.channels.core.registry import PluginRegistry +from yuxi.utils.crypto import decrypt_sensitive_fields __all__ = ["TransportManager"] @@ -61,6 +65,7 @@ class TransportManager: logger: LoggerPort, message_deliverer: Callable[[Any], Awaitable[Any]], transport_config: TransportConfig | None = None, + cache_port: CachePort | None = None, ) -> None: """初始化 TransportManager。 @@ -73,6 +78,8 @@ class TransportManager: logger: 日志端口。 message_deliverer: 入站消息投递回调。 transport_config: 传输配置，为 None 时使用默认配置。 + cache_port: 缓存端口，用于启动前预热凭证缓存。为 None 时 + 跳过预热（向后兼容）。 """ self._plugin_registry = plugin_registry self._persistence_port = persistence_port @@ -82,6 +89,7 @@ class TransportManager: self._logger = logger self._message_deliverer = message_deliverer self._config = transport_config or TransportConfig() + self._cache_port = cache_port self._puller_registry: dict[ChannelType, Any] = {} self._stream_connector_registry: dict[ChannelType, Any] = {} @@ -382,9 +390,7 @@ class TransportManager: if account_key in self._degraded_accounts: self._degraded_accounts.discard(account_key) if self._puller_worker is not None: - await self._puller_worker.stop_account( - channel_type, account_id, reason="recovered" - ) + await self._puller_worker.stop_account(channel_type, account_id, reason="recovered") await self._logger.info( "channel transport recovered from degraded mode", trace_id=trace_id, @@ -420,9 +426,7 @@ class TransportManager: 强制以 pull 模式启动 PullerWorker 使用。 """ transport_mode = ( - force_mode - if force_mode is not None - else await self._resolveTransportMode(channel_type, account_id) + force_mode if force_mode is not None else await self._resolveTransportMode(channel_type, account_id) ) puller_adapter = self._puller_registry.get(channel_type) stream_adapter = self._stream_connector_registry.get(channel_type) @@ -449,6 +453,11 @@ class TransportManager: # 等原因未真正启动 task，状态轻微不一致可接受（下次状态变化时纠正）。 await self._touchPluginStatus(channel_type, account_id, "running", trace_id) + # 启动前预热凭证缓存：CachePort 命中则跳过，未命中从 ConfigPort 解密回填。 + # 覆盖重启恢复与事件驱动路径，确保 worker 首次 poll 时凭证已就绪。 + if self._cache_port is not None: + await self._preheatCredentials(account_id, trace_id) + if transport_mode == "pull": if puller_adapter is not None and self._puller_worker is not None: await self._puller_worker.start_account(channel_type, account_id, puller_adapter) @@ -464,6 +473,55 @@ class TransportManager: elif puller_adapter is not None and self._puller_worker is not None: await self._puller_worker.start_account(channel_type, account_id, puller_adapter) + async def _preheatCredentials(self, account_id: str, trace_id: str) -> None: + """启动前预热凭证缓存。 + + CachePort 命中则跳过；未命中从 ConfigPort 读取加密凭证，解密后 + 回填 CachePort（无 TTL）。覆盖系统重启、Redis 缓存丢失等场景， + 确保插件客户端（如 ILinkClient）首次 poll 时凭证已就绪。 + + 预热失败不阻断 transport 启动：模式 A 渠道（feishu/wecom 等）不 + 使用 ``credentials:{account_id}`` 键，ConfigPort 查不到属正常。 + """ + cache_key = f"credentials:{account_id}" + try: + cached = await self._cache_port.get(cache_key) + if isinstance(cached, Some) and cached.unwrap(): + return + except Exception as exc: + await self._logger.warn( + "credential preheat: cache read failed, continue to ConfigPort", + trace_id=trace_id, + account_id=account_id, + error=str(exc), + ) + + try: + config_value = await self._config_port.get( + "credentials", + scope=ConfigScope.ACCOUNT, + target=account_id, + ) + except Exception: + # ConfigPort 无凭证记录（模式 A 渠道或未接入凭证），跳过预热 + return + + encrypted = config_value.value + if not encrypted: + return + + try: + decrypted = decrypt_sensitive_fields(encrypted) + if decrypted: + await self._cache_port.set(cache_key, decrypted, ttl_seconds=None) + except Exception as exc: + await self._logger.warn( + "credential preheat: decrypt or cache write failed", + trace_id=trace_id, + account_id=account_id, + error=str(exc), + ) + async def _restoreOnlineAccounts(self, trace_id: str) -> None: """重启恢复：扫描 DB 中 ACTIVE 状态账号，启动传输任务。 @@ -681,9 +739,7 @@ class TransportManager: account_key = self._make_account_key(channel_type, account_id) try: if self._stream_worker is not None: - await self._stream_worker.stop_account( - channel_type, account_id, reason="degraded" - ) + await self._stream_worker.stop_account(channel_type, account_id, reason="degraded") await self._startTransportForAccount( channel_type=channel_type, account_id=account_id, @@ -813,18 +869,12 @@ class TransportManager: # 停止现有 Worker（no-op if not running） if self._puller_worker is not None: - await self._puller_worker.stop_account( - channel_type, account_id, reason="config_changed" - ) + await self._puller_worker.stop_account(channel_type, account_id, reason="config_changed") if self._stream_worker is not None: - await self._stream_worker.stop_account( - channel_type, account_id, reason="config_changed" - ) + await self._stream_worker.stop_account(channel_type, account_id, reason="config_changed") # 启动新 Worker（使用最新配置） - await self._startTransportForAccount( - channel_type, account_id, trace_id, source="config_changed" - ) + await self._startTransportForAccount(channel_type, account_id, trace_id, source="config_changed") async def _touchPluginStatus( self, diff --git a/backend/package/yuxi/channels/application/transport/stream_worker.py b/backend/package/yuxi/channels/application/transport/stream_worker.py index c780ff5f..f9445b6b 100644 --- a/backend/package/yuxi/channels/application/transport/stream_worker.py +++ b/backend/package/yuxi/channels/application/transport/stream_worker.py @@ -126,9 +126,7 @@ class StreamWorker(BaseTransportWorker): # P0-3: 每次重连前重新加载持久化游标与版本号，确保使用最新值。 # _loadInitialCursor 抛 DependencyError 时由下方 except Exception # 转译为 TransportError(transient) 触发退避重连。 - persisted_cursor, account_version = await self._loadInitialCursor( - channel_type, account_id - ) + persisted_cursor, account_version = await self._loadInitialCursor(channel_type, account_id) # cursor_holder 由适配器在 sync 事件时写入 sync cursor， # _receiveLoop 读取后批量持久化到 transport_cursor cursor_holder: dict[str, str] = {} diff --git a/backend/package/yuxi/channels/application/usecase/admin_message_service.py b/backend/package/yuxi/channels/application/usecase/admin_message_service.py index 20f0b9f4..0a3d8c2b 100644 --- a/backend/package/yuxi/channels/application/usecase/admin_message_service.py +++ b/backend/package/yuxi/channels/application/usecase/admin_message_service.py @@ -23,10 +23,10 @@ from yuxi.channels.application.context.outbound_context import OutboundContext from yuxi.channels.application.pipeline.control_plane.audit_context_builder import ( AuditContextBuilder, ) -from yuxi.channels.application.rate_limit_checker import RateLimitChecker from yuxi.channels.application.pipeline.outbound.outbound_pipeline import ( OutboundPipeline, ) +from yuxi.channels.application.rate_limit_checker import RateLimitChecker from yuxi.channels.application.usecase.failure_detail import toFailureDetail from yuxi.channels.contract.dtos.admin import AdminSendCmd, AdminSendResult from yuxi.channels.contract.dtos.channel import ChannelSession, ChannelType diff --git a/backend/package/yuxi/channels/application/usecase/inbound_message_service.py b/backend/package/yuxi/channels/application/usecase/inbound_message_service.py index d75b9ea7..1ec29921 100644 --- a/backend/package/yuxi/channels/application/usecase/inbound_message_service.py +++ b/backend/package/yuxi/channels/application/usecase/inbound_message_service.py @@ -77,6 +77,11 @@ __all__ = ["InboundMessageService"] # 当 ``ack_fallback_timeout_seconds`` 配置读取失败时使用此默认值（FR-24）。 _DEFAULT_ACK_FALLBACK_TIMEOUT_SECONDS = 60 +# 同账号出站投递并发上限，防止任务堆积导致内存溢出。 +# 不同会话的出站管道并行执行（会话级锁保证同会话串行）， +# 超出上限时新任务排队等待信号量，自然形成背压。 +_OUTBOUND_CONCURRENCY_PER_ACCOUNT = 5 + class InboundMessageService: """入站消息用例服务，实现 InboundMessagePort。 @@ -89,23 +94,22 @@ class InboundMessageService: 分阶段 ACK（FR-24）： - AFTER_RECORD / AFTER_AGENT_DISPATCH：由 ``ReplyStage`` 在入站管道 末尾决策 ACK。 - - AFTER_PERSIST：出站管道成功后由本服务通过 ``AckDecisionMaker`` - 幂等 ACK（AC-52）。 + - AFTER_PERSIST：出站管道成功后由后台投递任务通过 ``AckDecisionMaker`` + 幂等 ACK（AC-52）。出站异步化后 ACK 延迟到后台任务完成， + ``receiveInbound`` 返回时 ``ack_decision`` 为 ``pending``。 - MANUAL：保持 ``pending``，由插件控制 ACK 时机；插件忘记 ACK 时 核心有兜底超时（默认 60s，配置项 ``ack_fallback_timeout_seconds``） 自动 ACK。 - ACK 互斥：若 ``ReplyStage`` 已 ACK（``ctx.acked=True``），出站 失败时不覆盖为 NACK，保持 ACK 语义不变。 - 入站管道创建 Agent 运行后（``agent_run_id`` 非空且非静默），本服务构造 - ``OutboundContext`` 并执行出站管道，将 Agent 响应投递至渠道侧（FR-13）。 - ``delivery_mode`` 由 ``streaming_enabled`` 配置决定，启用时走流式输出 - 路径（stream-chunk / typing-indicator / typing-stop），否则走持久化路径。 - - 静默命令（``is_silent=True``）携带响应内容时，本服务构造最小 - ``OutboundContext``（``delivery_mode="persistent"``、``agent_run_id=""``） - 执行出站管道投递命令响应，不创建 AgentRun（FR-15）。出站失败时通过 - ``ctx.outbound_error`` 透传错误至 ``InboundResult``（与 ACK 互斥）。 + 出站投递异步化：入站管道完成（幂等记录 completed + ACK 已决策）后， + ``receiveInbound`` 将出站投递（``_deliverAgentResponse`` / + ``_deliverSilentCommandResponse``）通过 ``_scheduleOutboundDelivery`` + 异步触发，立即返回，释放 worker 循环接收下一条消息。出站管道在后台 + 协程中执行，不阻塞入站接收层。同会话保序由 ``outbound:{conversation_id}`` + 会话级锁保证，同账号并发出站数由 ``_outbound_semaphores`` 信号量限制 + （默认 5），形成背压。 """ def __init__( @@ -167,22 +171,36 @@ class InboundMessageService: # MANUAL 兜底 ACK 任务集合（M-15）：持有引用避免被 GC，关闭时取消 # 未完成任务，避免 sleep 60s 期间持有请求级 ctx 与连接资源。 self._manual_fallback_tasks: set[asyncio.Task] = set() + # 出站投递异步任务集合：持有引用避免被 GC，shutdown 时统一取消。 + self._outbound_delivery_tasks: set[asyncio.Task] = set() + # 按账号隔离的信号量，限制同账号并发出站数，防止任务堆积。 + # 键格式 "channel_type:account_id"，不同账号互不影响。 + self._outbound_semaphores: dict[str, asyncio.Semaphore] = {} async def shutdown(self) -> None: - """关闭用例服务，取消所有未完成的 MANUAL 兜底 ACK 任务（M-15）。 + """关闭用例服务，取消所有未完成的后台任务（M-15 + 出站投递）。 - 关闭流程取消 ``_manual_fallback_tasks`` 中未完成的任务，避免任务 - 在请求返回后仍持有 ``AckDecisionMaker`` 引用与配置端口连接。已 + 关闭流程取消 ``_manual_fallback_tasks`` 与 ``_outbound_delivery_tasks`` + 中未完成的任务，避免任务在请求返回后仍持有引用与连接资源。已 完成的任务由 ``add_done_callback`` 自动从集合中移除。 + + 出站投递任务可能正在执行出站管道（含 stream-chunk 阻塞等待 Agent + 流式事件），取消时通过 ``asyncio.CancelledError`` 中断，由各阶段 + 的 finally 块释放资源（如会话级锁）。 """ - if not self._manual_fallback_tasks: + # 合并两类待取消任务，统一 gather 避免分两次 await + to_cancel: list[asyncio.Task] = [] + if self._manual_fallback_tasks: + # 复制集合避免 cancel() 触发 done_callback 修改集合大小导致迭代异常 + to_cancel.extend(self._manual_fallback_tasks) + if self._outbound_delivery_tasks: + to_cancel.extend(self._outbound_delivery_tasks) + if not to_cancel: return - # 复制集合避免 cancel() 触发 done_callback 修改集合大小导致迭代异常 - pending = list(self._manual_fallback_tasks) - for task in pending: + for task in to_cancel: task.cancel() # 等待所有任务完成取消，忽略 CancelledError - await asyncio.gather(*pending, return_exceptions=True) + await asyncio.gather(*to_cancel, return_exceptions=True) async def receiveInbound(self, cmd: InboundMessageCmd) -> InboundResult: """接收并处理入站消息。 @@ -196,14 +214,18 @@ class InboundMessageService: 执行 ``InboundPipeline.run(ctx)``；管道失败时抛出错误触发事务 回滚（§10.1），成功时正常退出触发提交。 4. 成功且创建 Agent 运行（``agent_run_id`` 非空且非静默）时， - 调用 ``_deliverAgentResponse`` 桥接到出站管道投递 Agent 响应 - （FR-13）。 + 通过 ``_scheduleOutboundDelivery`` 异步触发 + ``_deliverAgentResponse`` 桥接到出站管道投递 Agent 响应 + （FR-13）。出站在后台协程执行，不阻塞入站接收层。 5. 静默命令（``is_silent=True``）携带响应内容（``ctx.command.response`` - 非空）时，调用 ``_deliverSilentCommandResponse`` 通过出站管道 - 投递命令响应，不创建 AgentRun（FR-15）。 + 非空）时，通过 ``_scheduleOutboundDelivery`` 异步触发 + ``_deliverSilentCommandResponse`` 投递命令响应（FR-15）。 6. MANUAL 策略下调度兜底自动 ACK 任务（FR-24）。 7. 返回携带 ``ack_decision`` / ``agent_run_id`` / ``is_silent`` / ``error`` / ``command_response`` 的 ``InboundResult``。 + 注：出站异步化后，``ack_decision`` 仅反映入站管道 ACK 状态 + （AFTER_RECORD / AFTER_AGENT_DISPATCH 为 ack，其他为 pending）， + ``error`` 始终为 ``None``（出站错误由后台任务处理）。 8. 异常处理（HTTP 语义对齐）：``ChannelError`` 子类与翻译后的 ``OperationTimeoutError`` 向上抛由 ``unified_error_handler`` 统一 映射；真正未预期的非 ``ChannelError`` 异常返回 ``nack`` @@ -340,21 +362,29 @@ class InboundMessageService: exc_info=e, ) - # 桥接到出站管道：创建 Agent 运行后投递 Agent 响应至渠道侧（FR-13） + # 桥接到出站管道：异步触发，不阻塞入站接收层（FR-13 / FR-15）。 + # 出站投递在后台协程执行，同会话保序由 outbound:{conversation_id} 锁 + # 保证，同账号并发出站数由信号量限制。AFTER_PERSIST 策略下 ACK 由 + # 后台任务完成后设置，此处 ack_decision 保持 pending。 if ctx.agent_run_id and not ctx.is_silent: - await self._deliverAgentResponse(ctx) + self._scheduleOutboundDelivery(ctx, silent=False) elif ctx.is_silent and ctx.command is not None and ctx.command.response is not None: - # 静默命令响应投递：不创建 AgentRun，通过出站管道投递命令响应（FR-15） - await self._deliverSilentCommandResponse(ctx) + self._scheduleOutboundDelivery(ctx, silent=True) # MANUAL 策略兜底：插件忘记 ACK 时，调度延迟自动 ACK（FR-24） self._scheduleManualFallback(ctx) + # 出站异步化后，InboundResult 仅反映入站管道 ACK 状态： + # - AFTER_RECORD / AFTER_AGENT_DISPATCH：ReplyStage 已设置 ctx.acked=True + # 和 ctx.ack_decision="ack"，直接透传。 + # - AFTER_PERSIST / MANUAL / None：ACK 由后台任务完成后设置， + # 此处保持 pending。出站错误由后台任务通过 _handleOutboundFailure + # 处理（幂等记录回退），不再通过 InboundResult 透传。 return InboundResult( ack_decision=ctx.ack_decision, agent_run_id=ctx.agent_run_id, is_silent=ctx.is_silent, - error=ctx.outbound_error, + error=None, command_response=(ctx.command.response if ctx.command is not None else None), ) @@ -867,6 +897,70 @@ class InboundMessageService: error=str(e), ) + def _scheduleOutboundDelivery(self, ctx: InboundContext, *, silent: bool) -> None: + """调度出站投递后台任务（非阻塞，FR-13 / FR-15）。 + + 入站管道完成（幂等记录 completed + ACK 已决策）后立即触发出站投递， + 不阻塞 ``receiveInbound`` 返回，释放 worker 循环接收下一条消息。 + 任务在信号量控制下执行出站管道，同账号并发出站数受信号量限制。 + + 任务生命周期：加入 ``_outbound_delivery_tasks`` 集合持有引用避免 GC， + 完成时通过 ``add_done_callback`` 自动移除，``shutdown`` 时统一取消。 + + InboundContext 安全性：每条消息有独立 ctx 实例，``receiveInbound`` + 返回后无其他协程访问该 ctx；``_deliverAgentResponse`` / + ``_deliverSilentCommandResponse`` 仅读取不可变字段（trace_id、 + agent_run_id、channel_session、conversation_id 等），不访问 + ``ctx.tx``（事务已在入站管道结束时提交）。 + + 参数： + ctx: 入站管道上下文，携带 Agent 运行 ID、渠道会话等字段。 + silent: ``True`` 投递静默命令响应（``_deliverSilentCommandResponse``）， + ``False`` 投递 Agent 响应（``_deliverAgentResponse``）。 + """ + task = asyncio.create_task(self._deliverOutboundAsync(ctx, silent=silent)) + self._outbound_delivery_tasks.add(task) + task.add_done_callback(self._outbound_delivery_tasks.discard) + + async def _deliverOutboundAsync(self, ctx: InboundContext, *, silent: bool) -> None: + """出站投递后台任务，带信号量背压控制。 + + 信号量按账号隔离：``account_key = f"{channel_type}:{account_id}"``， + 不同账号互不影响。信号量耗尽时任务排队等待，自然形成背压，防止 + 同账号无限制 ``create_task`` 导致内存溢出。 + + ``_deliverAgentResponse`` / ``_deliverSilentCommandResponse`` 内部已有 + 完整 try/except + ``_handleOutboundFailure``，此处外层兜底防止异常 + 逃逸导致 ``create_task`` 静默失败。 + + 参数： + ctx: 入站管道上下文。 + silent: ``True`` 投递静默命令响应，``False`` 投递 Agent 响应。 + """ + account_key = f"{ctx.channel_type}:{ctx.account_id}" + sem = self._outbound_semaphores.get(account_key) + if sem is None: + sem = asyncio.Semaphore(_OUTBOUND_CONCURRENCY_PER_ACCOUNT) + self._outbound_semaphores[account_key] = sem + async with sem: + try: + if silent: + await self._deliverSilentCommandResponse(ctx) + else: + await self._deliverAgentResponse(ctx) + except Exception as e: + # _deliverAgentResponse 内部已有完整 try/except + + # _handleOutboundFailure，此处兜底防止异常逃逸导致 + # create_task 静默失败 + await self._logger.exception( + "outbound delivery task crashed", + trace_id=ctx.trace_id, + agent_run_id=ctx.agent_run_id, + channel_type=str(ctx.channel_type), + account_id=ctx.account_id, + exc_info=e, + ) + def _scheduleManualFallback(self, ctx: InboundContext) -> None: """调度 MANUAL 策略的兜底自动 ACK（FR-24）。 diff --git a/backend/package/yuxi/channels/core/service/credential_service.py b/backend/package/yuxi/channels/core/service/credential_service.py index 197346de..ab2f204e 100644 --- a/backend/package/yuxi/channels/core/service/credential_service.py +++ b/backend/package/yuxi/channels/core/service/credential_service.py @@ -48,10 +48,12 @@ __all__ = ["CredentialService"] #: ``fence:`` 约定（前缀 + ID）。 _CREDENTIAL_CACHE_KEY_PREFIX: str = "credentials:" -#: 凭证缓存 TTL（秒），避免明文凭证长期驻留缓存（H2-4）。 -#: 5 分钟窗口平衡缓存命中率与凭证轮换 / 撤销后的失效时延：超时后下次 -#: ``getCredentials`` 重新从 ConfigPort 解密回填，感知最新的撤销状态。 -_CREDENTIAL_CACHE_TTL: int = 300 +#: 凭证缓存 TTL：``None`` 表示不过期。 +#: 凭证缓存的失效由显式清理保证（``onAccountDisabled`` / +#: ``beforeAccountDelete`` 删除键，``_persistCredentials`` 覆写）， +#: 不依赖 TTL 感知撤销状态。设 TTL 会导致运行时过期后插件客户端 +#: （如 ILinkClient）只读 CachePort 不回填，触发无限退避重试。 +_CREDENTIAL_CACHE_TTL: int | None = None #: Task 3 新增的 ACCOUNT 作用域配置键名。 _CREDENTIALS_KEY: str = "credentials" diff --git a/backend/package/yuxi/channels/infrastructure/factory.py b/backend/package/yuxi/channels/infrastructure/factory.py index a390d9bd..3e97ca4b 100644 --- a/backend/package/yuxi/channels/infrastructure/factory.py +++ b/backend/package/yuxi/channels/infrastructure/factory.py @@ -851,6 +851,7 @@ async def create_host_bootstrap( circuit_breaker=channel_circuit_breaker, logger=logger, message_deliverer=_deliver_inbound_message, + cache_port=core.cache_port, ) # 8. 构造 HealthAggregator（注入 transport_manager 作为 transport_health_port） diff --git a/backend/package/yuxi/channels/plugins/README.md b/backend/package/yuxi/channels/plugins/README.md index 19e18336..ad79484a 100644 --- a/backend/package/yuxi/channels/plugins/README.md +++ b/backend/package/yuxi/channels/plugins/README.md @@ -426,7 +426,7 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest host.registerAdapter("lifecycle", WeChatWocLifecycleAdapter(cache_port, logger_port)) host.registerAdapter("probeable", WeChatWocProbeableAdapter(client, persistence_port, logger_port)) host.registerAdapter("doctor", WeChatWocDoctorAdapter(client, logger_port)) - host.registerAdapter("wizard", WeChatWocWizardAdapter(client, logger_port)) + host.registerAdapter("wizard", WeChatWocWizardAdapter(client, config_port, logger_port, manifest.channel_type)) host.registerAdapter("whitelist", WeChatWocWhitelistAdapter(cache_port, logger_port)) host.registerAdapter("directory", WeChatWocDirectoryAdapter(client, logger_port)) diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/outbound_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/outbound_adapter.py index 33d8c947..ae4e7fe0 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/outbound_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/outbound_adapter.py @@ -34,12 +34,12 @@ from uuid import uuid4 from yuxi.channels.contract.dtos.common import MessageContent, MessageFormat from yuxi.channels.contract.dtos.config import ConfigScope +from yuxi.channels.contract.dtos.option import Some from yuxi.channels.contract.dtos.outbound import ( BatchSendResult, FormattedMessage, ) from yuxi.channels.contract.dtos.outbox import MultiPartReceipt -from yuxi.channels.contract.dtos.option import Some from yuxi.channels.contract.errors import ( DependencyError, NotFoundError, diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py index 5aa61f6b..444305f6 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/stream_connector_adapter.py @@ -260,8 +260,10 @@ class WeChatWocStreamConnectorAdapter: # 消息），首次连接（persisted_cursor 为空）回退到 sync # 事件 cursor。cursor 可能为 0（从头补全），仅 # None/空字符串跳过。 - backfill_cursor = persisted_cursor if persisted_cursor else ( - str(cursor) if cursor is not None and cursor != "" else None + backfill_cursor = ( + persisted_cursor + if persisted_cursor + else (str(cursor) if cursor is not None and cursor != "" else None) ) if backfill_cursor is not None: await self._backfill(account_id, backfill_cursor, buffer, handle.trace_id) diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py index 4881262e..d7e05133 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/adapters/wizard_adapter.py @@ -39,6 +39,8 @@ from __future__ import annotations import asyncio from typing import Any +from yuxi.channels.contract.dtos.channel import ChannelType +from yuxi.channels.contract.dtos.config import ConfigScope from yuxi.channels.contract.dtos.wizard import ( WizardConfigPatch, WizardField, @@ -47,16 +49,19 @@ from yuxi.channels.contract.dtos.wizard import ( WizardStepResult, ) from yuxi.channels.contract.errors import ( + ConfigValidationError, DependencyError, + NotFoundError, NotImplementedError, OperationTimeoutError, RateLimitError, ValidationError, ) +from yuxi.channels.contract.ports.driven.config_port import ConfigPort from yuxi.channels.contract.ports.driven.logger_port import LoggerPort -from ..woc_bridge_client import WocBridgeClient from .._url_validators import _redact_url, _validate_bridge_url_host, _validate_bridge_url_scheme +from ..woc_bridge_client import WocBridgeClient # account_info 步骤 display_name 字段的占位提示文案 _DISPLAY_NAME_PLACEHOLDER = "微信 WechatOnCloud 账号" @@ -69,18 +74,22 @@ _BRIDGE_TOKEN_PLACEHOLDER = "WOC_BRIDGE_API_TOKEN" class WeChatWocWizardAdapter: """微信 wechat_woc 安装向导适配器。 - 通过 DI 接收 ``WocBridgeClient`` / ``LoggerPort``，不访问全局 settings - 或 logger。适配器仅执行步骤校验与配置补丁产出，不含业务规则（向导状态机 - 由 ``WizardService`` 承载）。 + 通过 DI 接收 ``WocBridgeClient`` / ``ConfigPort`` / ``LoggerPort``，不访问 + 全局 settings 或 logger。适配器仅执行步骤校验与配置补丁产出，不含业务规则 + （向导状态机由 ``WizardService`` 承载）。 """ def __init__( self, client: WocBridgeClient, + config_port: ConfigPort, logger_port: LoggerPort, + channel_type: ChannelType, ) -> None: self._client = client + self._config_port = config_port self._logger = logger_port + self._channel_type = channel_type # ------------------------------------------------------------------ # WizardAdapter Protocol 实现 @@ -174,9 +183,9 @@ class WeChatWocWizardAdapter: - ``account_info``：校验 ``display_name`` 为非空字符串，产出含 ``display_name`` 的配置补丁。 - ``bridge_config``：校验 ``bridge_url`` 非空且使用 ``https://`` - （开发环境 localhost 豁免需预先配置 ``allow_insecure_localhost``， - wizard 阶段无 ConfigPort 访问，默认强制 https），追加 host 校验 - 拦截内网 IP 段（SSRF 防护），校验 ``bridge_token`` 非空，产出含 + （开发环境 localhost 豁免需预先在 CHANNEL 作用域配置 + ``allow_insecure_localhost=true``，行为与 lifecycle 一致），追加 host + 校验拦截内网 IP 段（SSRF 防护），校验 ``bridge_token`` 非空，产出含 ``bridge_url`` 与 ``bridge_token`` 的配置补丁。 - ``verify``：从 ``applied_config`` 取 ``bridge_url`` 与 ``bridge_token`` 调 ``WocBridgeClient.get_status_with_url`` 探活（wizard 阶段账户未创建， @@ -220,9 +229,10 @@ class WeChatWocWizardAdapter: errors=("bridge_url_required",), ) url = bridge_url.strip() + allow_insecure = await self._allow_insecure_localhost() try: - _validate_bridge_url_scheme(url, False) - _validate_bridge_url_host(url, False) + _validate_bridge_url_scheme(url, allow_insecure) + _validate_bridge_url_host(url, allow_insecure) except ValidationError as exc: return WizardStepResult( step_id="bridge_config", @@ -390,8 +400,9 @@ class WeChatWocWizardAdapter: message="bridge_url must be a non-empty string", ) url = bridge_url.strip() - _validate_bridge_url_scheme(url, False) - _validate_bridge_url_host(url, False) + allow_insecure = await self._allow_insecure_localhost() + _validate_bridge_url_scheme(url, allow_insecure) + _validate_bridge_url_host(url, allow_insecure) bridge_token = values.get("bridge_token") if not isinstance(bridge_token, str) or not bridge_token.strip(): raise ValidationError( @@ -445,6 +456,28 @@ class WeChatWocWizardAdapter: """ raise NotImplementedError(operation="buildOAuthAuthorizeUrl") + async def _allow_insecure_localhost(self) -> bool: + """读取 channel 作用域的 ``allow_insecure_localhost`` 配置。 + + 与 lifecycle / lifecycle_adapter 保持一致：仅在开发环境显式开启时 + 允许 ``http://localhost`` / ``127.0.0.1`` / ``::1`` 的 bridge_url。 + 读取失败时记录告警并默认关闭，避免配置端口异常阻断 wizard 流程。 + """ + try: + value = await self._config_port.get( + "allow_insecure_localhost", + scope=ConfigScope.CHANNEL, + target=self._channel_type, + ) + except (ConfigValidationError, NotFoundError) as exc: + await self._logger.warn( + "wechat_woc wizard: read allow_insecure_localhost failed, default to false", + channel_type=self._channel_type, + error=str(exc), + ) + return False + return bool(value.value) + async def _fire_init_db_safely( self, *, diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/entry.py b/backend/package/yuxi/channels/plugins/wechat_woc/entry.py index 8b5aa920..30d7c9f7 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/entry.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/entry.py @@ -28,8 +28,11 @@ client 实例化、LifecycleHandler 构造与 ``registerAdapter`` 调用。 ``sse_heartbeat_interval_ms`` / ``sse_connect_timeout_ms`` / ``max_batch_size``。 - ``outbound`` 接收 ``(client, config_port, logger_port)``：需从 ConfigPort CHANNEL 作用域读取 ``max_message_length``（P2-2）。 -- ``inbound`` / ``doctor`` / ``wizard`` / ``directory`` 接收 - ``(client, logger_port)``：仅做协议转换。 +- ``wizard`` 接收 ``(client, config_port, logger_port, channel_type)``： + 需从 ConfigPort CHANNEL 作用域读取 ``allow_insecure_localhost`` 以兼容 + 开发环境 localhost http bridge_url。 +- ``inbound`` / ``doctor`` / ``directory`` 接收 ``(client, logger_port)``： + 仅做协议转换。 - ``login`` 接收 ``(client, logger_port)``：扫码状态由 ``QrLoginService`` 承载，适配器不缓存登录态。 - ``lifecycle`` 接收 ``(cache_port, logger_port)``：账户删除/禁用时清理 @@ -165,7 +168,7 @@ def channel_entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest host.registerAdapter("lifecycle", WeChatWocLifecycleAdapter(cache_port, logger_port)) host.registerAdapter("probeable", WeChatWocProbeableAdapter(client, persistence_port, logger_port)) host.registerAdapter("doctor", WeChatWocDoctorAdapter(client, logger_port)) - host.registerAdapter("wizard", WeChatWocWizardAdapter(client, logger_port)) + host.registerAdapter("wizard", WeChatWocWizardAdapter(client, config_port, logger_port, manifest.channel_type)) host.registerAdapter("whitelist", WeChatWocWhitelistAdapter(cache_port, logger_port)) host.registerAdapter("directory", WeChatWocDirectoryAdapter(client, logger_port)) diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py b/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py index 6ecf49c0..3082a5fd 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py @@ -57,6 +57,49 @@ _USER_AGENT_TEMPLATE = "yuxi-channels-wechat-woc/{version}" # 放大延迟（最坏 15 次 HTTP 请求 + 35s+ 等待）。 _HTTP_NETWORK_ERROR_MAX_ATTEMPTS = 2 + +class _ReentrantAsyncLock: + """可重入的 asyncio 锁。 + + 同一协程（asyncio.Task）可多次 acquire，仅在首次实际获取底层 + ``asyncio.Lock``，后续重入仅增加计数。release 时递减计数，计数归零 + 才释放底层锁。不同协程之间仍保持互斥。 + + 用途：wechat_woc 要求同账号 send_text/image/file 串行调用，但一次 + 出站管道可能在同一协程内连续调用多个 send_*（长文本分片、多媒体、 + 流式续发）。非可重入锁会导致同协程死锁，因此需要可重入语义。 + """ + + def __init__(self) -> None: + self._lock = asyncio.Lock() + self._owner: asyncio.Task | None = None + self._count = 0 + + async def acquire(self) -> None: + task = asyncio.current_task() + if task is self._owner: + self._count += 1 + return + await self._lock.acquire() + self._owner = task + self._count = 1 + + def release(self) -> None: + task = asyncio.current_task() + if task is not self._owner: + raise RuntimeError("release unlocked lock") + self._count -= 1 + if self._count == 0: + self._owner = None + self._lock.release() + + async def __aenter__(self) -> None: + await self.acquire() + + async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None: + self.release() + + # bridge 能力缓存 TTL（秒），与 CapabilityProof.cache_ttl 默认值一致 _CAPABILITIES_CACHE_TTL_SECONDS = 300 # 能力协商失败时保守默认值的短 TTL（秒）：避免 bridge 持续不可用时每次出站 @@ -265,6 +308,9 @@ class WocBridgeClient: # onUnload 时 close_all_sse 先于 detach_http_client 关闭所有 SSE 长连接， # 避免独立 httpx client（不复用共享连接池）成为孤儿连接（P1-6）。 self._active_sse_handles: dict[str, SseStreamHandle] = {} + # 按账号隔离的发送锁，保证同账号 send_text/image/file 逐一调用 bridge。 + # 使用可重入锁，避免同一出站协程内多次调用 send_* 时死锁。 + self._send_locks: dict[str, _ReentrantAsyncLock] = {} # ------------------------------------------------------------------ # 连接池管理 @@ -348,6 +394,22 @@ class WocBridgeClient: ) return False + def _get_send_lock(self, account_id: str) -> _ReentrantAsyncLock: + """获取指定账号的发送锁，保证同账号出站请求逐一调用 bridge。 + + 按 ``account_id`` 隔离，不同账号互不阻塞；同一账号内 ``send_text`` / + ``send_image`` / ``send_file`` 串行执行。锁为可重入锁，同一协程内 + 连续调用多个 send_* 不会死锁。 + + @consistency + - 幂等：重复获取返回同一锁对象 + """ + lock = self._send_locks.get(account_id) + if lock is None: + lock = _ReentrantAsyncLock() + self._send_locks[account_id] = lock + return lock + # ------------------------------------------------------------------ # 凭证读取 # ------------------------------------------------------------------ @@ -1232,12 +1294,13 @@ class WocBridgeClient: to_wxid=to_wxid, content_length=len(content), ) - resp = await self._execute_http( - "POST", - "/api/send/text", - account_id=account_id, - json_body={"to_wxid": to_wxid, "content": content}, - ) + async with self._get_send_lock(account_id): + resp = await self._execute_http( + "POST", + "/api/send/text", + account_id=account_id, + json_body={"to_wxid": to_wxid, "content": content}, + ) result = self._parse_json_response(resp) await self._logger.debug( "woc bridge send_text response", @@ -1275,12 +1338,13 @@ class WocBridgeClient: to_wxid=to_wxid, file_path=file_path, ) - resp = await self._execute_http( - "POST", - "/api/send/image", - account_id=account_id, - json_body={"to_wxid": to_wxid, "file_path": file_path}, - ) + async with self._get_send_lock(account_id): + resp = await self._execute_http( + "POST", + "/api/send/image", + account_id=account_id, + json_body={"to_wxid": to_wxid, "file_path": file_path}, + ) result = self._parse_json_response(resp) await self._logger.debug( "woc bridge send_image response", @@ -1318,12 +1382,13 @@ class WocBridgeClient: to_wxid=to_wxid, file_path=file_path, ) - resp = await self._execute_http( - "POST", - "/api/send/file", - account_id=account_id, - json_body={"to_wxid": to_wxid, "file_path": file_path}, - ) + async with self._get_send_lock(account_id): + resp = await self._execute_http( + "POST", + "/api/send/file", + account_id=account_id, + json_body={"to_wxid": to_wxid, "file_path": file_path}, + ) result = self._parse_json_response(resp) await self._logger.debug( "woc bridge send_file response",