test: 修复并完善channel单元测试

1.  删除大量冗余空的__init__.py测试文件
2.  调整测试代码格式与导入，统一代码风格
3.  补充缺失的测试用例与断言逻辑
4.  修正参数类型不匹配与方法定义错误
5.  更新断言与返回值处理逻辑

backend/package/yuxi/channel/application/pipeline/middlewares/auth_middleware.py

@@ -14,6 +14,9 @@ class AuthMiddleware:
         return "auth"
 
     async def process(self, ctx: MessageContext, call_next: CallNext) -> MessageContext:
+        if ctx.message.metadata.get("auth_method"):
+            return await call_next(ctx)
+
         auth_header = ctx.message.raw_payload.get("authorization", "")
         client_ip = ctx.message.metadata.get("client_ip", "unknown")
 
@@ -24,4 +27,5 @@ class AuthMiddleware:
             ctx.abort(reason, code)
             return ctx
 
+        ctx.message.metadata["auth_method"] = "pipeline_bearer"
         return await call_next(ctx)

backend/package/yuxi/channel/application/pipeline/middlewares/dedup_middleware.py

@@ -25,8 +25,14 @@ class DedupMiddleware:
         idem_key = ctx.message.metadata.get("idempotency_key")
         if idem_key is None:
             idem_key = ctx.message.raw_payload.get("idempotency_key")
-        dedup_key = f"channel:dedup:{idem_key}" if idem_key else f"channel:dedup:{ctx.message.message_id}"
-        ttl = self._idempotency_ttl if idem_key else 900
+        if idem_key:
+            dedup_key = f"channel:dedup:{idem_key}"
+            ttl = self._idempotency_ttl
+        elif ctx.message.message_id:
+            dedup_key = f"channel:dedup:{ctx.message.message_id}"
+            ttl = 900
+        else:
+            return await call_next(ctx)
 
         added = await self._cache.set(dedup_key, "1", nx=True, ex=ttl)
         if not added:

backend/package/yuxi/channel/application/pipeline/middlewares/validation_middleware.py

@@ -34,8 +34,9 @@ class ValidationMiddleware:
             ctx.abort("missing message_id", "VALIDATION_ERROR")
             return ctx
         if not msg.content or not msg.content.strip():
-            ctx.abort("empty content", "VALIDATION_ERROR")
-            return ctx
+            if not msg.attachments:
+                ctx.abort("empty content", "VALIDATION_ERROR")
+                return ctx
 
         body_size = len(msg.content.encode()) + len(str(msg.raw_payload).encode())
         if body_size > self._max_body_bytes:

backend/package/yuxi/channel/application/service/auth_service.py

@@ -15,12 +15,14 @@ class AuthService:
         password: str | None = None,
         max_attempts: int = 5,
         lockout_seconds: int = 300,
+        allow_anonymous: bool = False,
     ) -> None:
         self._rate_limiter = rate_limit_port
         self._token = token
         self._password = password
         self._max_attempts = max_attempts
         self._lockout_seconds = lockout_seconds
+        self._allow_anonymous = allow_anonymous
 
     def update_credentials(self, *, token: str | None = None, password: str | None = None) -> None:
         if token is not None:
@@ -30,7 +32,9 @@ class AuthService:
 
     async def authenticate(self, auth_header: str, *, client_id: str = "unknown") -> tuple[bool, str]:
         if not self._token and not self._password:
-            return True, ""
+            if self._allow_anonymous:
+                return True, ""
+            return False, "no credentials configured"
 
         locked, remaining = await self._rate_limiter.is_locked(f"channel:auth:lockout:{client_id}")
         if locked:

backend/package/yuxi/channel/application/service/binding_service.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from yuxi.channel.domain.exception.duplicate_binding import DuplicateBindingException
 from yuxi.channel.domain.model.binding.channel_binding import ChannelBinding
 from yuxi.channel.domain.repository.binding_repository import BindingRepositoryPort
 
@@ -15,16 +16,30 @@ class BindingService:
         account_id: str,
         group_id: str,
         agent_config_id: int,
+        created_by: str | None = None,
     ) -> ChannelBinding:
+        existing = await self._repo.find_active_binding(channel_type=channel_type, account_id=account_id, group_id=group_id)
+        if existing:
+            raise DuplicateBindingException(channel_type, account_id, group_id)
         return await self._repo.create_binding(
             channel_type=channel_type,
             account_id=account_id,
             group_id=group_id,
             agent_config_id=agent_config_id,
+            created_by=created_by,
         )
 
-    async def delete(self, binding_id: int) -> bool:
-        return await self._repo.delete_binding(binding_id)
+    async def update(
+        self,
+        binding_id: int,
+        *,
+        agent_config_id: int | None = None,
+        is_enabled: bool | None = None,
+    ) -> ChannelBinding | None:
+        return await self._repo.update_binding(binding_id, agent_config_id=agent_config_id, is_enabled=is_enabled)
+
+    async def delete(self, binding_id: int, *, updated_by: str | None = None) -> ChannelBinding | None:
+        return await self._repo.delete_binding(binding_id, updated_by=updated_by)
 
     async def get(self, binding_id: int) -> ChannelBinding | None:
         return await self._repo.get_binding(binding_id)

backend/package/yuxi/channel/application/service/config_service.py

@@ -1,12 +1,18 @@
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import json
+import logging
 
+from yuxi.channel.application.service.auth_service import AuthService
 from yuxi.channel.domain.middleware.configurable import Configurable
+from yuxi.channel.domain.port.external.channel_request_verifier_port import ChannelRequestVerifierPort
 from yuxi.channel.domain.port.config_reload_port import ConfigReloadPort
 from yuxi.channel.domain.service.pipeline import Pipeline
-from yuxi.channel.infrastructure.config.channel_config import ChannelConfig
+from yuxi.channel.infrastructure.configuration.channel_config import ChannelConfig
+
+logger = logging.getLogger(__name__)
 
 
 class ConfigService:
@@ -17,29 +23,43 @@ class ConfigService:
         pipeline: Pipeline,
         *,
         channel_config: ChannelConfig | None = None,
+        auth_service: AuthService | None = None,
+        verifiers: dict[str, ChannelRequestVerifierPort] | None = None,
     ) -> None:
         self._config = config_data
         self._config_reload = config_reload
         self._pipeline = pipeline
         self._channel_config = channel_config
+        self._auth_service = auth_service
+        self._verifiers = verifiers
+        self._last_config_hash: str | None = None
+        self._reload_lock = asyncio.Lock()
 
     async def reload(self) -> tuple[list[str], str | None]:
-        reloaded = await self._config_reload.reload()
-        if reloaded:
-            self._config = reloaded
+        async with self._reload_lock:
+            reloaded = await self._config_reload.reload()
+            if reloaded:
+                self._config = reloaded
 
-        updated = self._update_configurable_middlewares()
+            config_hash = hashlib.sha256(
+                json.dumps(self._config, sort_keys=True).encode(),
+            ).hexdigest()[:8]
 
-        if self._channel_config:
-            items = await self._channel_config.on_config_updated(self._config)
-            if items:
-                updated.extend(items)
+            if config_hash == self._last_config_hash:
+                return [], config_hash
 
-        config_hash = hashlib.sha256(
-            json.dumps(self._config, sort_keys=True).encode(),
-        ).hexdigest()[:8]
+            self._last_config_hash = config_hash
 
-        return updated, config_hash
+            updated = self._update_configurable_middlewares()
+
+            if self._channel_config:
+                items = await self._channel_config.on_config_updated(self._config)
+                if items:
+                    updated.extend(items)
+                    self._sync_auth_credentials(items)
+                    self._rebuild_verifiers(items)
+
+            return updated, config_hash
 
     @property
     def config(self) -> dict:
@@ -55,3 +75,25 @@ class ConfigService:
                     updated.extend(items)
 
         return updated
+
+    def _sync_auth_credentials(self, updated_items: list[str]) -> None:
+        if not self._auth_service:
+            return
+        if "auth_token" in updated_items or "auth_password" in updated_items:
+            self._auth_service.update_credentials(
+                token=self._channel_config.auth_token,
+                password=self._channel_config.auth_password,
+            )
+
+    def _rebuild_verifiers(self, updated_items: list[str]) -> None:
+        if "feishu_encrypt_key" not in updated_items:
+            return
+        if not self._verifiers or not self._channel_config:
+            return
+        from yuxi.channel.channels.feishu.verifier import FeishuRequestVerifier
+
+        feishu_verifier = FeishuRequestVerifier(
+            verification_token=self._channel_config.feishu_verification_token or "",
+            encrypt_key=self._channel_config.feishu_encrypt_key or "",
+        )
+        self._verifiers["feishu"] = feishu_verifier

backend/package/yuxi/channel/application/service/delivery_service.py

@@ -91,8 +91,16 @@ class DeliveryService:
 
         adapter = self._adapters.get(channel_type)
         if not adapter:
+            await self._outbox.enqueue(
+                message_id=message_id,
+                session_id=session_id,
+                channel_type=channel_type,
+                content=full_response,
+                trace_id=trace_id,
+            )
             if self._metrics:
                 await self._metrics.record_worker_dispatch_total(channel_type, "no_adapter")
+                await self._metrics.record_outbox_enqueued(channel_type)
             return DispatchResult(success=False, message_id=message_id, error=f"no adapter for {channel_type}")
 
         sent = await self._try_send(adapter, session_id, full_response, channel_type, trace_id, message_id)

backend/package/yuxi/channel/application/service/inbound_service.py

@@ -63,10 +63,12 @@ class InboundService:
             return
         try:
             pipeline_result = "skipped" if ctx.is_skipped else ("aborted" if ctx.is_aborted else "passed")
+            status = "processing" if pipeline_result == "passed" else "completed"
             await self._message_log_repo.update_pipeline_result(
                 trace_id=ctx.trace_id,
                 message_id=ctx.message.message_id,
                 pipeline_result=pipeline_result,
+                status=status,
                 abort_reason=ctx.abort_reason if ctx.is_aborted else None,
             )
         except Exception:
@@ -75,13 +77,19 @@ class InboundService:
     async def _publish_event(self, ctx: MessageContext) -> None:
         if not self._events:
             return
+        session_id = (
+            ctx.message.metadata.get("group_id")
+            or ctx.message.metadata.get("session_id")
+            or ctx.message.sender.id
+            or ""
+        )
         try:
             if ctx.is_aborted:
                 await self._events.publish(
                     MessageBlocked(
                         message_id=ctx.message.message_id,
                         channel_type=ctx.channel_type,
-                        session_id="",
+                        session_id=session_id,
                         reason=ctx.abort_reason,
                         abort_code=ctx.abort_code,
                         trace_id=ctx.trace_id,
@@ -92,9 +100,9 @@ class InboundService:
                     MessageReceived(
                         message_id=ctx.message.message_id,
                         channel_type=ctx.channel_type,
-                        session_id="",
+                        session_id=session_id,
                         sender_id=ctx.message.sender.id,
-                        content_summary=ctx.message.content[:200],
+                        content_summary=(ctx.message.content or "")[:200],
                         trace_id=ctx.trace_id,
                     )
                 )

backend/package/yuxi/channel/application/service/session_resolver.py

@@ -4,6 +4,7 @@ import logging
 
 from yuxi.channel.domain.model.binding.channel_binding import ChannelBinding
 from yuxi.channel.domain.model.session.channel_session import ChannelSession
+from yuxi.channel.domain.port import AgentPort
 from yuxi.channel.domain.repository.binding_repository import BindingRepositoryPort
 from yuxi.channel.domain.repository.session_repository import SessionRepositoryPort
 
@@ -15,12 +16,14 @@ class SessionResolver:
         self,
         session_repo: SessionRepositoryPort,
         binding_repo: BindingRepositoryPort,
+        agent_port: AgentPort,
         *,
         default_agent_config_id: int = 1,
         session_key_strategy: str = "auto",
     ) -> None:
         self._session_repo = session_repo
         self._binding_repo = binding_repo
+        self._agent_port = agent_port
         self._default_agent_config_id = default_agent_config_id
         self._session_key_strategy = session_key_strategy
 
@@ -33,19 +36,20 @@ class SessionResolver:
         if not agent_config_id:
             agent_config_id = await self._resolve_agent_config(channel_type, metadata)
 
-        binding = await self._binding_repo.find_binding(
+        binding = await self._binding_repo.find_active_binding(
             channel_type=channel_type,
             account_id=metadata.get("account_id", ""),
             group_id=metadata.get("group_id", ""),
         )
 
         session_key = self._resolve_session_key(payload, channel_type, sender_id, binding)
+        agent_id = await self._agent_port.resolve_agent_id(agent_config_id)
 
         try:
             session = await self._session_repo.get_or_create(
                 channel_type=channel_type,
                 account_id=session_key,
-                agent_config_id=agent_config_id,
+                agent_id=agent_id,
             )
             return session
         except Exception:
@@ -57,7 +61,7 @@ class SessionResolver:
             return None
 
     async def _resolve_agent_config(self, channel_type: str, metadata: dict) -> int:
-        binding = await self._binding_repo.find_binding(
+        binding = await self._binding_repo.find_active_binding(
             channel_type=channel_type,
             account_id=metadata.get("account_id", ""),
             group_id=metadata.get("group_id", ""),

backend/package/yuxi/channel/channels/__init__.py

@@ -1,4 +1,4 @@
 import yuxi.channel.channels.feishu
 import yuxi.channel.channels.dingtalk
 import yuxi.channel.channels.web
-import yuxi.channel.channels.hooks
+import yuxi.channel.channels.hooks  # noqa: F401

backend/package/yuxi/channel/channels/feishu/adapter.py

@@ -9,7 +9,7 @@ from yuxi.channel.domain.model.message.dispatch_result import SendResult
 from yuxi.channel.domain.model.message.unified_message import UnifiedMessage
 from yuxi.channel.domain.model.shared.channel_capabilities import ChannelCapabilities
 from yuxi.channel.domain.model.shared.channel_type import ChannelType
-from yuxi.channel.domain.port.channel_route_contributor import ChannelRouteContributor
+from yuxi.channel.domain.port.channel_route_contributor_port import ChannelRouteContributorPort
 from yuxi.channel.domain.port.ws_connection_port import WsConnectionPort
 
 logger = logging.getLogger(__name__)
@@ -57,7 +57,7 @@ class FeishuAdapter:
         return self._ws
 
     @property
-    def route_contributor(self) -> ChannelRouteContributor | None:
+    def route_contributor(self) -> ChannelRouteContributorPort | None:
         return _FeishuRouteContributor()
 
     @classmethod

backend/package/yuxi/channel/channels/feishu/routes.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import json
 import logging
 from uuid import uuid4
 
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, Body, Depends, HTTPException, Request
 
 from yuxi.channel.channels.feishu.translator import FeishuTranslator
+from yuxi.channel.container import ChannelContainer, get_channel
+from yuxi.channel.domain.event.message_dropped import MessageDropped
 
 logger = logging.getLogger(__name__)
 
@@ -13,64 +16,86 @@ router = APIRouter()
 
 
 @router.post("/channel/feishu/event")
-async def feishu_event(request: Request):
-    body = await request.json()
+async def feishu_event(
+    request: Request,
+    body: bytes = Body(..., max_length=256 * 1024),
+    container: ChannelContainer = Depends(get_channel),
+):
+    try:
+        parsed = json.loads(body)
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=400, detail="invalid JSON")
 
-    if body.get("type") == "url_verification":
-        return {"challenge": body.get("challenge", "")}
+    if parsed.get("type") == "url_verification":
+        challenge = parsed.get("challenge", "")
+        if not challenge:
+            raise HTTPException(status_code=400, detail="missing challenge field")
+        verifier = container.verifiers.get("feishu")
+        token = parsed.get("token", "")
+        if verifier:
+            if verifier.has_verification_token:
+                if not verifier.verify_token(token):
+                    raise HTTPException(status_code=403, detail="verification token mismatch")
+            else:
+                logger.warning("feishu verification_token not configured, skipping token check")
+        return {"challenge": challenge}
 
-    from yuxi.channel.container import get_channel
-
-    container = get_channel(request)
-    if not container:
-        raise HTTPException(status_code=503, detail="channel not initialized")
+    verifier = container.verifiers.get("feishu")
+    verify_result = None
+    if verifier and verifier.enabled:
+        headers = {k.lower(): v for k, v in request.headers.items()}
+        verify_result = await verifier.verify(body, headers)
+        if not verify_result.passed:
+            raise HTTPException(status_code=401, detail=verify_result.reason)
 
     adapter = container.adapters.get("feishu")
     if not adapter:
-        logger.warning("feishu adapter not available, dropping event")
-        return {"code": 0}
+        logger.warning("feishu adapter not available")
+        if container.event_publisher:
+            await container.event_publisher.publish(
+                MessageDropped(message_id="", channel_type="feishu", reason="adapter_not_available")
+            )
+        raise HTTPException(status_code=500, detail="feishu adapter not available")
 
     try:
-        message = FeishuTranslator.translate_event(body)
+        message = FeishuTranslator.translate_event(parsed)
     except Exception:
-        logger.exception("feishu event parse failed, body=%s", body)
-        return {"code": 0}
+        logger.exception("feishu event parse failed")
+        if container.metrics:
+            await container.metrics.record_pipeline_aborted("feishu", "PARSE_FAILED")
+        if container.event_publisher:
+            await container.event_publisher.publish(
+                MessageDropped(message_id="", channel_type="feishu", reason="parse_failed")
+            )
+        raise HTTPException(status_code=500, detail="event parse failed")
 
-    trace_id = body.get("header", {}).get("event_id", str(uuid4()))
+    trace_id = parsed.get("header", {}).get("event_id", str(uuid4()))
     message.metadata["trace_id"] = trace_id
     message.metadata["idempotency_key"] = trace_id
     message.metadata["client_ip"] = request.client.host if request.client else "unknown"
+    message.metadata.pop("auth_method", None)
+    if verifier and verifier.enabled:
+        message.metadata["auth_method"] = verify_result.method
 
-    result = await container.inbound_service.submit(message, channel_type="feishu", trace_id=trace_id)
+    submit_result = await container.inbound_service.submit(message, channel_type="feishu", trace_id=trace_id)
 
-    if result.is_aborted:
-        logger.warning("feishu message aborted: %s [%s]", result.abort_reason, trace_id)
-        if container.event_publisher:
-            from yuxi.channel.domain.event.message_blocked import MessageBlocked
-
-            await container.event_publisher.publish(
-                MessageBlocked(
-                    message_id=message.message_id,
-                    channel_type="feishu",
-                    session_id="",
-                    reason=result.abort_reason or "pipeline_aborted",
-                    trace_id=trace_id,
-                )
-            )
+    if submit_result.is_aborted:
+        logger.warning("feishu message aborted: %s [%s]", submit_result.abort_reason, trace_id)
 
     return {"code": 0}
 
 
 @router.get("/channel/feishu/verify")
-async def feishu_verify(challenge: str, token: str, request: Request):
-    from yuxi.channel.container import get_channel
-
-    container = get_channel(request)
-    if not container:
-        raise HTTPException(status_code=503, detail="channel not initialized")
-
-    expected_token = container.channel_config.feishu_verification_token
-    if expected_token and token != expected_token:
-        raise HTTPException(status_code=403, detail="verification token mismatch")
-
+async def feishu_verify(
+    challenge: str,
+    token: str,
+    container: ChannelContainer = Depends(get_channel),
+):
+    verifier = container.verifiers.get("feishu")
+    if verifier:
+        if verifier.has_verification_token:
+            if not verifier.verify_token(token):
+                raise HTTPException(status_code=403, detail="verification token mismatch")
+        else:
+            logger.warning("feishu verification_token not configured, skipping token check")
     return {"challenge": challenge}

backend/package/yuxi/channel/channels/feishu/translator.py

@@ -32,6 +32,7 @@ class FeishuTranslator:
                 "is_group": is_group,
                 "group_id": group_id,
                 "chat_type": chat_type,
+                "account_id": sender_id,
             },
             attachments=attachments,
             raw_payload=raw,

backend/package/yuxi/channel/channels/feishu/verifier.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import logging
+import time
+
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    VerifyResult,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class FeishuRequestVerifier:
+    def __init__(
+        self,
+        verification_token: str = "",
+        encrypt_key: str = "",
+        timestamp_tolerance: int = 300,
+    ):
+        self._verification_token = verification_token
+        self._encrypt_key = encrypt_key
+        self._timestamp_tolerance = timestamp_tolerance
+
+    @property
+    def channel_type(self) -> str:
+        return "feishu"
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self._encrypt_key)
+
+    @property
+    def has_verification_token(self) -> bool:
+        return bool(self._verification_token)
+
+    async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult:
+        if not self._encrypt_key:
+            return VerifyResult(passed=True, method="feishu_none", reason="no encrypt_key configured")
+
+        signature = headers.get("x-lark-signature", "")
+        timestamp = headers.get("x-lark-request-timestamp", "")
+        nonce = headers.get("x-lark-request-nonce", "")
+
+        if not signature or not timestamp:
+            return VerifyResult(passed=False, method="feishu_signature", reason="missing signature headers")
+
+        try:
+            ts = float(timestamp)
+            if abs(time.time() - ts) > self._timestamp_tolerance:
+                return VerifyResult(passed=False, method="feishu_signature", reason="timestamp expired")
+        except (ValueError, TypeError):
+            return VerifyResult(passed=False, method="feishu_signature", reason="invalid timestamp")
+
+        sign_string = f"{timestamp}{nonce}{self._encrypt_key}{body.decode('utf-8', errors='replace')}"
+        expected = hashlib.sha256(sign_string.encode("utf-8")).hexdigest()
+        if not hmac.compare_digest(signature.lower(), expected.lower()):
+            return VerifyResult(passed=False, method="feishu_signature", reason="signature mismatch")
+
+        return VerifyResult(passed=True, method="feishu_signature")
+
+    def verify_token(self, token: str) -> bool:
+        if not self._verification_token:
+            return True
+        return hmac.compare_digest(token, self._verification_token)

backend/package/yuxi/channel/channels/hooks/adapter.py

@@ -8,7 +8,7 @@ from yuxi.channel.domain.model.message.dispatch_result import SendResult
 from yuxi.channel.domain.model.message.unified_message import UnifiedMessage
 from yuxi.channel.domain.model.shared.channel_capabilities import ChannelCapabilities
 from yuxi.channel.domain.model.shared.channel_type import ChannelType
-from yuxi.channel.domain.port.channel_route_contributor import ChannelRouteContributor
+from yuxi.channel.domain.port.channel_route_contributor_port import ChannelRouteContributorPort
 from yuxi.channel.domain.port.ws_connection_port import WsConnectionPort
 
 logger = logging.getLogger(__name__)
@@ -55,7 +55,7 @@ class HooksAdapter:
         return None
 
     @property
-    def route_contributor(self) -> ChannelRouteContributor | None:
+    def route_contributor(self) -> ChannelRouteContributorPort | None:
         return _HooksRouteContributor()
 
     @classmethod

backend/package/yuxi/channel/channels/hooks/routes.py

@@ -1,16 +1,15 @@
 from __future__ import annotations
 
-import hashlib
-import hmac
 import json
 import logging
 from uuid import uuid4
 
-from fastapi import APIRouter, Body, HTTPException, Request
+from fastapi import APIRouter, Body, Depends, HTTPException, Request
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
 from yuxi.channel.channels.hooks.translator import HooksTranslator
+from yuxi.channel.container import ChannelContainer, get_channel
 from yuxi.channel.domain.exception.abort_mapping import ABORT_CODE_TO_HTTP
 
 logger = logging.getLogger(__name__)
@@ -31,41 +30,13 @@ class AcceptedResponse(BaseModel):
     message_id: str | None = None
 
 
-def _verify_hook_auth(mapping, request: Request, body: bytes) -> None:
-    if not mapping.secret:
-        return
-
-    auth_header = request.headers.get("authorization", "")
-    if auth_header.startswith("Bearer "):
-        token = auth_header[7:]
-        if hmac.compare_digest(token, mapping.secret):
-            return
-        raise HTTPException(status_code=401, detail="invalid bearer token")
-
-    signature = request.headers.get("x-signature-256") or request.headers.get("x-hub-signature-256")
-    if signature:
-        if signature.startswith("sha256="):
-            signature = signature[7:]
-        expected = hmac.new(mapping.secret.encode(), body, hashlib.sha256).hexdigest()
-        if hmac.compare_digest(signature, expected):
-            return
-        raise HTTPException(status_code=401, detail="invalid signature")
-
-    raise HTTPException(status_code=401, detail="authentication required")
-
-
 @router.post("/channel/hooks/{match_path:path}", response_model=AcceptedResponse)
 async def receive_hook(
     match_path: str,
     request: Request,
-    body: bytes = Body(..., max_length=256 * 1024),
+    body: bytes = Body(..., max_length=1024 * 1024),
+    container: ChannelContainer = Depends(get_channel),
 ):
-    from yuxi.channel.container import get_channel
-
-    container = get_channel(request)
-    if not container:
-        raise HTTPException(status_code=503, detail="channel not initialized")
-
     adapter = container.adapters.get("hooks")
     if not adapter:
         raise HTTPException(status_code=503, detail="hooks adapter not available")
@@ -74,30 +45,42 @@ async def receive_hook(
     if not mapping:
         raise HTTPException(status_code=404, detail=f"no hook mapping for path: {match_path}")
 
+    if len(body) > mapping.max_body_bytes:
+        raise HTTPException(status_code=413, detail="payload too large")
+
+    verifier = container.verifiers.get("hooks")
+    verify_result = None
+    if verifier:
+        headers = {k.lower(): v for k, v in request.headers.items()}
+        verify_result = await verifier.verify(body, headers, secret=mapping.secret)
+        if not verify_result.passed:
+            raise HTTPException(status_code=401, detail=verify_result.reason)
+    elif mapping.secret is None:
+        raise HTTPException(status_code=401, detail="hook secret not configured")
+    elif not mapping.secret:
+        logger.warning("hook %s has no secret configured, authentication skipped", mapping.match_path)
+
     try:
         if container.metrics:
             await container.metrics.record_hooks_received(match_path)
     except Exception:
         logger.debug("failed to record hooks_received metric")
 
-    _verify_hook_auth(mapping, request, body)
-
-    if len(body) > mapping.max_body_bytes:
-        raise HTTPException(status_code=413, detail="payload too large")
-
     try:
         raw = json.loads(body)
     except json.JSONDecodeError:
         raise HTTPException(status_code=400, detail="invalid JSON")
 
     trace_id = raw.get("trace_id", request.headers.get("x-trace-id", str(uuid4())))
-    idempotency_key = request.headers.get("idempotency-key")
+    idempotency_key = request.headers.get("idempotency-key") or trace_id
 
     message = HooksTranslator.translate(raw, mapping)
     message.metadata["trace_id"] = trace_id
-    if idempotency_key:
-        message.metadata["idempotency_key"] = idempotency_key
+    message.metadata["idempotency_key"] = idempotency_key
     message.metadata["client_ip"] = request.client.host if request.client else "unknown"
+    message.metadata.pop("auth_method", None)
+    if verify_result:
+        message.metadata["auth_method"] = verify_result.method
 
     result = await container.inbound_service.submit(message, channel_type="hooks", trace_id=trace_id)
 
@@ -116,18 +99,10 @@ async def receive_hook(
     if result.is_skipped:
         return JSONResponse(
             status_code=202,
-            content=AcceptedResponse(
-                trace_id=trace_id,
-                status="skipped",
-                message_id=message.message_id,
-            ).model_dump(),
+            content=AcceptedResponse(trace_id=trace_id, status="skipped", message_id=message.message_id).model_dump(),
         )
 
     return JSONResponse(
         status_code=202,
-        content=AcceptedResponse(
-            trace_id=trace_id,
-            status="accepted",
-            message_id=message.message_id,
-        ).model_dump(),
+        content=AcceptedResponse(trace_id=trace_id, status="accepted", message_id=message.message_id).model_dump(),
     )

backend/package/yuxi/channel/channels/hooks/translator.py

@@ -36,6 +36,7 @@ class HooksTranslator:
                 "session_key": session_key,
                 "session_key_prefix": mapping.match_path,
                 "deliver": mapping.deliver,
+                "account_id": raw.get("account_id", mapping.match_path),
             },
             agent_config_id=agent_id,
             raw_payload=raw,

backend/package/yuxi/channel/channels/hooks/verifier.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    VerifyResult,
+)
+
+
+class HooksRequestVerifier:
+    @property
+    def channel_type(self) -> str:
+        return "hooks"
+
+    @property
+    def enabled(self) -> bool:
+        return True
+
+    async def verify(self, body: bytes, headers: dict[str, str], *, secret: str | None = None) -> VerifyResult:
+        if secret is None:
+            return VerifyResult(passed=False, method="hooks_none", reason="secret not configured, request rejected")
+
+        if not secret:
+            return VerifyResult(passed=True, method="hooks_none", reason="no secret configured")
+
+        auth_header = headers.get("authorization", "")
+        if auth_header.startswith("Bearer "):
+            token = auth_header[7:]
+            if hmac.compare_digest(token, secret):
+                return VerifyResult(passed=True, method="hooks_bearer")
+            return VerifyResult(passed=False, method="hooks_bearer", reason="invalid bearer token")
+
+        signature = headers.get("x-signature-256") or headers.get("x-hub-signature-256")
+        if signature:
+            sig = signature[7:] if signature.startswith("sha256=") else signature
+            expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+            if hmac.compare_digest(sig, expected):
+                return VerifyResult(passed=True, method="hooks_hmac")
+            return VerifyResult(passed=False, method="hooks_hmac", reason="invalid signature")
+
+        return VerifyResult(passed=False, method="hooks_none", reason="authentication required")

backend/package/yuxi/channel/channels/web/adapter.py

@@ -8,7 +8,7 @@ from yuxi.channel.domain.model.message.dispatch_result import SendResult
 from yuxi.channel.domain.model.message.unified_message import UnifiedMessage
 from yuxi.channel.domain.model.shared.channel_capabilities import ChannelCapabilities
 from yuxi.channel.domain.model.shared.channel_type import ChannelType
-from yuxi.channel.domain.port.channel_route_contributor import ChannelRouteContributor
+from yuxi.channel.domain.port.channel_route_contributor_port import ChannelRouteContributorPort
 from yuxi.channel.domain.port.sse_push_port import SsePushPort
 from yuxi.channel.domain.port.ws_connection_port import WsConnectionPort
 
@@ -41,7 +41,7 @@ class WebAdapter:
         return None
 
     @property
-    def route_contributor(self) -> ChannelRouteContributor | None:
+    def route_contributor(self) -> ChannelRouteContributorPort | None:
         return _WebRouteContributor()
 
     @classmethod

backend/package/yuxi/channel/channels/web/routes.py

@@ -4,10 +4,11 @@ import json
 import logging
 from uuid import uuid4
 
-from fastapi import APIRouter, Body, HTTPException, Request
+from fastapi import APIRouter, Body, Depends, HTTPException, Request
 from fastapi.responses import JSONResponse
 
 from yuxi.channel.channels.web.translator import WebTranslator
+from yuxi.channel.container import ChannelContainer, get_channel
 from yuxi.channel.domain.exception.abort_mapping import ABORT_CODE_TO_HTTP
 
 logger = logging.getLogger(__name__)
@@ -19,17 +20,20 @@ router = APIRouter()
 async def web_message(
     request: Request,
     body: bytes = Body(..., max_length=256 * 1024),
+    container: ChannelContainer = Depends(get_channel),
 ):
-    from yuxi.channel.container import get_channel
-
-    container = get_channel(request)
-    if not container:
-        raise HTTPException(status_code=503, detail="channel not initialized")
-
     adapter = container.adapters.get("web")
     if not adapter:
         raise HTTPException(status_code=503, detail="web adapter not available")
 
+    verifier = container.verifiers.get("web")
+    verify_result = None
+    if verifier and verifier.enabled:
+        headers = {k.lower(): v for k, v in request.headers.items()}
+        verify_result = await verifier.verify(body, headers)
+        if not verify_result.passed:
+            raise HTTPException(status_code=401, detail=verify_result.reason)
+
     try:
         raw = json.loads(body)
     except json.JSONDecodeError:
@@ -38,11 +42,13 @@ async def web_message(
     message = WebTranslator.translate_message(raw)
 
     trace_id = raw.get("trace_id", request.headers.get("x-trace-id", str(uuid4())))
-    idempotency_key = request.headers.get("idempotency-key")
+    idempotency_key = request.headers.get("idempotency-key") or trace_id
     message.metadata["trace_id"] = trace_id
-    if idempotency_key:
-        message.metadata["idempotency_key"] = idempotency_key
+    message.metadata["idempotency_key"] = idempotency_key
     message.metadata["client_ip"] = request.client.host if request.client else "unknown"
+    message.metadata.pop("auth_method", None)
+    if verify_result:
+        message.metadata["auth_method"] = verify_result.method
 
     result = await container.inbound_service.submit(message, channel_type="web", trace_id=trace_id)
 

backend/package/yuxi/channel/channels/web/translator.py

@@ -17,6 +17,9 @@ class WebTranslator:
                 kind="user",
             ),
             content=raw.get("content", ""),
-            metadata=raw.get("metadata", {}),
+            metadata={
+                **raw.get("metadata", {}),
+                "account_id": raw.get("account_id", raw.get("sender_id", "")),
+            },
             raw_payload=raw,
         )

backend/package/yuxi/channel/channels/web/verifier.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    VerifyResult,
+)
+from yuxi.channel.domain.port.external.authentication_port import (
+    AuthenticationPort,
+)
+
+
+class WebRequestVerifier:
+    def __init__(self, auth_service: AuthenticationPort | None = None, *, allow_anonymous: bool = False):
+        self._auth_service = auth_service
+        self._allow_anonymous = allow_anonymous
+
+    @property
+    def channel_type(self) -> str:
+        return "web"
+
+    @property
+    def enabled(self) -> bool:
+        return self._auth_service is not None
+
+    async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult:
+        auth_header = headers.get("authorization", "")
+        if not auth_header:
+            if self._allow_anonymous:
+                return VerifyResult(passed=True, method="web_none", reason="no auth header, anonymous allowed")
+            if not self._auth_service:
+                return VerifyResult(passed=True, method="web_none", reason="no auth_service configured")
+            return VerifyResult(passed=False, method="web_bearer", reason="no credentials configured")
+
+        if not self._auth_service:
+            return VerifyResult(passed=True, method="web_none", reason="no auth_service configured")
+
+        passed, reason = await self._auth_service.authenticate(auth_header)
+        if passed:
+            return VerifyResult(passed=True, method="web_bearer")
+        return VerifyResult(passed=False, method="web_bearer", reason=reason)

backend/package/yuxi/channel/container.py

@@ -18,31 +18,38 @@ from yuxi.channel.application.service.dispatch_service import DispatchService
 from yuxi.channel.application.service.inbound_service import InboundService
 from yuxi.channel.application.service.session_resolver import SessionResolver
 from yuxi.channel.channels._registry import get_registered_channels
-from yuxi.channel.domain.port.agent_port import AgentPort
-from yuxi.channel.domain.port.cache_port import CachePort
-from yuxi.channel.domain.port.channel_adapter_port import ChannelAdapterPort
-from yuxi.channel.domain.port.config_reload_port import ConfigReloadPort
-from yuxi.channel.domain.port.content_filter_port import ContentFilterPort
-from yuxi.channel.domain.port.event_publisher_port import EventPublisherPort
-from yuxi.channel.domain.port.metrics_port import MetricsPort
-from yuxi.channel.domain.port.queue_port import QueuePort
-from yuxi.channel.domain.port.rate_limit_port import RateLimitPort
-from yuxi.channel.domain.port.signature_verify_port import SignatureVerifyPort
-from yuxi.channel.domain.port.bot_loop_guard_port import BotLoopGuardPort
-from yuxi.channel.domain.port.circuit_breaker_port import CircuitBreakerPort
-from yuxi.channel.domain.repository.binding_repository import BindingRepositoryPort
-from yuxi.channel.domain.repository.message_log_repository import MessageLogRepositoryPort
-from yuxi.channel.domain.repository.message_repository import MessageRepositoryPort
-from yuxi.channel.domain.repository.outbox_repository import OutboxRepositoryPort
-from yuxi.channel.domain.repository.session_repository import SessionRepositoryPort
+from yuxi.channel.domain.port import (
+    AgentPort,
+    AuthenticationPort,
+    BotLoopGuardPort,
+    CachePort,
+    ChannelAdapterPort,
+    ChannelRequestVerifierPort,
+    CircuitBreakerPort,
+    ConfigReloadPort,
+    ContentFilterPort,
+    EventPublisherPort,
+    MetricsPort,
+    QueuePort,
+    RateLimitPort,
+    SignatureVerifyPort,
+)
+from yuxi.channel.domain.repository import (
+    BindingRepositoryPort,
+    MessageLogRepositoryPort,
+    MessageRepositoryPort,
+    OutboxRepositoryPort,
+    SessionRepositoryPort,
+)
 from yuxi.channel.domain.service.pipeline import Pipeline
 from yuxi.channel.infrastructure.agent.agent_adapter import AgentAdapter
-from yuxi.channel.infrastructure.cache.redis_bot_loop_guard import RedisBotLoopGuard
-from yuxi.channel.infrastructure.cache.redis_cache import RedisCache
-from yuxi.channel.infrastructure.cache.redis_circuit_breaker import RedisCircuitBreaker
-from yuxi.channel.infrastructure.cache.redis_rate_limiter import RedisRateLimiter
-from yuxi.channel.infrastructure.config.channel_config import ChannelConfig
-from yuxi.channel.infrastructure.config.redis_config_reload import RedisConfigReload
+from yuxi.channel.infrastructure.cache.repository.caching_binding_repository import CachingBindingRepository
+from yuxi.channel.infrastructure.cache_infra.redis_bot_loop_guard import RedisBotLoopGuard
+from yuxi.channel.infrastructure.cache_infra.redis_cache import RedisCache
+from yuxi.channel.infrastructure.cache_infra.redis_circuit_breaker import RedisCircuitBreaker
+from yuxi.channel.infrastructure.cache_infra.redis_rate_limiter import RedisRateLimiter
+from yuxi.channel.infrastructure.configuration.channel_config import ChannelConfig
+from yuxi.channel.infrastructure.configuration.redis_config_reload import RedisConfigReload
 from yuxi.channel.infrastructure.content_filter.composite_content_filter import CompositeContentFilter
 from yuxi.channel.infrastructure.content_filter.llm_content_filter import LlmContentFilter
 from yuxi.channel.infrastructure.content_filter.redis_content_filter import RedisContentFilter
@@ -61,7 +68,6 @@ from yuxi.channel.interfaces.websocket.manager import WsConnectionManager
 from yuxi.channel.worker.outbox_retry import OutboxRetryWorker
 from yuxi.channel.worker.pool import WorkerPool, WorkerPoolConfig
 from yuxi.channel.worker.session_factory import WorkerSessionFactory
-from yuxi.repositories.agent_config_repository import AgentConfigRepository
 from yuxi.storage.postgres.manager import pg_manager
 
 logger = logging.getLogger(__name__)
@@ -125,6 +131,7 @@ class _WorkerBundle:
     outbox_repo: OutboxRepositoryPort
     message_repo: MessageRepositoryPort
     message_log_repo: MessageLogRepositoryPort
+    verifiers: dict[str, ChannelRequestVerifierPort] = field(default_factory=dict)
 
 
 @dataclass
@@ -159,6 +166,7 @@ class ChannelContainer:
     circuit_breaker: CircuitBreakerPort | None = None
     metrics: MetricsPort | None = None
     signature_verifier: SignatureVerifyPort | None = None
+    verifiers: dict[str, ChannelRequestVerifierPort] = field(default_factory=dict)
     startup_tracer: StartupTracer = field(default_factory=StartupTracer)
     redis: aioredis.Redis | None = None
     pubsub_subscriber: RedisPubSubSubscriber | None = None
@@ -211,13 +219,6 @@ def get_channel(request: Request) -> ChannelContainer:
     return container
 
 
-async def _resolve_agent_id(session_factory, agent_config_id: int) -> str:
-    async with session_factory() as session:
-        repo = AgentConfigRepository(session)
-        config = await repo.get_by_id(config_id=agent_config_id)
-        return config.agent_id if config else "chatbot"
-
-
 class ChannelContainerFactory:
     @staticmethod
     async def create(
@@ -271,6 +272,7 @@ class ChannelContainerFactory:
             config,
             pipeline,
             redis,
+            auth_service,
             default_agent_config_id=default_agent_config_id,
             mq_workers=mq_workers,
             mq_max_concurrent=mq_max_concurrent,
@@ -319,6 +321,7 @@ class ChannelContainerFactory:
             circuit_breaker=infra.circuit_breaker,
             metrics=infra.metrics,
             signature_verifier=infra.signature_verifier,
+            verifiers=workers.verifiers,
             startup_tracer=tracer,
             redis=redis,
             pubsub_subscriber=pubsub_subscriber,
@@ -445,6 +448,9 @@ class ChannelContainerFactory:
                 overrides["app_id"] = app_id
             if app_secret:
                 overrides["app_secret"] = app_secret
+            encrypt_key = os.getenv("FEISHU_ENCRYPT_KEY", "")
+            if encrypt_key:
+                overrides["encrypt_key"] = encrypt_key
         elif channel_name == "dingtalk":
             client_id = os.getenv("DINGTALK_CLIENT_ID", "")
             client_secret = os.getenv("DINGTALK_CLIENT_SECRET", "")
@@ -495,6 +501,7 @@ class ChannelContainerFactory:
         config: ChannelConfig,
         pipeline: Pipeline,
         redis: aioredis.Redis,
+        auth_service: AuthService,
         *,
         default_agent_config_id: int = 1,
         mq_workers: int = 4,
@@ -503,13 +510,13 @@ class ChannelContainerFactory:
         session_factory = WorkerSessionFactory(pg_manager)
         db_session_factory = pg_manager.get_async_session_context
 
-        binding_repo = PgBindingRepository(db_session_factory, infra.cache_port)
-        session_repo = PgSessionRepository(
-            db_session_factory,
-            infra.cache_port,
-            agent_id_resolver=lambda aid: _resolve_agent_id(db_session_factory, aid),
+        raw_binding_repo = PgBindingRepository(db_session_factory)
+        binding_repo: BindingRepositoryPort = (
+            CachingBindingRepository(raw_binding_repo, infra.cache_port) if infra.cache_port else raw_binding_repo
         )
-        outbox_repo = PgOutboxRepository(db_session_factory, redis)
+
+        session_repo = PgSessionRepository(db_session_factory, infra.cache_port)
+        outbox_repo = PgOutboxRepository(db_session_factory, infra.cache_port)
         message_repo = PgMessageRepository(db_session_factory)
         message_log_repo = PgMessageLogRepository(db_session_factory)
 
@@ -518,6 +525,7 @@ class ChannelContainerFactory:
         session_resolver = SessionResolver(
             session_repo=session_repo,
             binding_repo=binding_repo,
+            agent_port=effective_agent_port,
             default_agent_config_id=default_agent_config_id,
         )
 
@@ -547,14 +555,17 @@ class ChannelContainerFactory:
             config=WorkerPoolConfig(num_workers=mq_workers, max_concurrent=mq_max_concurrent),
         )
 
-        outbox_worker = OutboxRetryWorker(outbox_repo, adapters, redis, metrics=infra.metrics)
+        outbox_worker = OutboxRetryWorker(outbox_repo, adapters, infra.cache_port, metrics=infra.metrics)
 
         binding_service = BindingService(binding_repo)
+        verifiers = ChannelContainerFactory._build_verifiers(config, auth_service)
         config_service = ConfigService(
             config_data=config.raw_data,
             config_reload=infra.config_reload,
             pipeline=pipeline,
             channel_config=config,
+            auth_service=auth_service,
+            verifiers=verifiers,
         )
 
         return _WorkerBundle(
@@ -569,4 +580,29 @@ class ChannelContainerFactory:
             outbox_repo=outbox_repo,
             message_repo=message_repo,
             message_log_repo=message_log_repo,
+            verifiers=verifiers,
         )
+
+    @staticmethod
+    def _build_verifiers(
+        config: ChannelConfig,
+        auth_service: AuthenticationPort,
+    ) -> dict[str, ChannelRequestVerifierPort]:
+        from yuxi.channel.channels.feishu.verifier import FeishuRequestVerifier
+        from yuxi.channel.channels.hooks.verifier import HooksRequestVerifier
+        from yuxi.channel.channels.web.verifier import WebRequestVerifier
+
+        verifiers: dict[str, ChannelRequestVerifierPort] = {}
+
+        feishu_verifier = FeishuRequestVerifier(
+            verification_token=config.feishu_verification_token or "",
+            encrypt_key=config.feishu_encrypt_key or "",
+        )
+        verifiers["feishu"] = feishu_verifier
+
+        verifiers["hooks"] = HooksRequestVerifier()
+
+        web_verifier = WebRequestVerifier(auth_service=auth_service, allow_anonymous=config.allow_anonymous)
+        verifiers["web"] = web_verifier
+
+        return verifiers

backend/package/yuxi/channel/domain/__init__.py

@@ -29,7 +29,7 @@ _LAZY_IMPORTS = {
     "KeywordMatcherPort": "yuxi.channel.domain.port",
     "CachePort": "yuxi.channel.domain.port",
     "ChannelAdapterPort": "yuxi.channel.domain.port",
-    "ChannelRouteContributor": "yuxi.channel.domain.port",
+    "ChannelRouteContributorPort": "yuxi.channel.domain.port",
     "CircuitBreakerPort": "yuxi.channel.domain.port",
     "ConfigReloadPort": "yuxi.channel.domain.port",
     "ContentFilterPort": "yuxi.channel.domain.port",
@@ -43,9 +43,9 @@ _LAZY_IMPORTS = {
     "SsePushPort": "yuxi.channel.domain.port",
     "WsConnectionPort": "yuxi.channel.domain.port",
     "BindingRepositoryPort": "yuxi.channel.domain.repository",
-    "ChannelMessageData": "yuxi.channel.domain.repository",
-    "MessageLogData": "yuxi.channel.domain.repository",
-    "MessageLogQuery": "yuxi.channel.domain.repository",
+    "ChannelMessageData": "yuxi.channel.domain.model.message.channel_message_data",
+    "MessageLogData": "yuxi.channel.domain.model.message_log.message_log_data",
+    "MessageLogQuery": "yuxi.channel.domain.model.message_log.message_log_data",
     "MessageLogRepositoryPort": "yuxi.channel.domain.repository",
     "MessageRepositoryPort": "yuxi.channel.domain.repository",
     "OutboxRepositoryPort": "yuxi.channel.domain.repository",

backend/package/yuxi/channel/domain/event/__init__.py

@@ -2,3 +2,10 @@ from yuxi.channel.domain.event.agent_error import AgentError
 from yuxi.channel.domain.event.message_blocked import MessageBlocked
 from yuxi.channel.domain.event.message_received import MessageReceived
 from yuxi.channel.domain.event.message_replied import MessageReplied
+
+__all__ = [
+    "AgentError",
+    "MessageBlocked",
+    "MessageReceived",
+    "MessageReplied",
+]

backend/package/yuxi/channel/domain/event/message_dropped.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class MessageDropped:
+    message_id: str
+    channel_type: str
+    reason: str
+    session_id: str = ""
+    trace_id: str = ""
+    metadata: dict[str, Any] = field(default_factory=dict)

backend/package/yuxi/channel/domain/exception/__init__.py

@@ -1,5 +1,23 @@
 from yuxi.channel.domain.exception.abort_mapping import ABORT_CODE_TO_HTTP
 from yuxi.channel.domain.exception.agent_crash_error import AgentCrashError
+from yuxi.channel.domain.exception.cache_error import CacheError
 from yuxi.channel.domain.exception.channel_error import ChannelError
+from yuxi.channel.domain.exception.concurrency_error import ConcurrencyError
+from yuxi.channel.domain.exception.duplicate_entity_error import DuplicateEntityError
+from yuxi.channel.domain.exception.entity_not_found_error import EntityNotFoundError
 from yuxi.channel.domain.exception.recoverable_error import RecoverableError
+from yuxi.channel.domain.exception.session_not_found_error import SessionNotFoundError
 from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+__all__ = [
+    "ABORT_CODE_TO_HTTP",
+    "AgentCrashError",
+    "CacheError",
+    "ChannelError",
+    "ConcurrencyError",
+    "DuplicateEntityError",
+    "EntityNotFoundError",
+    "RecoverableError",
+    "SessionNotFoundError",
+    "UnrecoverableError",
+]

backend/package/yuxi/channel/domain/exception/abort_mapping.py

@@ -15,4 +15,6 @@ ABORT_CODE_TO_HTTP: dict[str, int] = {
     "PAIRING_REQUIRED": 403,
     "RATE_LIMITED": 429,
     "DEDUP_SKIPPED": 200,
+    "ACCESS_DENIED": 403,
+    "PIPELINE_ABORTED": 400,
 }

backend/package/yuxi/channel/domain/exception/cache_error.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.recoverable_error import RecoverableError
+
+
+class CacheError(RecoverableError):
+    pass

backend/package/yuxi/channel/domain/exception/concurrency_error.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+
+class ConcurrencyError(UnrecoverableError):
+    pass

backend/package/yuxi/channel/domain/exception/duplicate_binding.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.channel_error import ChannelError
+
+
+class DuplicateBindingException(ChannelError):
+    def __init__(self, channel_type: str, account_id: str, group_id: str):
+        self.channel_type = channel_type
+        self.account_id = account_id
+        self.group_id = group_id
+        super().__init__(f"duplicate binding: {channel_type}/{account_id}/{group_id}")

backend/package/yuxi/channel/domain/exception/duplicate_entity_error.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.recoverable_error import RecoverableError
+
+
+class DuplicateEntityError(RecoverableError):
+    pass

backend/package/yuxi/channel/domain/exception/entity_not_found_error.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+
+class EntityNotFoundError(UnrecoverableError):
+    pass

backend/package/yuxi/channel/domain/exception/invalid_agent_config.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.channel_error import ChannelError
+
+
+class InvalidAgentConfigException(ChannelError):
+    def __init__(self, agent_config_id: int):
+        self.agent_config_id = agent_config_id
+        super().__init__(f"agent_config_id {agent_config_id} does not exist")

backend/package/yuxi/channel/domain/exception/session_not_found_error.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.exception.channel_error import ChannelError
+
+
+class SessionNotFoundError(ChannelError):
+    def __init__(self, *, thread_id: str):
+        self.thread_id = thread_id
+        super().__init__(f"Session not found for thread_id: {thread_id}")

backend/package/yuxi/channel/domain/middleware/__init__.py

@@ -1,2 +1,4 @@
 from yuxi.channel.domain.middleware.configurable import Configurable
 from yuxi.channel.domain.middleware.middleware import CallNext, Middleware
+
+__all__ = ["CallNext", "Configurable", "Middleware"]

backend/package/yuxi/channel/domain/model/__init__.py

@@ -15,3 +15,19 @@ from yuxi.channel.domain.model.shared import (
     MessagePriority,
     MessageType,
 )
+
+__all__ = [
+    "Attachment",
+    "ChannelBinding",
+    "ChannelCapabilities",
+    "ChannelSession",
+    "ChannelType",
+    "DispatchResult",
+    "MessagePriority",
+    "MessageType",
+    "OutboxEntry",
+    "Peer",
+    "SendResult",
+    "StreamChatRequest",
+    "UnifiedMessage",
+]

backend/package/yuxi/channel/domain/model/binding/__init__.py

@@ -1 +1,3 @@
 from yuxi.channel.domain.model.binding.channel_binding import ChannelBinding
+
+__all__ = ["ChannelBinding"]

backend/package/yuxi/channel/domain/model/binding/channel_binding.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import datetime
 
 
 @dataclass
@@ -14,8 +15,8 @@ class ChannelBinding:
     is_enabled: bool = True
     created_by: str | None = None
     updated_by: str | None = None
-    created_at: str = ""
-    updated_at: str = ""
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
 
     def resolve_session_key_strategy(self) -> str:
         if self.session_key_strategy != "auto":

backend/package/yuxi/channel/domain/model/message/__init__.py

@@ -3,3 +3,12 @@ from yuxi.channel.domain.model.message.dispatch_result import DispatchResult, Se
 from yuxi.channel.domain.model.message.peer import Peer
 from yuxi.channel.domain.model.message.stream_chat_request import StreamChatRequest
 from yuxi.channel.domain.model.message.unified_message import UnifiedMessage
+
+__all__ = [
+    "Attachment",
+    "DispatchResult",
+    "Peer",
+    "SendResult",
+    "StreamChatRequest",
+    "UnifiedMessage",
+]

backend/package/yuxi/channel/domain/model/message/channel_message_data.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class ChannelMessageData:
+    id: int
+    thread_id: str
+    role: str
+    content: str
+    message_id: str | None
+    extra_metadata: dict | None

backend/package/yuxi/channel/domain/model/message_log/message_log_data.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class MessageLogData:
+    id: int
+    trace_id: str
+    message_id: str
+    channel_type: str
+    conversation_id: int | None
+    session_id: str | None
+    direction: str
+    sender_id: str | None
+    content_summary: str | None
+    agent_config_id: int | None
+    status: str
+    pipeline_result: str | None
+    abort_reason: str | None
+    worker_result: str | None
+    error_message: str | None
+    processing_time_ms: int | None
+
+
+@dataclass
+class MessageLogQuery:
+    channel_type: str | None = None
+    conversation_id: int | None = None
+    status: str | None = None
+    direction: str | None = None
+    pipeline_result: str | None = None
+    worker_result: str | None = None
+    limit: int = 50
+    offset: int = 0

backend/package/yuxi/channel/domain/model/outbox/__init__.py

@@ -1 +1,3 @@
 from yuxi.channel.domain.model.outbox.outbox_entry import OutboxEntry
+
+__all__ = ["OutboxEntry"]

backend/package/yuxi/channel/domain/model/outbox/outbox_entry.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import datetime
 
 
 @dataclass
@@ -13,7 +14,7 @@ class OutboxEntry:
     status: str
     retry_count: int
     max_retries: int
-    next_retry_at: str | None
+    next_retry_at: datetime | None
     last_error: str | None
     trace_id: str | None
     extra_metadata: dict | None

backend/package/yuxi/channel/domain/model/outbox/outbox_status.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from enum import StrEnum
+
+
+class OutboxStatus(StrEnum):
+    PENDING = "pending"
+    PROCESSING = "processing"
+    RETRYING = "retrying"
+    SENT = "sent"
+    DEAD = "dead"
+    FAILED = "failed"

backend/package/yuxi/channel/domain/model/session/__init__.py

@@ -1 +1,3 @@
 from yuxi.channel.domain.model.session.channel_session import ChannelSession
+
+__all__ = ["ChannelSession"]

backend/package/yuxi/channel/domain/model/shared/__init__.py

@@ -3,3 +3,11 @@ from yuxi.channel.domain.model.shared.channel_type import ChannelType
 from yuxi.channel.domain.model.shared.content_chunker import chunk_content
 from yuxi.channel.domain.model.shared.message_priority import MessagePriority
 from yuxi.channel.domain.model.shared.message_type import MessageType
+
+__all__ = [
+    "ChannelCapabilities",
+    "ChannelType",
+    "MessagePriority",
+    "MessageType",
+    "chunk_content",
+]

backend/package/yuxi/channel/domain/port/__init__.py

@@ -1,16 +1,49 @@
-from yuxi.channel.domain.port.agent_port import AgentPort
-from yuxi.channel.domain.port.bot_loop_guard_port import BotLoopGuardPort
-from yuxi.channel.domain.port.cache_port import CachePort
-from yuxi.channel.domain.port.channel_adapter_port import ChannelAdapterPort
-from yuxi.channel.domain.port.channel_route_contributor import ChannelRouteContributor
-from yuxi.channel.domain.port.circuit_breaker_port import CircuitBreakerPort
-from yuxi.channel.domain.port.config_reload_port import ConfigReloadPort
-from yuxi.channel.domain.port.content_filter_port import ContentFilterPort, FilterResult
-from yuxi.channel.domain.port.event_publisher_port import DomainEvent, EventPublisherPort
-from yuxi.channel.domain.port.keyword_matcher_port import KeywordMatcherPort
-from yuxi.channel.domain.port.metrics_port import MetricsPort
-from yuxi.channel.domain.port.queue_port import QueuePort
-from yuxi.channel.domain.port.rate_limit_port import RateLimitPort
-from yuxi.channel.domain.port.signature_verify_port import SignatureVerifyPort
-from yuxi.channel.domain.port.sse_push_port import SsePushPort
-from yuxi.channel.domain.port.ws_connection_port import WsConnectionPort
+from yuxi.channel.domain.port.external import (
+    AuthenticationPort,
+    ChannelAdapterPort,
+    ChannelRequestVerifierPort,
+    ChannelRouteContributorPort,
+    SignatureVerifyPort,
+    VerifyResult,
+    WsConnectionPort,
+)
+from yuxi.channel.domain.port.internal import (
+    AgentPort,
+    BotLoopGuardPort,
+    CachePort,
+    CircuitBreakerPort,
+    ConfigReloadPort,
+    ContentFilterPort,
+    DomainEvent,
+    EventPublisherPort,
+    FilterResult,
+    KeywordMatcherPort,
+    MetricsPort,
+    QueuePort,
+    RateLimitPort,
+    SsePushPort,
+)
+
+__all__ = [
+    "AgentPort",
+    "AuthenticationPort",
+    "BotLoopGuardPort",
+    "CachePort",
+    "ChannelAdapterPort",
+    "ChannelRequestVerifierPort",
+    "ChannelRouteContributorPort",
+    "CircuitBreakerPort",
+    "ConfigReloadPort",
+    "ContentFilterPort",
+    "DomainEvent",
+    "EventPublisherPort",
+    "FilterResult",
+    "KeywordMatcherPort",
+    "MetricsPort",
+    "QueuePort",
+    "RateLimitPort",
+    "SignatureVerifyPort",
+    "SsePushPort",
+    "VerifyResult",
+    "WsConnectionPort",
+]

backend/package/yuxi/channel/domain/port/external/__init__.py

@@ -0,0 +1,19 @@
+from yuxi.channel.domain.port.external.authentication_port import AuthenticationPort
+from yuxi.channel.domain.port.external.channel_adapter_port import ChannelAdapterPort
+from yuxi.channel.domain.port.external.channel_request_verifier_port import (
+    ChannelRequestVerifierPort,
+    VerifyResult,
+)
+from yuxi.channel.domain.port.external.channel_route_contributor_port import ChannelRouteContributorPort
+from yuxi.channel.domain.port.external.signature_verify_port import SignatureVerifyPort
+from yuxi.channel.domain.port.external.ws_connection_port import WsConnectionPort
+
+__all__ = [
+    "AuthenticationPort",
+    "ChannelAdapterPort",
+    "ChannelRequestVerifierPort",
+    "ChannelRouteContributorPort",
+    "SignatureVerifyPort",
+    "VerifyResult",
+    "WsConnectionPort",
+]

backend/package/yuxi/channel/domain/port/external/authentication_port.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class AuthenticationPort(Protocol):
+    async def authenticate(self, auth_header: str, *, client_id: str = "unknown") -> tuple[bool, str]: ...

backend/package/yuxi/channel/domain/port/external/channel_request_verifier_port.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Protocol, runtime_checkable
+
+
+@dataclass(frozen=True)
+class VerifyResult:
+    passed: bool
+    method: str = ""
+    reason: str = ""
+
+
+@runtime_checkable
+class ChannelRequestVerifierPort(Protocol):
+    @property
+    def channel_type(self) -> str: ...
+
+    @property
+    def enabled(self) -> bool: ...
+
+    async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult: ...

backend/package/yuxi/channel/domain/port/external/channel_route_contributor_port.py

@@ -4,6 +4,6 @@ from typing import Protocol, runtime_checkable
 
 
 @runtime_checkable
-class ChannelRouteContributor(Protocol):
+class ChannelRouteContributorPort(Protocol):
     @property
     def router(self) -> object: ...

backend/package/yuxi/channel/domain/port/internal/__init__.py

@@ -0,0 +1,29 @@
+from yuxi.channel.domain.port.internal.agent_port import AgentPort
+from yuxi.channel.domain.port.internal.bot_loop_guard_port import BotLoopGuardPort
+from yuxi.channel.domain.port.internal.cache_port import CachePort
+from yuxi.channel.domain.port.internal.circuit_breaker_port import CircuitBreakerPort
+from yuxi.channel.domain.port.internal.config_reload_port import ConfigReloadPort
+from yuxi.channel.domain.port.internal.content_filter_port import ContentFilterPort, FilterResult
+from yuxi.channel.domain.port.internal.event_publisher_port import DomainEvent, EventPublisherPort
+from yuxi.channel.domain.port.internal.keyword_matcher_port import KeywordMatcherPort
+from yuxi.channel.domain.port.internal.metrics_port import MetricsPort
+from yuxi.channel.domain.port.internal.queue_port import QueuePort
+from yuxi.channel.domain.port.internal.rate_limit_port import RateLimitPort
+from yuxi.channel.domain.port.internal.sse_push_port import SsePushPort
+
+__all__ = [
+    "AgentPort",
+    "BotLoopGuardPort",
+    "CachePort",
+    "CircuitBreakerPort",
+    "ConfigReloadPort",
+    "ContentFilterPort",
+    "DomainEvent",
+    "EventPublisherPort",
+    "FilterResult",
+    "KeywordMatcherPort",
+    "MetricsPort",
+    "QueuePort",
+    "RateLimitPort",
+    "SsePushPort",
+]

backend/package/yuxi/channel/domain/port/internal/agent_port.py

@@ -11,3 +11,5 @@ class AgentPort(Protocol):
     async def stream_chat(self, request: StreamChatRequest) -> str: ...
 
     async def stream_chat_iter(self, request: StreamChatRequest) -> AsyncGenerator[str, None]: ...
+
+    async def resolve_agent_id(self, agent_config_id: int) -> str: ...

backend/package/yuxi/channel/domain/port/internal/cache_port.py

@@ -18,3 +18,5 @@ class CachePort(Protocol):
     async def ttl(self, key: str) -> int: ...
 
     async def eval(self, script: str, keys: list[str], args: list[str | int]) -> tuple: ...
+
+    async def publish(self, channel: str, message: str) -> None: ...

backend/package/yuxi/channel/domain/port/internal/event_publisher_port.py

@@ -4,11 +4,12 @@ from typing import Protocol, runtime_checkable
 
 from yuxi.channel.domain.event.gateway_shutdown import GatewayShutdown
 from yuxi.channel.domain.event.message_blocked import MessageBlocked
+from yuxi.channel.domain.event.message_dropped import MessageDropped
 from yuxi.channel.domain.event.message_received import MessageReceived
 from yuxi.channel.domain.event.message_replied import MessageReplied
 from yuxi.channel.domain.event.agent_error import AgentError
 
-DomainEvent = MessageReceived | MessageReplied | MessageBlocked | AgentError | GatewayShutdown
+DomainEvent = MessageReceived | MessageReplied | MessageBlocked | MessageDropped | AgentError | GatewayShutdown
 
 
 @runtime_checkable

backend/package/yuxi/channel/domain/port/internal/unit_of_work.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class UnitOfWork(Protocol):
+    async def commit(self) -> None: ...
+
+    async def rollback(self) -> None: ...
+
+
+@runtime_checkable
+class HasSession(Protocol):
+    def session(self) -> object: ...

backend/package/yuxi/channel/domain/repository/__init__.py

@@ -1,12 +1,13 @@
 from yuxi.channel.domain.repository.binding_repository import BindingRepositoryPort
-from yuxi.channel.domain.repository.message_log_repository import (
-    MessageLogData,
-    MessageLogQuery,
-    MessageLogRepositoryPort,
-)
-from yuxi.channel.domain.repository.message_repository import (
-    ChannelMessageData,
-    MessageRepositoryPort,
-)
+from yuxi.channel.domain.repository.message_log_repository import MessageLogRepositoryPort
+from yuxi.channel.domain.repository.message_repository import MessageRepositoryPort
 from yuxi.channel.domain.repository.outbox_repository import OutboxRepositoryPort
 from yuxi.channel.domain.repository.session_repository import SessionRepositoryPort
+
+__all__ = [
+    "BindingRepositoryPort",
+    "MessageLogRepositoryPort",
+    "MessageRepositoryPort",
+    "OutboxRepositoryPort",
+    "SessionRepositoryPort",
+]

backend/package/yuxi/channel/domain/repository/binding_repository.py

@@ -7,7 +7,9 @@ from yuxi.channel.domain.model.binding.channel_binding import ChannelBinding
 
 @runtime_checkable
 class BindingRepositoryPort(Protocol):
-    async def find_binding(self, *, channel_type: str, account_id: str, group_id: str) -> ChannelBinding | None: ...
+    async def find_active_binding(
+        self, *, channel_type: str, account_id: str, group_id: str
+    ) -> ChannelBinding | None: ...
 
     async def create_binding(
         self,
@@ -28,12 +30,10 @@ class BindingRepositoryPort(Protocol):
         updated_by: str | None = None,
     ) -> ChannelBinding | None: ...
 
-    async def delete_binding(self, binding_id: int, *, updated_by: str | None = None) -> bool: ...
+    async def delete_binding(self, binding_id: int, *, updated_by: str | None = None) -> ChannelBinding | None: ...
 
     async def get_binding(self, binding_id: int) -> ChannelBinding | None: ...
 
     async def list_bindings(
         self, *, channel_type: str | None = None, offset: int = 0, limit: int = 50
     ) -> tuple[list[ChannelBinding], int]: ...
-
-    async def invalidate_cache(self, *, channel_type: str, account_id: str, group_id: str) -> None: ...

backend/package/yuxi/channel/domain/repository/message_log_repository.py

@@ -1,39 +1,11 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
 from typing import Protocol, runtime_checkable
 
-
-@dataclass
-class MessageLogData:
-    id: int
-    trace_id: str
-    message_id: str
-    channel_type: str
-    conversation_id: int | None
-    session_id: str | None
-    direction: str
-    sender_id: str | None
-    content_summary: str | None
-    agent_config_id: int | None
-    status: str
-    pipeline_result: str | None
-    abort_reason: str | None
-    worker_result: str | None
-    error_message: str | None
-    processing_time_ms: int | None
-
-
-@dataclass
-class MessageLogQuery:
-    channel_type: str | None = None
-    conversation_id: int | None = None
-    status: str | None = None
-    direction: str | None = None
-    pipeline_result: str | None = None
-    worker_result: str | None = None
-    limit: int = 50
-    offset: int = 0
+from yuxi.channel.domain.model.message_log.message_log_data import (
+    MessageLogData,
+    MessageLogQuery,
+)
 
 
 @runtime_checkable
@@ -50,6 +22,7 @@ class MessageLogRepositoryPort(Protocol):
         agent_config_id: int | None = None,
         session_id: str | None = None,
         conversation_id: int | None = None,
+        status: str = "received",
     ) -> MessageLogData: ...
 
     async def update_pipeline_result(
@@ -58,6 +31,7 @@ class MessageLogRepositoryPort(Protocol):
         trace_id: str,
         message_id: str,
         pipeline_result: str,
+        status: str,
         abort_reason: str | None = None,
     ) -> bool: ...
 
@@ -71,7 +45,7 @@ class MessageLogRepositoryPort(Protocol):
         error_message: str | None = None,
         processing_time_ms: int | None = None,
         agent_config_id: int | None = None,
-        session_id: int | None = None,
+        session_id: str | None = None,
         conversation_id: int | None = None,
     ) -> bool: ...
 
@@ -79,4 +53,4 @@ class MessageLogRepositoryPort(Protocol):
 
     async def get_by_message_id(self, message_id: str) -> list[MessageLogData]: ...
 
-    async def query_logs(self, query: MessageLogQuery) -> list[MessageLogData]: ...
+    async def query_logs(self, query: MessageLogQuery) -> tuple[list[MessageLogData], int]: ...

backend/package/yuxi/channel/domain/repository/message_repository.py

@@ -1,17 +1,8 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
 from typing import Protocol, runtime_checkable
 
-
-@dataclass
-class ChannelMessageData:
-    id: int
-    thread_id: str
-    role: str
-    content: str
-    message_id: str | None
-    extra_metadata: dict | None
+from yuxi.channel.domain.model.message.channel_message_data import ChannelMessageData
 
 
 @runtime_checkable

backend/package/yuxi/channel/domain/repository/outbox_repository.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from typing import Protocol, runtime_checkable
 
 from yuxi.channel.domain.model.outbox.outbox_entry import OutboxEntry
+from yuxi.channel.domain.port.internal.unit_of_work import UnitOfWork
 
 
 @runtime_checkable
@@ -16,8 +17,11 @@ class OutboxRepositoryPort(Protocol):
         content: str,
         trace_id: str | None = None,
         extra_metadata: dict | None = None,
+        uow: UnitOfWork | None = None,
     ) -> OutboxEntry: ...
 
+    async def fetch_and_mark_pending(self, *, limit: int = 20) -> list[OutboxEntry]: ...
+
     async def fetch_pending(self, *, limit: int = 20) -> list[OutboxEntry]: ...
 
     async def get_by_id(self, entry_id: int) -> OutboxEntry | None: ...
@@ -35,5 +39,6 @@ class OutboxRepositoryPort(Protocol):
         *,
         channel_type: str | None = None,
         status: str | None = None,
+        offset: int = 0,
         limit: int = 50,
     ) -> list[OutboxEntry]: ...

backend/package/yuxi/channel/domain/repository/session_repository.py

@@ -12,7 +12,7 @@ class SessionRepositoryPort(Protocol):
         *,
         channel_type: str,
         account_id: str,
-        agent_config_id: int,
+        agent_id: str,
     ) -> ChannelSession: ...
 
     async def get_by_thread_id(self, thread_id: str) -> ChannelSession | None: ...

backend/package/yuxi/channel/infrastructure/agent/agent_adapter.py

@@ -6,6 +6,7 @@ from collections.abc import AsyncGenerator
 from sqlalchemy import select
 
 from yuxi.channel.domain.model.message.stream_chat_request import StreamChatRequest
+from yuxi.repositories.agent_config_repository import AgentConfigRepository
 from yuxi.storage.postgres.models_business import User
 from yuxi.utils.logging_config import logger
 
@@ -76,3 +77,9 @@ class AgentAdapter:
                         return
                 except (json.JSONDecodeError, AttributeError):
                     pass
+
+    async def resolve_agent_id(self, agent_config_id: int) -> str:
+        async with self._session_factory() as session:
+            repo = AgentConfigRepository(session)
+            config = await repo.get_by_id(config_id=agent_config_id)
+            return config.agent_id if config else "chatbot"

backend/package/yuxi/channel/infrastructure/cache_infra/redis_bot_loop_guard.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.domain.port.cache_port import CachePort
+
+logger = logging.getLogger(__name__)
+
+
+class RedisBotLoopGuard:
+    def __init__(
+        self,
+        cache_port: CachePort,
+        *,
+        dm_budget: int = 10,
+        group_budget: int = 20,
+        window_seconds: int = 60,
+        cooldown_seconds: int = 300,
+    ) -> None:
+        self._cache = cache_port
+        self._dm_budget = dm_budget
+        self._group_budget = group_budget
+        self._window = window_seconds
+        self._cooldown = cooldown_seconds
+
+    async def check(self, session_id: str, *, sender_id: str = "", is_group: bool = False) -> bool:
+        if is_group and not sender_id:
+            logger.warning(
+                "bot loop guard: group message without sender_id rejected, session=%s",
+                session_id,
+            )
+            return False
+
+        if is_group and sender_id:
+            key = f"channel:bot_loop:group:{session_id}:{sender_id}"
+            budget = self._group_budget
+        else:
+            key = f"channel:bot_loop:dm:{session_id}"
+            budget = self._dm_budget
+
+        count = await self._cache.incr(key)
+        if count == 1:
+            await self._cache.expire(key, self._window)
+
+        if count > budget:
+            await self._cache.expire(key, self._cooldown)
+            return False
+
+        return True
+
+    async def reset(self, session_id: str, *, sender_id: str = "", is_group: bool = False) -> None:
+        if is_group and sender_id:
+            key = f"channel:bot_loop:group:{session_id}:{sender_id}"
+        else:
+            key = f"channel:bot_loop:dm:{session_id}"
+        await self._cache.delete(key)

backend/package/yuxi/channel/infrastructure/cache_infra/redis_cache.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import logging
+
+import redis.asyncio as aioredis
+
+from yuxi.channel.domain.exception.cache_error import CacheError
+from yuxi.channel.domain.port import CachePort
+
+logger = logging.getLogger(__name__)
+
+
+class RedisCache(CachePort):
+    def __init__(self, redis: aioredis.Redis) -> None:
+        self._redis = redis
+
+    async def get(self, key: str) -> str | None:
+        try:
+            raw = await self._redis.get(key)
+            if raw is None:
+                return None
+            return raw.decode() if isinstance(raw, bytes) else raw
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache get failed for key {key}") from exc
+
+    async def set(self, key: str, value: str, *, ex: int | None = None, nx: bool = False) -> bool:
+        try:
+            result = await self._redis.set(key, value, ex=ex, nx=nx)
+            if result is None:
+                return False
+            if isinstance(result, bool):
+                return result
+            if isinstance(result, bytes):
+                return result == b"OK"
+            return str(result) == "OK"
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache set failed for key {key}") from exc
+
+    async def delete(self, key: str) -> None:
+        try:
+            await self._redis.delete(key)
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache delete failed for key {key}") from exc
+
+    async def incr(self, key: str) -> int:
+        try:
+            return await self._redis.incr(key)
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache incr failed for key {key}") from exc
+
+    async def expire(self, key: str, seconds: int) -> None:
+        try:
+            await self._redis.expire(key, seconds)
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache expire failed for key {key}") from exc
+
+    async def ttl(self, key: str) -> int:
+        try:
+            return await self._redis.ttl(key)
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache ttl failed for key {key}") from exc
+
+    async def eval(self, script: str, keys: list[str], args: list[str | int]) -> tuple:
+        try:
+            return await self._redis.eval(script, len(keys), *keys, *args)
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError("Cache eval failed") from exc
+
+    async def publish(self, channel: str, message: str) -> None:
+        try:
+            await self._redis.publish(channel, message)
+        except (aioredis.RedisError, OSError, ConnectionError) as exc:
+            raise CacheError(f"Cache publish failed for channel {channel}") from exc

backend/package/yuxi/channel/infrastructure/cache_infra/redis_circuit_breaker.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.port.cache_port import CachePort
+
+
+_IS_AVAILABLE_SCRIPT = """
+local state_key = KEYS[1]
+local probe_key = KEYS[2]
+local recovery_timeout = tonumber(ARGV[1])
+local half_open_max = tonumber(ARGV[2])
+
+local state = redis.call('GET', state_key)
+if not state then
+    return {1, 'closed'}
+end
+
+if state == 'open' then
+    local ttl = redis.call('TTL', state_key)
+    if ttl > 0 then
+        return {0, 'open'}
+    end
+    redis.call('SET', state_key, 'half_open', 'EX', recovery_timeout)
+    redis.call('DEL', probe_key)
+    return {1, 'half_open'}
+end
+
+if state == 'half_open' then
+    local count = redis.call('INCR', probe_key)
+    if count == 1 then
+        redis.call('EXPIRE', probe_key, recovery_timeout)
+    end
+    if count <= half_open_max then
+        return {1, 'half_open'}
+    end
+    return {0, 'half_open'}
+end
+
+return {1, 'closed'}
+"""
+
+_RECORD_FAILURE_SCRIPT = """
+local failure_key = KEYS[1]
+local state_key = KEYS[2]
+local probe_key = KEYS[3]
+local recovery_timeout = tonumber(ARGV[1])
+local failure_threshold = tonumber(ARGV[2])
+
+local count = redis.call('INCR', failure_key)
+if count == 1 then
+    redis.call('EXPIRE', failure_key, recovery_timeout * 2)
+end
+
+if count >= failure_threshold then
+    redis.call('SET', state_key, 'open', 'EX', recovery_timeout)
+    redis.call('DEL', probe_key)
+    return count
+end
+
+return count
+"""
+
+
+class RedisCircuitBreaker:
+    def __init__(
+        self,
+        cache_port: CachePort,
+        *,
+        failure_threshold: int = 5,
+        recovery_timeout: int = 30,
+        half_open_max: int = 1,
+    ) -> None:
+        self._cache = cache_port
+        self._failure_threshold = failure_threshold
+        self._recovery_timeout = recovery_timeout
+        self._half_open_max = half_open_max
+
+    async def is_available(self, agent_config_id: int) -> bool:
+        state_key = f"channel:circuit:state:{agent_config_id}"
+        probe_key = f"channel:circuit:probe:{agent_config_id}"
+
+        result = await self._cache.eval(
+            _IS_AVAILABLE_SCRIPT,
+            keys=[state_key, probe_key],
+            args=[str(self._recovery_timeout), str(self._half_open_max)],
+        )
+        allowed = result[0]
+        return bool(allowed)
+
+    async def record_success(self, agent_config_id: int) -> None:
+        state_key = f"channel:circuit:state:{agent_config_id}"
+        failure_key = f"channel:circuit:failures:{agent_config_id}"
+        probe_key = f"channel:circuit:probe:{agent_config_id}"
+        await self._cache.delete(failure_key)
+        await self._cache.delete(probe_key)
+        await self._cache.delete(state_key)
+
+    async def record_failure(self, agent_config_id: int) -> None:
+        failure_key = f"channel:circuit:failures:{agent_config_id}"
+        state_key = f"channel:circuit:state:{agent_config_id}"
+        probe_key = f"channel:circuit:probe:{agent_config_id}"
+
+        await self._cache.eval(
+            _RECORD_FAILURE_SCRIPT,
+            keys=[failure_key, state_key, probe_key],
+            args=[str(self._recovery_timeout), str(self._failure_threshold)],
+        )

backend/package/yuxi/channel/infrastructure/cache_infra/redis_rate_limiter.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import logging
+
+import redis.asyncio as aioredis
+
+logger = logging.getLogger(__name__)
+
+
+class RedisRateLimiter:
+    def __init__(self, redis: aioredis.Redis) -> None:
+        self._redis = redis
+
+    async def check_and_incr(
+        self, key: str, *, max_attempts: int, window_seconds: int, lockout_seconds: int = 0
+    ) -> bool:
+        count = await self._redis.incr(key)
+        if count == 1:
+            await self._redis.expire(key, window_seconds)
+        if count <= max_attempts:
+            return True
+        if lockout_seconds > 0:
+            lockout_key = key.replace(":attempts:", ":lockout:")
+            await self._redis.set(lockout_key, "1", ex=lockout_seconds)
+        return False
+
+    async def is_locked(self, key: str) -> tuple[bool, int]:
+        ttl = await self._redis.ttl(key)
+        if ttl is None or ttl < 0:
+            return False, 0
+        return True, ttl
+
+    async def reset(self, key: str) -> None:
+        await self._redis.delete(key)

backend/package/yuxi/channel/infrastructure/configuration/channel_config.py

@@ -28,11 +28,19 @@ class ChannelConfig:
 
     async def on_config_updated(self, config: dict) -> list[str]:
         old_auth_token = self._data.get("auth", {}).get("token")
+        old_auth_password = self._data.get("auth", {}).get("password")
+        old_feishu_encrypt_key = self._data.get("feishu", {}).get("encrypt_key")
         self._data = config
         updated = []
         new_auth_token = config.get("auth", {}).get("token")
+        new_auth_password = config.get("auth", {}).get("password")
+        new_feishu_encrypt_key = config.get("feishu", {}).get("encrypt_key")
         if old_auth_token != new_auth_token:
             updated.append("auth_token")
+        if old_auth_password != new_auth_password:
+            updated.append("auth_password")
+        if old_feishu_encrypt_key != new_feishu_encrypt_key:
+            updated.append("feishu_encrypt_key")
         return updated
 
     @property
@@ -64,6 +72,10 @@ class ChannelConfig:
     def feishu_verification_token(self) -> str | None:
         return self._data.get("feishu", {}).get("verification_token")
 
+    @property
+    def feishu_encrypt_key(self) -> str | None:
+        return self._data.get("feishu", {}).get("encrypt_key")
+
     @property
     def max_auth_attempts(self) -> int:
         return self._data.get("auth", {}).get("max_attempts", 5)
@@ -72,6 +84,10 @@ class ChannelConfig:
     def lockout_seconds(self) -> int:
         return self._data.get("auth", {}).get("lockout_seconds", 300)
 
+    @property
+    def allow_anonymous(self) -> bool:
+        return self._data.get("auth", {}).get("allow_anonymous", False)
+
     @property
     def mention_gate_config(self) -> dict:
         return self._data.get("mention_gate", {})

backend/package/yuxi/channel/infrastructure/persistence/converter/binding_converter.py

@@ -11,10 +11,10 @@ def to_domain(row: ChannelBindingModel) -> ChannelBinding:
         account_id=row.account_id,
         group_id=row.group_id,
         agent_config_id=row.agent_config_id,
-        session_key_strategy=getattr(row, "session_key_strategy", "auto"),
+        session_key_strategy=row.session_key_strategy,
         is_enabled=bool(row.is_enabled),
         created_by=row.created_by,
         updated_by=row.updated_by,
-        created_at=row.created_at.isoformat() if row.created_at else "",
-        updated_at=row.updated_at.isoformat() if row.updated_at else "",
+        created_at=row.created_at,
+        updated_at=row.updated_at,
     )

backend/package/yuxi/channel/infrastructure/persistence/converter/message_log_converter.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.model.message_log.message_log_data import MessageLogData
+from yuxi.storage.postgres.models_channel import ChannelMessageLog
+
+
+def to_data(row: ChannelMessageLog) -> MessageLogData:
+    return MessageLogData(
+        id=row.id,
+        trace_id=row.trace_id,
+        message_id=row.message_id,
+        channel_type=row.channel_type,
+        conversation_id=row.conversation_id,
+        session_id=row.session_id,
+        direction=row.direction,
+        sender_id=row.sender_id,
+        content_summary=row.content_summary,
+        agent_config_id=row.agent_config_id,
+        status=row.status,
+        pipeline_result=row.pipeline_result,
+        abort_reason=row.abort_reason,
+        worker_result=row.worker_result,
+        error_message=row.error_message,
+        processing_time_ms=row.processing_time_ms,
+    )

backend/package/yuxi/channel/infrastructure/persistence/converter/outbox_converter.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.model.outbox.outbox_entry import OutboxEntry
+from yuxi.storage.postgres.models_channel import ChannelOutbox
+
+
+def to_domain(row: ChannelOutbox) -> OutboxEntry:
+    return OutboxEntry(
+        id=row.id,
+        message_id=row.message_id,
+        session_id=row.session_id,
+        channel_type=row.channel_type,
+        content=row.content,
+        status=row.status,
+        retry_count=row.retry_count,
+        max_retries=row.max_retries,
+        next_retry_at=row.next_retry_at,
+        last_error=row.last_error,
+        trace_id=row.trace_id,
+        extra_metadata=row.extra_metadata,
+    )

backend/package/yuxi/channel/infrastructure/persistence/converter/session_converter.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from yuxi.channel.domain.model.session.channel_session import ChannelSession
+from yuxi.storage.postgres.models_business import Conversation
+
+
+def to_domain(row: Conversation) -> ChannelSession:
+    return ChannelSession(
+        id=row.id,
+        thread_id=row.thread_id,
+        user_id=row.user_id,
+        agent_id=row.agent_id,
+        channel_type=row.channel_type,
+        channel_session_key=row.channel_session_key,
+        status=row.status,
+        title=row.title,
+    )

backend/package/yuxi/channel/infrastructure/persistence/exception_translator.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from sqlalchemy.exc import IntegrityError as SaIntegrityError
+from sqlalchemy.exc import OperationalError as SaOperationalError
+
+from yuxi.channel.domain.exception.concurrency_error import ConcurrencyError
+from yuxi.channel.domain.exception.duplicate_entity_error import DuplicateEntityError
+from yuxi.channel.domain.exception.entity_not_found_error import EntityNotFoundError
+from yuxi.channel.domain.exception.unrecoverable_error import UnrecoverableError
+
+
+def translate_db_exception(exc: Exception) -> UnrecoverableError:
+    if isinstance(exc, SaIntegrityError):
+        msg = str(exc.orig) if hasattr(exc, "orig") else str(exc)
+        if "unique" in msg.lower() or "duplicate" in msg.lower():
+            return DuplicateEntityError(msg)
+        if "foreign key" in msg.lower():
+            return EntityNotFoundError(msg)
+        return ConcurrencyError(msg)
+    if isinstance(exc, SaOperationalError):
+        return UnrecoverableError(f"Database operational error: {exc}")
+    return UnrecoverableError(str(exc))

backend/package/yuxi/channel/infrastructure/persistence/repository/pg_binding_repository.py

@@ -1,46 +1,29 @@
 from __future__ import annotations
 
-import json
 import logging
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError as SaIntegrityError
 
 from yuxi.channel.domain.model.binding.channel_binding import ChannelBinding
-from yuxi.channel.domain.port.cache_port import CachePort
+from yuxi.channel.domain.repository.binding_repository import BindingRepositoryPort
 from yuxi.channel.infrastructure.persistence.converter.binding_converter import (
     to_domain,
 )
+from yuxi.channel.infrastructure.persistence.exception_translator import (
+    translate_db_exception,
+)
 from yuxi.storage.postgres.models_channel import ChannelBinding as ChannelBindingModel
 from yuxi.utils.datetime_utils import utc_now_naive
 
 logger = logging.getLogger(__name__)
 
-_BINDING_CACHE_TTL = 60
-_BINDING_NULL_CACHE_TTL = 10
-_BINDING_NULL_MARKER = "__none__"
 
-
-class PgBindingRepository:
-    def __init__(self, session_factory, cache_port: CachePort | None = None):
+class PgBindingRepository(BindingRepositoryPort):
+    def __init__(self, session_factory):
         self._session_factory = session_factory
-        self._cache = cache_port
-
-    def _cache_key(self, channel_type: str, account_id: str, group_id: str) -> str:
-        return f"channel:binding:{channel_type}:{account_id}:{group_id}"
-
-    async def find_binding(self, *, channel_type: str, account_id: str, group_id: str) -> ChannelBinding | None:
-        if self._cache:
-            try:
-                cache_key = self._cache_key(channel_type, account_id, group_id)
-                cached = await self._cache.get(cache_key)
-                if cached:
-                    if cached == _BINDING_NULL_MARKER:
-                        return None
-                    data = json.loads(cached)
-                    return ChannelBinding(**data)
-            except Exception:
-                logger.warning("Redis cache read failed, falling back to DB")
 
+    async def find_active_binding(self, *, channel_type: str, account_id: str, group_id: str) -> ChannelBinding | None:
         async with self._session_factory() as db:
             stmt = select(ChannelBindingModel).where(
                 ChannelBindingModel.channel_type == channel_type,
@@ -51,28 +34,7 @@ class PgBindingRepository:
             )
             result = await db.execute(stmt)
             row = result.scalar_one_or_none()
-            if not row:
-                if self._cache:
-                    try:
-                        cache_key = self._cache_key(channel_type, account_id, group_id)
-                        await self._cache.set(cache_key, _BINDING_NULL_MARKER, ex=_BINDING_NULL_CACHE_TTL)
-                    except Exception:
-                        logger.warning("Redis null cache write failed")
-                return None
-
-            binding = to_domain(row)
-            if self._cache:
-                try:
-                    cache_key = self._cache_key(channel_type, account_id, group_id)
-                    await self._cache.set(
-                        cache_key,
-                        json.dumps(binding.__dict__, ensure_ascii=False),
-                        ex=_BINDING_CACHE_TTL,
-                    )
-                except Exception:
-                    logger.warning("Redis cache write failed, skipping cache")
-
-            return binding
+            return to_domain(row) if row else None
 
     async def create_binding(
         self,
@@ -83,20 +45,23 @@ class PgBindingRepository:
         agent_config_id: int,
         created_by: str | None = None,
     ) -> ChannelBinding:
-        async with self._session_factory() as db:
-            row = ChannelBindingModel(
-                channel_type=channel_type,
-                account_id=account_id,
-                group_id=group_id,
-                agent_config_id=agent_config_id,
-                is_enabled=True,
-                created_by=created_by,
-                updated_by=created_by,
-            )
-            db.add(row)
-            await db.flush()
-            await db.refresh(row)
-            return to_domain(row)
+        try:
+            async with self._session_factory() as db:
+                row = ChannelBindingModel(
+                    channel_type=channel_type,
+                    account_id=account_id,
+                    group_id=group_id,
+                    agent_config_id=agent_config_id,
+                    is_enabled=True,
+                    created_by=created_by,
+                    updated_by=created_by,
+                )
+                db.add(row)
+                await db.flush()
+                await db.refresh(row)
+                return to_domain(row)
+        except SaIntegrityError as exc:
+            raise translate_db_exception(exc) from exc
 
     async def update_binding(
         self,
@@ -116,8 +81,6 @@ class PgBindingRepository:
             if not row:
                 return None
 
-            cache_keys = (row.channel_type, row.account_id, row.group_id)
-
             if agent_config_id is not None:
                 row.agent_config_id = agent_config_id
             if is_enabled is not None:
@@ -126,19 +89,10 @@ class PgBindingRepository:
             row.updated_at = utc_now_naive()
 
             await db.flush()
-
-            await self.invalidate_cache(
-                channel_type=cache_keys[0],
-                account_id=cache_keys[1],
-                group_id=cache_keys[2],
-            )
-
             await db.refresh(row)
-            binding = to_domain(row)
+            return to_domain(row)
 
-        return binding
-
-    async def delete_binding(self, binding_id: int, *, updated_by: str | None = None) -> bool:
+    async def delete_binding(self, binding_id: int, *, updated_by: str | None = None) -> ChannelBinding | None:
         async with self._session_factory() as db:
             stmt = select(ChannelBindingModel).where(
                 ChannelBindingModel.id == binding_id,
@@ -147,21 +101,14 @@ class PgBindingRepository:
             result = await db.execute(stmt)
             row = result.scalar_one_or_none()
             if not row:
-                return False
+                return None
 
             row.is_deleted = 1
             row.deleted_at = utc_now_naive()
             row.updated_by = updated_by
 
             await db.flush()
-
-            await self.invalidate_cache(
-                channel_type=row.channel_type,
-                account_id=row.account_id,
-                group_id=row.group_id,
-            )
-
-        return True
+            return to_domain(row)
 
     async def get_binding(self, binding_id: int) -> ChannelBinding | None:
         async with self._session_factory() as db:
@@ -190,11 +137,3 @@ class PgBindingRepository:
             stmt = base.order_by(ChannelBindingModel.id.asc()).offset(offset).limit(limit)
             result = await db.execute(stmt)
             return [to_domain(row) for row in result.scalars().all()], total
-
-    async def invalidate_cache(self, *, channel_type: str, account_id: str, group_id: str) -> None:
-        if self._cache:
-            try:
-                cache_key = self._cache_key(channel_type, account_id, group_id)
-                await self._cache.delete(cache_key)
-            except Exception:
-                logger.warning("Redis cache invalidation failed, cache will expire via TTL")

backend/package/yuxi/channel/infrastructure/persistence/repository/pg_message_log_repository.py

@@ -2,42 +2,23 @@ from __future__ import annotations
 
 import logging
 
-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 
-from yuxi.channel.domain.repository.message_log_repository import (
+from yuxi.channel.domain.model.message_log.message_log_data import (
     MessageLogData,
     MessageLogQuery,
 )
+from yuxi.channel.domain.repository.message_log_repository import MessageLogRepositoryPort
+from yuxi.channel.infrastructure.persistence.converter.message_log_converter import (
+    to_data,
+)
 from yuxi.storage.postgres.models_channel import ChannelMessageLog
 from yuxi.utils.datetime_utils import utc_now_naive
 
 logger = logging.getLogger(__name__)
 
-_CONTENT_SUMMARY_MAX = 500
 
-
-def _to_data(row: ChannelMessageLog) -> MessageLogData:
-    return MessageLogData(
-        id=row.id,
-        trace_id=row.trace_id,
-        message_id=row.message_id,
-        channel_type=row.channel_type,
-        conversation_id=row.conversation_id,
-        session_id=row.session_id,
-        direction=row.direction,
-        sender_id=row.sender_id,
-        content_summary=row.content_summary,
-        agent_config_id=row.agent_config_id,
-        status=row.status,
-        pipeline_result=row.pipeline_result,
-        abort_reason=row.abort_reason,
-        worker_result=row.worker_result,
-        error_message=row.error_message,
-        processing_time_ms=row.processing_time_ms,
-    )
-
-
-class PgMessageLogRepository:
+class PgMessageLogRepository(MessageLogRepositoryPort):
     def __init__(self, session_factory):
         self._session_factory = session_factory
 
@@ -53,10 +34,8 @@ class PgMessageLogRepository:
         agent_config_id: int | None = None,
         session_id: str | None = None,
         conversation_id: int | None = None,
+        status: str = "received",
     ) -> MessageLogData:
-        if content_summary and len(content_summary) > _CONTENT_SUMMARY_MAX:
-            content_summary = content_summary[:_CONTENT_SUMMARY_MAX]
-
         async with self._session_factory() as db:
             row = ChannelMessageLog(
                 trace_id=trace_id,
@@ -68,12 +47,12 @@ class PgMessageLogRepository:
                 agent_config_id=agent_config_id,
                 session_id=session_id,
                 conversation_id=conversation_id,
-                status="received",
+                status=status,
             )
             db.add(row)
             await db.flush()
             await db.refresh(row)
-            return _to_data(row)
+            return to_data(row)
 
     async def update_pipeline_result(
         self,
@@ -81,6 +60,7 @@ class PgMessageLogRepository:
         trace_id: str,
         message_id: str,
         pipeline_result: str,
+        status: str,
         abort_reason: str | None = None,
     ) -> bool:
         async with self._session_factory() as db:
@@ -94,7 +74,7 @@ class PgMessageLogRepository:
                 .values(
                     pipeline_result=pipeline_result,
                     abort_reason=abort_reason,
-                    status="processing" if pipeline_result == "accepted" else "completed",
+                    status=status,
                     updated_at=utc_now_naive(),
                 )
             )
@@ -111,7 +91,7 @@ class PgMessageLogRepository:
         error_message: str | None = None,
         processing_time_ms: int | None = None,
         agent_config_id: int | None = None,
-        session_id: int | None = None,
+        session_id: str | None = None,
         conversation_id: int | None = None,
     ) -> bool:
         values: dict = {
@@ -154,7 +134,7 @@ class PgMessageLogRepository:
                 .order_by(ChannelMessageLog.created_at.asc())
             )
             result = await db.execute(stmt)
-            return [_to_data(row) for row in result.scalars().all()]
+            return [to_data(row) for row in result.scalars().all()]
 
     async def get_by_message_id(self, message_id: str) -> list[MessageLogData]:
         async with self._session_factory() as db:
@@ -167,23 +147,28 @@ class PgMessageLogRepository:
                 .order_by(ChannelMessageLog.created_at.asc())
             )
             result = await db.execute(stmt)
-            return [_to_data(row) for row in result.scalars().all()]
+            return [to_data(row) for row in result.scalars().all()]
 
-    async def query_logs(self, query: MessageLogQuery) -> list[MessageLogData]:
+    async def query_logs(self, query: MessageLogQuery) -> tuple[list[MessageLogData], int]:
         async with self._session_factory() as db:
-            stmt = select(ChannelMessageLog).where(ChannelMessageLog.is_deleted == 0)
+            base = select(ChannelMessageLog).where(ChannelMessageLog.is_deleted == 0)
             if query.channel_type:
-                stmt = stmt.where(ChannelMessageLog.channel_type == query.channel_type)
+                base = base.where(ChannelMessageLog.channel_type == query.channel_type)
             if query.conversation_id:
-                stmt = stmt.where(ChannelMessageLog.conversation_id == query.conversation_id)
+                base = base.where(ChannelMessageLog.conversation_id == query.conversation_id)
             if query.status:
-                stmt = stmt.where(ChannelMessageLog.status == query.status)
+                base = base.where(ChannelMessageLog.status == query.status)
             if query.direction:
-                stmt = stmt.where(ChannelMessageLog.direction == query.direction)
+                base = base.where(ChannelMessageLog.direction == query.direction)
             if query.pipeline_result:
-                stmt = stmt.where(ChannelMessageLog.pipeline_result == query.pipeline_result)
+                base = base.where(ChannelMessageLog.pipeline_result == query.pipeline_result)
             if query.worker_result:
-                stmt = stmt.where(ChannelMessageLog.worker_result == query.worker_result)
-            stmt = stmt.order_by(ChannelMessageLog.created_at.desc()).offset(query.offset).limit(query.limit)
+                base = base.where(ChannelMessageLog.worker_result == query.worker_result)
+
+            count_stmt = select(func.count()).select_from(base.subquery())
+            count_result = await db.execute(count_stmt)
+            total = count_result.scalar() or 0
+
+            stmt = base.order_by(ChannelMessageLog.created_at.desc()).offset(query.offset).limit(query.limit)
             result = await db.execute(stmt)
-            return [_to_data(row) for row in result.scalars().all()]
+            return [to_data(row) for row in result.scalars().all()], total

backend/package/yuxi/channel/infrastructure/persistence/repository/pg_message_repository.py

@@ -2,18 +2,30 @@ from __future__ import annotations
 
 import logging
 
-from yuxi.channel.domain.repository.message_repository import (
-    ChannelMessageData,
-)
-from yuxi.repositories.conversation_repository import ConversationRepository
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError as SaIntegrityError
+
+from yuxi.channel.domain.exception.session_not_found_error import SessionNotFoundError
+from yuxi.channel.domain.model.message.channel_message_data import ChannelMessageData
+from yuxi.channel.domain.repository.message_repository import MessageRepositoryPort
+from yuxi.channel.infrastructure.persistence.exception_translator import translate_db_exception
+from yuxi.storage.postgres.models_business import Conversation, Message
 
 logger = logging.getLogger(__name__)
 
 
-class PgMessageRepository:
+class PgMessageRepository(MessageRepositoryPort):
     def __init__(self, session_factory):
         self._session_factory = session_factory
 
+    async def _get_conversation_id(self, db, thread_id: str) -> int:
+        stmt = select(Conversation.id).where(Conversation.thread_id == thread_id)
+        result = await db.execute(stmt)
+        conv_id = result.scalar_one_or_none()
+        if conv_id is None:
+            raise SessionNotFoundError(thread_id=thread_id)
+        return conv_id
+
     async def save_user_message(
         self,
         *,
@@ -22,21 +34,28 @@ class PgMessageRepository:
         message_id: str | None = None,
         extra_metadata: dict | None = None,
     ) -> ChannelMessageData:
-        meta = extra_metadata or {}
+        meta = {**(extra_metadata or {})}
         if message_id:
             meta["channel_message_id"] = message_id
 
-        async with self._session_factory() as db:
-            conv_repo = ConversationRepository(db)
-            msg = await conv_repo.add_message_by_thread_id(
-                thread_id=thread_id,
-                role="user",
-                content=content,
-                extra_metadata=meta,
-            )
+        try:
+            async with self._session_factory() as db:
+                conversation_id = await self._get_conversation_id(db, thread_id)
+                row = Message(
+                    conversation_id=conversation_id,
+                    role="user",
+                    content=content,
+                    extra_metadata=meta,
+                )
+                db.add(row)
+                await db.flush()
+                await db.refresh(row)
+                msg_id = row.id
+        except SaIntegrityError as exc:
+            raise translate_db_exception(exc) from exc
 
         return ChannelMessageData(
-            id=msg.id,
+            id=msg_id,
             thread_id=thread_id,
             role="user",
             content=content,
@@ -51,17 +70,24 @@ class PgMessageRepository:
         content: str,
         extra_metadata: dict | None = None,
     ) -> ChannelMessageData:
-        async with self._session_factory() as db:
-            conv_repo = ConversationRepository(db)
-            msg = await conv_repo.add_message_by_thread_id(
-                thread_id=thread_id,
-                role="assistant",
-                content=content,
-                extra_metadata=extra_metadata,
-            )
+        try:
+            async with self._session_factory() as db:
+                conversation_id = await self._get_conversation_id(db, thread_id)
+                row = Message(
+                    conversation_id=conversation_id,
+                    role="assistant",
+                    content=content,
+                    extra_metadata=extra_metadata,
+                )
+                db.add(row)
+                await db.flush()
+                await db.refresh(row)
+                msg_id = row.id
+        except SaIntegrityError as exc:
+            raise translate_db_exception(exc) from exc
 
         return ChannelMessageData(
-            id=msg.id,
+            id=msg_id,
             thread_id=thread_id,
             role="assistant",
             content=content,

backend/package/yuxi/channel/infrastructure/persistence/repository/pg_outbox_repository.py

@@ -4,10 +4,15 @@ import json
 import logging
 from datetime import timedelta
 
-import redis.asyncio as aioredis
-from sqlalchemy import select
+from sqlalchemy import select, update
 
 from yuxi.channel.domain.model.outbox.outbox_entry import OutboxEntry
+from yuxi.channel.domain.port import CachePort
+from yuxi.channel.domain.port.internal.unit_of_work import UnitOfWork
+from yuxi.channel.domain.repository.outbox_repository import OutboxRepositoryPort
+from yuxi.channel.infrastructure.persistence.converter.outbox_converter import (
+    to_domain,
+)
 from yuxi.storage.postgres.models_channel import ChannelOutbox
 from yuxi.utils.datetime_utils import utc_now_naive
 
@@ -16,23 +21,7 @@ logger = logging.getLogger(__name__)
 _BASE_DELAY = 5
 _MAX_DELAY = 300
 _OUTBOX_NOTIFY_CHANNEL = "channel:outbox:notify"
-
-
-def _to_domain(row: ChannelOutbox) -> OutboxEntry:
-    return OutboxEntry(
-        id=row.id,
-        message_id=row.message_id,
-        session_id=row.session_id,
-        channel_type=row.channel_type,
-        content=row.content,
-        status=row.status,
-        retry_count=row.retry_count,
-        max_retries=row.max_retries,
-        next_retry_at=row.next_retry_at.isoformat() if row.next_retry_at else None,
-        last_error=row.last_error,
-        trace_id=row.trace_id,
-        extra_metadata=row.extra_metadata,
-    )
+_STALE_PROCESSING_MINUTES = 5
 
 
 def _calc_next_retry(retry_count: int):
@@ -40,14 +29,14 @@ def _calc_next_retry(retry_count: int):
     return utc_now_naive() + timedelta(seconds=delay)
 
 
-class PgOutboxRepository:
+class PgOutboxRepository(OutboxRepositoryPort):
     def __init__(
         self,
         session_factory,
-        redis: aioredis.Redis | None = None,
+        cache_port: CachePort | None = None,
     ):
         self._session_factory = session_factory
-        self._redis = redis
+        self._cache = cache_port
 
     async def enqueue(
         self,
@@ -58,6 +47,7 @@ class PgOutboxRepository:
         content: str,
         trace_id: str | None = None,
         extra_metadata: dict | None = None,
+        uow: UnitOfWork | None = None,
     ) -> OutboxEntry:
         async with self._session_factory() as db:
             row = ChannelOutbox(
@@ -72,11 +62,11 @@ class PgOutboxRepository:
             db.add(row)
             await db.flush()
             await db.refresh(row)
-            entry = _to_domain(row)
+            entry = to_domain(row)
 
-        if self._redis:
+        if self._cache:
             try:
-                await self._redis.publish(
+                await self._cache.publish(
                     _OUTBOX_NOTIFY_CHANNEL,
                     json.dumps({"entry_id": entry.id}),
                 )
@@ -85,9 +75,21 @@ class PgOutboxRepository:
 
         return entry
 
-    async def fetch_pending(self, *, limit: int = 20) -> list[OutboxEntry]:
+    async def fetch_and_mark_pending(self, *, limit: int = 20) -> list[OutboxEntry]:
         now = utc_now_naive()
+        stale_threshold = now - timedelta(minutes=_STALE_PROCESSING_MINUTES)
+
         async with self._session_factory() as db:
+            await db.execute(
+                update(ChannelOutbox)
+                .where(
+                    ChannelOutbox.status == "processing",
+                    ChannelOutbox.updated_at < stale_threshold,
+                    ChannelOutbox.is_deleted == 0,
+                )
+                .values(status="pending", updated_at=now)
+            )
+
             stmt = (
                 select(ChannelOutbox)
                 .where(
@@ -100,7 +102,21 @@ class PgOutboxRepository:
                 .with_for_update(skip_locked=True)
             )
             result = await db.execute(stmt)
-            return [_to_domain(row) for row in result.scalars().all()]
+            rows = result.scalars().all()
+            if not rows:
+                return []
+
+            entry_ids = [row.id for row in rows]
+            await db.execute(
+                update(ChannelOutbox)
+                .where(ChannelOutbox.id.in_(entry_ids))
+                .values(status="processing", updated_at=utc_now_naive())
+            )
+            await db.flush()
+            return [to_domain(row) for row in rows]
+
+    async def fetch_pending(self, *, limit: int = 20) -> list[OutboxEntry]:
+        return await self.fetch_and_mark_pending(limit=limit)
 
     async def get_by_id(self, entry_id: int) -> OutboxEntry | None:
         async with self._session_factory() as db:
@@ -110,80 +126,90 @@ class PgOutboxRepository:
             )
             result = await db.execute(stmt)
             row = result.scalar_one_or_none()
-            return _to_domain(row) if row else None
+            return to_domain(row) if row else None
 
     async def mark_retrying(self, entry_id: int, *, last_error: str | None = None) -> OutboxEntry | None:
         async with self._session_factory() as db:
-            stmt = select(ChannelOutbox).where(
-                ChannelOutbox.id == entry_id,
-                ChannelOutbox.is_deleted == 0,
-            )
-            result = await db.execute(stmt)
-            row = result.scalar_one_or_none()
+            check_stmt = select(
+                ChannelOutbox.retry_count,
+                ChannelOutbox.max_retries,
+            ).where(ChannelOutbox.id == entry_id, ChannelOutbox.is_deleted == 0)
+            result = await db.execute(check_stmt)
+            row = result.one_or_none()
             if not row:
                 return None
 
-            row.retry_count += 1
-            row.status = "retrying"
-            row.next_retry_at = _calc_next_retry(row.retry_count)
-            if last_error:
-                row.last_error = last_error
-            await db.flush()
-            await db.refresh(row)
-            return _to_domain(row)
+            current_count, max_retries = row
+            new_count = current_count + 1
+
+            if new_count >= max_retries:
+                stmt = (
+                    update(ChannelOutbox)
+                    .where(ChannelOutbox.id == entry_id, ChannelOutbox.is_deleted == 0)
+                    .values(
+                        status="dead",
+                        last_error=last_error or "max_retries_exceeded",
+                        updated_at=utc_now_naive(),
+                    )
+                    .returning(ChannelOutbox)
+                )
+            else:
+                next_retry = _calc_next_retry(new_count)
+                values: dict = {
+                    "retry_count": new_count,
+                    "status": "retrying",
+                    "next_retry_at": next_retry,
+                    "updated_at": utc_now_naive(),
+                }
+                if last_error:
+                    values["last_error"] = last_error
+                stmt = (
+                    update(ChannelOutbox)
+                    .where(ChannelOutbox.id == entry_id, ChannelOutbox.is_deleted == 0)
+                    .values(**values)
+                    .returning(ChannelOutbox)
+                )
+
+            result = await db.execute(stmt)
+            updated = result.scalar_one_or_none()
+            return to_domain(updated) if updated else None
 
     async def mark_sent(self, entry_id: int) -> bool:
         async with self._session_factory() as db:
-            stmt = select(ChannelOutbox).where(
-                ChannelOutbox.id == entry_id,
-                ChannelOutbox.is_deleted == 0,
+            stmt = (
+                update(ChannelOutbox)
+                .where(ChannelOutbox.id == entry_id, ChannelOutbox.is_deleted == 0)
+                .values(status="sent", updated_at=utc_now_naive())
             )
             result = await db.execute(stmt)
-            row = result.scalar_one_or_none()
-            if not row:
-                return False
-
-            row.status = "sent"
-            await db.flush()
-            return True
+            return result.rowcount > 0
 
     async def mark_dead(self, entry_id: int, *, last_error: str) -> bool:
         async with self._session_factory() as db:
-            stmt = select(ChannelOutbox).where(
-                ChannelOutbox.id == entry_id,
-                ChannelOutbox.is_deleted == 0,
+            stmt = (
+                update(ChannelOutbox)
+                .where(ChannelOutbox.id == entry_id, ChannelOutbox.is_deleted == 0)
+                .values(status="dead", last_error=last_error, updated_at=utc_now_naive())
             )
             result = await db.execute(stmt)
-            row = result.scalar_one_or_none()
-            if not row:
-                return False
-
-            row.status = "dead"
-            row.last_error = last_error
-            await db.flush()
-            return True
+            return result.rowcount > 0
 
     async def mark_failed(self, entry_id: int, *, last_error: str) -> bool:
         async with self._session_factory() as db:
-            stmt = select(ChannelOutbox).where(
-                ChannelOutbox.id == entry_id,
-                ChannelOutbox.is_deleted == 0,
+            stmt = (
+                update(ChannelOutbox)
+                .where(ChannelOutbox.id == entry_id, ChannelOutbox.is_deleted == 0)
+                .values(status="failed", last_error=last_error, updated_at=utc_now_naive())
             )
             result = await db.execute(stmt)
-            row = result.scalar_one_or_none()
-            if not row:
-                return False
-
-            row.status = "failed"
-            row.last_error = last_error
-            await db.flush()
-            return True
+            return result.rowcount > 0
 
     async def list_outbox(
         self,
         *,
         channel_type: str | None = None,
         status: str | None = None,
+        offset: int = 0,
         limit: int = 50,
     ) -> list[OutboxEntry]:
         async with self._session_factory() as db:
@@ -192,6 +218,6 @@ class PgOutboxRepository:
                 stmt = stmt.where(ChannelOutbox.channel_type == channel_type)
             if status:
                 stmt = stmt.where(ChannelOutbox.status == status)
-            stmt = stmt.order_by(ChannelOutbox.created_at.desc()).limit(limit)
+            stmt = stmt.order_by(ChannelOutbox.created_at.desc()).offset(offset).limit(limit)
             result = await db.execute(stmt)
-            return [_to_domain(row) for row in result.scalars().all()]
+            return [to_domain(row) for row in result.scalars().all()]

backend/package/yuxi/channel/infrastructure/persistence/repository/pg_session_repository.py

@@ -3,45 +3,41 @@ from __future__ import annotations
 import asyncio
 import logging
 import uuid as uuid_lib
-from collections.abc import Awaitable, Callable
 
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
 from yuxi.channel.domain.model.session.channel_session import ChannelSession
-from yuxi.channel.domain.port.cache_port import CachePort
+from yuxi.channel.domain.port import CachePort
+from yuxi.channel.domain.repository.session_repository import SessionRepositoryPort
+from yuxi.channel.infrastructure.persistence.converter.session_converter import (
+    to_domain,
+)
 from yuxi.storage.postgres.models_business import Conversation
 
 logger = logging.getLogger(__name__)
 
 _LOCK_TIMEOUT = 10
 
-
-def _to_domain(row: Conversation) -> ChannelSession:
-    return ChannelSession(
-        id=row.id,
-        thread_id=row.thread_id,
-        user_id=row.user_id,
-        agent_id=row.agent_id,
-        channel_type=row.channel_type,
-        channel_session_key=row.channel_session_key,
-        status=row.status,
-        title=row.title,
-    )
+_LOCK_RELEASE_SCRIPT = """
+if redis.call("get", KEYS[1]) == ARGV[1] then
+    return redis.call("del", KEYS[1])
+else
+    return 0
+end
+"""
 
 
-class PgSessionRepository:
+class PgSessionRepository(SessionRepositoryPort):
     def __init__(
         self,
         session_factory,
         cache_port: CachePort | None = None,
         *,
-        agent_id_resolver: Callable[[int], Awaitable[str]],
         lock_timeout: int = _LOCK_TIMEOUT,
     ):
         self._session_factory = session_factory
         self._cache = cache_port
-        self._agent_id_resolver = agent_id_resolver
         self._lock_timeout = lock_timeout
 
     async def get_or_create(
@@ -49,27 +45,28 @@ class PgSessionRepository:
         *,
         channel_type: str,
         account_id: str,
-        agent_config_id: int,
+        agent_id: str,
     ) -> ChannelSession:
-        agent_id = await self._agent_id_resolver(agent_config_id)
         session_key = account_id
 
         existing = await self._find_existing(channel_type, session_key, agent_id)
         if existing:
-            return _to_domain(existing)
+            return existing
 
         lock_key = f"channel:session_lock:{session_key}:{agent_id}"
+        lock_value = str(uuid_lib.uuid4())
         lock_acquired = False
 
         if self._cache:
             try:
-                lock_acquired = await self._cache.set(lock_key, "1", ex=self._lock_timeout, nx=True)
+                lock_acquired = await self._cache.set(lock_key, lock_value, ex=self._lock_timeout, nx=True)
                 if not lock_acquired:
                     await asyncio.sleep(0.1)
                     existing = await self._find_existing(channel_type, session_key, agent_id)
                     if existing:
-                        return _to_domain(existing)
-                    lock_acquired = await self._cache.set(lock_key, "1", ex=self._lock_timeout, nx=True)
+                        return existing
+                    lock_value = str(uuid_lib.uuid4())
+                    lock_acquired = await self._cache.set(lock_key, lock_value, ex=self._lock_timeout, nx=True)
             except Exception:
                 logger.warning("Redis lock failed, proceeding without lock")
 
@@ -81,26 +78,27 @@ class PgSessionRepository:
                     agent_id=agent_id,
                     channel_type=channel_type,
                     channel_session_key=session_key,
-                    extra_metadata={
-                        "source": "channel",
-                        "agent_config_id": agent_config_id,
-                    },
+                    extra_metadata={"source": "channel"},
                 )
                 db.add(conv)
                 await db.flush()
                 await db.refresh(conv)
-                return _to_domain(conv)
+                return to_domain(conv)
         except IntegrityError:
             existing = await self._find_existing(channel_type, session_key, agent_id)
             if existing:
-                return _to_domain(existing)
+                return existing
             raise
         finally:
             if self._cache and lock_acquired:
                 try:
-                    await self._cache.delete(lock_key)
+                    await self._cache.eval(
+                        _LOCK_RELEASE_SCRIPT,
+                        keys=[lock_key],
+                        args=[lock_value],
+                    )
                 except Exception:
-                    pass
+                    logger.warning("Lock release failed for %s", lock_key)
 
     async def get_by_thread_id(self, thread_id: str) -> ChannelSession | None:
         async with self._session_factory() as db:
@@ -110,9 +108,9 @@ class PgSessionRepository:
             )
             result = await db.execute(stmt)
             row = result.scalar_one_or_none()
-            return _to_domain(row) if row else None
+            return to_domain(row) if row else None
 
-    async def _find_existing(self, channel_type: str, session_key: str, agent_id: str) -> Conversation | None:
+    async def _find_existing(self, channel_type: str, session_key: str, agent_id: str) -> ChannelSession | None:
         async with self._session_factory() as db:
             stmt = select(Conversation).where(
                 Conversation.channel_type == channel_type,
@@ -121,7 +119,8 @@ class PgSessionRepository:
                 Conversation.status != "deleted",
             )
             result = await db.execute(stmt)
-            return result.scalar_one_or_none()
+            row = result.scalar_one_or_none()
+            return to_domain(row) if row else None
 
     async def count_active_sessions(self, channel_type: str) -> int:
         from sqlalchemy import func

backend/package/yuxi/channel/infrastructure/persistence/uow.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+class SqlAlchemyUnitOfWork:
+    def __init__(self, session: AsyncSession):
+        self._session = session
+
+    async def commit(self) -> None:
+        await self._session.commit()
+
+    async def rollback(self) -> None:
+        await self._session.rollback()
+
+    def session(self) -> AsyncSession:
+        return self._session

backend/package/yuxi/channel/interfaces/rest/router/bindings.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import logging
+from datetime import datetime
 
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field
 
 from yuxi.channel.container import ChannelContainer, get_channel
+from yuxi.channel.domain.exception.duplicate_binding import DuplicateBindingException
+from yuxi.channel.domain.exception.invalid_agent_config import InvalidAgentConfigException
 from yuxi.channel.interfaces.rest.auth.depends import channel_auth_depends
 
 logger = logging.getLogger(__name__)
@@ -20,6 +23,11 @@ class BindingCreateRequest(BaseModel):
     agent_config_id: int = Field(..., gt=0)
 
 
+class BindingUpdateRequest(BaseModel):
+    agent_config_id: int | None = None
+    is_enabled: bool | None = None
+
+
 class BindingResponse(BaseModel):
     id: int
     channel_type: str
@@ -27,7 +35,7 @@ class BindingResponse(BaseModel):
     group_id: str
     agent_config_id: int
     is_enabled: bool
-    created_at: str
+    created_at: datetime
 
 
 class PaginatedMeta(BaseModel):
@@ -47,12 +55,17 @@ async def create_binding(
     channel: ChannelContainer = Depends(get_channel),
     _: bool = Depends(channel_auth_depends),
 ):
-    created = await channel.binding_service.create(
-        channel_type=req.channel_type,
-        account_id=req.account_id,
-        group_id=req.group_id,
-        agent_config_id=req.agent_config_id,
-    )
+    try:
+        created = await channel.binding_service.create(
+            channel_type=req.channel_type,
+            account_id=req.account_id,
+            group_id=req.group_id,
+            agent_config_id=req.agent_config_id,
+        )
+    except DuplicateBindingException as e:
+        raise HTTPException(status_code=409, detail=str(e))
+    except InvalidAgentConfigException as e:
+        raise HTTPException(status_code=400, detail=str(e))
 
     return BindingResponse(
         id=created.id,
@@ -68,8 +81,8 @@ async def create_binding(
 @router.get("/channel/bindings", response_model=BindingListResponse)
 async def list_bindings(
     channel_type: str | None = None,
-    offset: int = Query(0, ge=0),
-    limit: int = Query(50, ge=1, le=200),
+    offset: int = 0,
+    limit: int = 50,
     channel: ChannelContainer = Depends(get_channel),
     _: bool = Depends(channel_auth_depends),
 ):
@@ -95,6 +108,31 @@ async def list_bindings(
     )
 
 
+@router.patch("/channel/bindings/{binding_id}", response_model=BindingResponse)
+async def update_binding(
+    binding_id: int,
+    req: BindingUpdateRequest,
+    channel: ChannelContainer = Depends(get_channel),
+    _: bool = Depends(channel_auth_depends),
+):
+    if req.agent_config_id is None and req.is_enabled is None:
+        raise HTTPException(status_code=400, detail="no fields to update")
+    updated = await channel.binding_service.update(
+        binding_id, agent_config_id=req.agent_config_id, is_enabled=req.is_enabled
+    )
+    if not updated:
+        raise HTTPException(status_code=404, detail="binding not found")
+    return BindingResponse(
+        id=updated.id,
+        channel_type=updated.channel_type,
+        account_id=updated.account_id,
+        group_id=updated.group_id,
+        agent_config_id=updated.agent_config_id,
+        is_enabled=updated.is_enabled,
+        created_at=updated.created_at,
+    )
+
+
 @router.delete("/channel/bindings/{binding_id}")
 async def delete_binding(
     binding_id: int,

backend/package/yuxi/channel/interfaces/rest/router/health.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import os
 
 from fastapi import APIRouter, Depends
 from fastapi.responses import JSONResponse
@@ -36,24 +37,34 @@ class ReadyzResponse(BaseModel):
 
 @router.get("/healthz", response_model=HealthzResponse)
 async def liveness_check(channel: ChannelContainer = Depends(get_channel)):
+    show_timeline = os.getenv("CHANNEL_SHOW_STARTUP_TIMELINE", "false").lower() == "true"
     return HealthzResponse(
         status="ok",
-        startup_timeline=channel.startup_tracer.to_dict() if channel else None,
+        startup_timeline=channel.startup_tracer.to_dict() if show_timeline else None,
     )
 
 
 @router.get("/readyz")
 async def readiness_check(channel: ChannelContainer = Depends(get_channel)):
-    checks = {
-        "redis": await _check_redis(channel),
-        "postgres": await _check_postgres(),
-        "workers": await _check_workers(channel),
-        "channels": await _check_channels(channel),
-        "ws_connections": await _check_ws_connections(channel),
-    }
+    keys = ["redis", "postgres", "workers", "channels", "ws_connections"]
+    results = await asyncio.gather(
+        _check_redis(channel),
+        _check_postgres(),
+        _check_workers(channel),
+        _check_channels(channel),
+        _check_ws_connections(channel),
+    )
+    checks = dict(zip(keys, results))
 
-    overall = "ready" if all(v.status == "ok" for v in checks.values()) else "not_ready"
-    status_code = 200 if overall == "ready" else 503
+    has_error = any(v.status == "error" for v in checks.values())
+    has_degraded = any(v.status == "degraded" for v in checks.values())
+
+    if has_error:
+        overall, status_code = "not_ready", 503
+    elif has_degraded:
+        overall, status_code = "degraded", 200
+    else:
+        overall, status_code = "ready", 200
 
     return JSONResponse(
         status_code=status_code,
@@ -75,7 +86,7 @@ async def _check_postgres() -> ReadyzComponentStatus:
     try:
         from yuxi.storage.postgres.manager import pg_manager
 
-        if not pg_manager._initialized:
+        if not pg_manager.is_initialized:
             return ReadyzComponentStatus(status="error", detail="not initialized")
         async with pg_manager.get_async_session_context() as session:
             await session.execute(select(1))

backend/package/yuxi/channel/interfaces/rest/router/metrics.py

@@ -7,6 +7,7 @@ from fastapi.responses import Response
 from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
 
 from yuxi.channel.container import ChannelContainer, get_channel
+from yuxi.channel.interfaces.rest.auth.depends import channel_auth_depends
 
 logger = logging.getLogger(__name__)
 
@@ -14,8 +15,11 @@ router = APIRouter()
 
 
 @router.get("/metrics")
-async def metrics(channel: ChannelContainer = Depends(get_channel)):
-    if channel and channel.sse_endpoint and channel.metrics:
+async def metrics(
+    channel: ChannelContainer = Depends(get_channel),
+    _: bool = Depends(channel_auth_depends),
+):
+    if channel.sse_endpoint and channel.metrics:
         await channel.metrics.set_sse_connections(channel.sse_endpoint.connection_count)
 
     data = generate_latest()

backend/package/yuxi/channel/interfaces/rest/router/registry.py

@@ -5,7 +5,7 @@ import logging
 from fastapi import FastAPI
 
 from yuxi.channel.domain.port.channel_adapter_port import ChannelAdapterPort
-from yuxi.channel.domain.port.channel_route_contributor import ChannelRouteContributor
+from yuxi.channel.domain.port.channel_route_contributor_port import ChannelRouteContributorPort
 
 logger = logging.getLogger(__name__)
 
@@ -30,11 +30,11 @@ def register_all_channel_routes(
     return registered
 
 
-def _get_route_contributor(adapter: ChannelAdapterPort) -> ChannelRouteContributor | None:
+def _get_route_contributor(adapter: ChannelAdapterPort) -> ChannelRouteContributorPort | None:
     contributor = getattr(adapter, "route_contributor", None)
     if contributor is None:
         return None
-    if isinstance(contributor, ChannelRouteContributor):
+    if isinstance(contributor, ChannelRouteContributorPort):
         return contributor
     if hasattr(contributor, "router"):
         return contributor