refactor(wechat-woc,transport): 批量代码优化与功能增强

1.  修复代码格式与缩进问题，统一代码风格
2.  调整凭证缓存TTL为不过期，显式管理缓存失效
3.  重构wechat-woc出站请求，添加可重入锁保证同账号串行调用
4.  新增wechat-woc向导适配器配置端口支持，兼容开发环境localhost
5.  实现入站消息出站投递异步化，添加账号级并发限制
6.  新增transport启动前凭证缓存预热逻辑
7.  调整导入顺序与依赖位置，优化代码结构

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

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(

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

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(

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

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

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":

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,

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] = {}

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

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）。
 

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"

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）

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

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,

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)

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,
         *,

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

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",