refactor(channel-router): 清理冗余空行并优化代码结构

本次提交包含多类优化：
1.  移除多个路由文件中多余的空导入行，统一代码格式
2.  重构Query参数定义，将长参数拆分为多行提升可读性
3.  新增多个业务端点：
    - 渠道能力画像矩阵查询CAP-03
    - 配对审批计数接口用于待办角标
    - 批量查询对端目录资料接口
    - 向导扫码登录相关端点
    - 会话实时事件SSE推送端点
    - 工作台待办统计接口
4.  完善异常处理逻辑，补充OperationTimeoutError导入并优化NotImplementedError的细节返回
5.  调整路由导入顺序，修复动态路由路径冲突隐患
6.  更新文档注释与接口清单，修正部分接口描述细节

backend/server/routers/channels/__init__.py

@@ -17,7 +17,6 @@ from typing import Any, Literal
 
 from fastapi import APIRouter, Depends, Request
 from sqlalchemy.ext.asyncio import AsyncSession
-
 from yuxi.channels.contract.dtos.common import Operator, OperatorRole
 from yuxi.channels.contract.dtos.control import ControlResult
 from yuxi.channels.contract.errors import (
@@ -35,6 +34,7 @@ from yuxi.channels.contract.errors import (
     LifecycleHookError,
     NotFoundError,
     NotImplementedError,
+    OperationTimeoutError,
     PairingExpiredError,
     PermissionDeniedError,
     PipelineConfigError,
@@ -44,7 +44,6 @@ from yuxi.channels.contract.errors import (
     RateLimitError,
     RuleViolationError,
     SchemaInitializationError,
-    OperationTimeoutError,
     ValidationError,
 )
 from yuxi.channels.infrastructure import DependencyInjectionContainer
@@ -351,7 +350,11 @@ def raiseOnControlFailure(result: ControlResult) -> None:
             message=message,
         )
     if code == "NOT_IMPLEMENTED":
-        raise NotImplementedError(operation=details.get("operation", ""), message=message)
+        raise NotImplementedError(
+            operation=details.get("operation", ""),
+            message=message,
+            details={k: v for k, v in details.items() if k != "operation"},
+        )
     raise InternalError(message=message)
 
 
@@ -406,23 +409,23 @@ channels_router = APIRouter(prefix="/channels", tags=["channels"])
 # 注册顺序：静态路径 router（如 health_router）必须先于含 /{channel_type}/...
 # 动态路径的 router 注册，避免 /channels/health 被捕获为 channel_type="health"
 # （规范 §6.5）。
-from server.routers.channels.health_router import health_router
+from server.routers.channels.health_router import health_router  # noqa: E402
 
 channels_router.include_router(health_router)
 
-from server.routers.channels.audit_router import audit_router
+from server.routers.channels.audit_router import audit_router  # noqa: E402
 
 channels_router.include_router(audit_router)
 
-from server.routers.channels.analytics_router import analytics_router
+from server.routers.channels.analytics_router import analytics_router  # noqa: E402
 
 channels_router.include_router(analytics_router)
 
-from server.routers.channels.account_router import account_router
+from server.routers.channels.account_router import account_router  # noqa: E402
-from server.routers.channels.session_router import session_router
+from server.routers.channels.directory_router import directory_router  # noqa: E402
-from server.routers.channels.login_router import login_router
+from server.routers.channels.login_router import login_router  # noqa: E402
-from server.routers.channels.directory_router import directory_router
+from server.routers.channels.pairing_router import pairing_router  # noqa: E402
-from server.routers.channels.pairing_router import pairing_router
+from server.routers.channels.session_router import session_router  # noqa: E402
 
 channels_router.include_router(account_router)
 channels_router.include_router(session_router)
@@ -430,46 +433,46 @@ channels_router.include_router(login_router)
 channels_router.include_router(directory_router)
 channels_router.include_router(pairing_router)
 
-from server.routers.channels.allowlist_router import allowlist_router
+from server.routers.channels.allowlist_router import allowlist_router  # noqa: E402
 
 channels_router.include_router(allowlist_router)
 
-from server.routers.channels.message_router import message_router
+from server.routers.channels.message_router import message_router  # noqa: E402
 
 channels_router.include_router(message_router)
 
-from server.routers.channels.outbox_router import outbox_router
+from server.routers.channels.config_router import config_router  # noqa: E402
-from server.routers.channels.webhook_router import webhook_router
+from server.routers.channels.outbox_router import outbox_router  # noqa: E402
-from server.routers.channels.config_router import config_router
+from server.routers.channels.webhook_router import webhook_router  # noqa: E402
 
 channels_router.include_router(outbox_router)
 channels_router.include_router(webhook_router)
 channels_router.include_router(config_router)
 
-from server.routers.channels.capability_router import capability_router
+from server.routers.channels.capability_router import capability_router  # noqa: E402
 
 channels_router.include_router(capability_router)
 
-from server.routers.channels.doctor_router import doctor_router
+from server.routers.channels.doctor_router import doctor_router  # noqa: E402
 
 channels_router.include_router(doctor_router)
 
-from server.routers.channels.plugin_router import plugin_router
+from server.routers.channels.plugin_router import plugin_router  # noqa: E402
 
 channels_router.include_router(plugin_router)
 
-from server.routers.channels.content_review_router import content_review_router
+from server.routers.channels.content_review_router import content_review_router  # noqa: E402
 
 channels_router.include_router(content_review_router)
 
-from server.routers.channels.dashboard_router import dashboard_router
+from server.routers.channels.dashboard_router import dashboard_router  # noqa: E402
 
 channels_router.include_router(dashboard_router)
 
-from server.routers.channels.reports_router import reports_router
+from server.routers.channels.wizard_router import wizard_router  # noqa: E402
-
-channels_router.include_router(reports_router)
-
-from server.routers.channels.wizard_router import wizard_router
 
 channels_router.include_router(wizard_router)
+
+from server.routers.channels.route_binding_router import route_binding_router  # noqa: E402
+
+channels_router.include_router(route_binding_router)

backend/server/routers/channels/account_router.py

@@ -38,7 +38,6 @@ from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field, model_validator
-
 from yuxi.channels.contract.dtos.channel import (
     BatchStateChangeCmd,
     ChannelType,

backend/server/routers/channels/allowlist_router.py

@@ -36,7 +36,6 @@ from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.whitelist import (
     UNSET,
@@ -313,9 +312,7 @@ async def export_whitelist(
     account_id: str,
     policy_type: str,
     request: Request,
-    export_format: str = Query(
+    export_format: str = Query(default="json", alias="format", description="导出格式：json / csv"),
-        default="json", alias="format", description="导出格式：json / csv"
-    ),
     use_cases=Depends(get_channel_use_cases),
     current_user: User = Depends(get_admin_user),
 ) -> dict[str, Any]:

backend/server/routers/channels/audit_router.py

@@ -52,7 +52,6 @@ from zoneinfo import ZoneInfo
 from fastapi import APIRouter, Depends, Query, Request
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.audit import (
     AuditOperationType,
     AuditQuery,

backend/server/routers/channels/capability_router.py

@@ -1,4 +1,4 @@
-"""能力查询路由（CAP-01~CAP-02）。
+"""能力查询路由（CAP-01~CAP-03）。
 
 统一采用模板 A（控制面端口路由）：鉴权依赖 get_admin_user，
 调用 use_cases.capability_query.<method>，raiseOnControlFailure 转译失败，
@@ -11,7 +11,6 @@ from __future__ import annotations
 from typing import Any
 
 from fastapi import APIRouter, Depends, Request
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.storage.postgres.models_business import User
 
@@ -45,6 +44,25 @@ async def list_capabilities(
     return {"success": True, "data": serialize_control_data(result.data)}
 
 
+@capability_router.get("/capabilities/matrix", response_model=dict)
+async def list_capability_matrix(
+    request: Request,
+    use_cases=Depends(get_channel_use_cases),
+    current_user: User = Depends(get_admin_user),
+) -> dict[str, Any]:
+    """列出所有渠道能力画像矩阵（CAP-03）。对应控制面操作 capability/matrix。
+
+    返回每个插件的完整 ``ChannelCapabilityProfile``，包含静态能力声明、
+    运行时可用性、插件元数据与配置 schema，供能力查询页渲染高密度矩阵。
+    """
+    operator = build_operator(current_user, request)
+    result = await use_cases.capability_query.listCapabilityMatrix(
+        operator=operator,
+    )
+    raiseOnControlFailure(result)
+    return {"success": True, "data": serialize_control_data(result.data)}
+
+
 @capability_router.get("/capabilities/{channel_type}", response_model=dict)
 async def get_capability(
     channel_type: ChannelType,

backend/server/routers/channels/config_router.py

@@ -43,7 +43,6 @@ from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.config import (
     BatchUpdateConfigCmd,
     ConfigExportCmd,

backend/server/routers/channels/content_review_router.py

@@ -64,7 +64,6 @@ from typing import Any, Literal
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.content_review import (
     BatchReviewDecisionCmd,
@@ -141,9 +140,7 @@ class BatchReviewDecisionRequest(BaseModel):
     decisions: list[ReviewDecisionItemRequest] = Field(
         ..., min_length=1, max_length=100, description="决定条目列表（1-100）"
     )
-    apply_to_pending_messages: bool = Field(
+    apply_to_pending_messages: bool = Field(default=False, description="是否关联处理待审消息（本期预留）")
-        default=False, description="是否关联处理待审消息（本期预留）"
-    )
 
 
 def _parse_verdict(raw: str) -> ContentReviewVerdict:

backend/server/routers/channels/dashboard_router.py

@@ -47,15 +47,27 @@ dashboard_router = APIRouter(tags=["channels-dashboard"])
 @dashboard_router.get("/dashboard/overview", response_model=dict)
 async def get_dashboard_overview(
     request: Request,
+    channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
+    start_time: str | None = Query(default=None, description="消息统计起始时间（ISO 8601）；仅 messages 子域受影响"),
+    end_time: str | None = Query(default=None, description="消息统计截止时间（ISO 8601）；仅 messages 子域受影响"),
     use_cases=Depends(get_channel_use_cases),
     current_user: User = Depends(get_admin_user),
 ) -> dict[str, Any]:
     """查询全局总览（DSB-01）。对应控制面操作 dashboard/overview。
 
     聚合账户 / 消息 / 会话 / 投递四类子域 KPI 计数。
+    ``start_time`` / ``end_time`` 仅影响消息统计；账户、会话、投递为当前
+    快照，不受时间范围影响。
     """
     operator = build_operator(current_user, request)
-    result = await use_cases.dashboard_view.getDashboardOverview(operator=operator)
+    start_time_dt = parse_datetime("start_time", start_time)
+    end_time_dt = parse_datetime("end_time", end_time)
+    result = await use_cases.dashboard_view.getDashboardOverview(
+        channel_type=channel_type,
+        start_time=start_time_dt.isoformat() if start_time_dt else None,
+        end_time=end_time_dt.isoformat() if end_time_dt else None,
+        operator=operator,
+    )
     raiseOnControlFailure(result)
     return {"success": True, "data": serialize_control_data(result.data)}
 
@@ -93,7 +105,9 @@ async def get_dashboard_messages(
 
     返回指定时间范围内的消息按渠道 / 角色 / 投递状态分组的计数快照。
 
-    时间参数由 parse_datetime 在 Router 层翻译为 UTC datetime，再转回 ISO 8601 字符串传给端口（端口签名为 str）。非法格式抛 ValidationError（400）。
+    时间参数由 parse_datetime 在 Router 层翻译为 UTC datetime，再转回
+    ISO 8601 字符串传给端口（端口签名为 str）。非法格式抛
+    ValidationError（400）。
     """
     operator = build_operator(current_user, request)
     start_time_dt = parse_datetime("start_time", start_time)
@@ -158,6 +172,26 @@ async def get_dashboard_delivery(
     return {"success": True, "data": serialize_control_data(result.data)}
 
 
+@dashboard_router.get("/dashboard/todos", response_model=dict)
+async def get_dashboard_todos(
+    request: Request,
+    channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
+    use_cases=Depends(get_channel_use_cases),
+    current_user: User = Depends(get_admin_user),
+) -> dict[str, Any]:
+    """查询工作台待办计数（DSB-TODOS）。对应控制面操作 dashboard/todos。
+
+    聚合待审批配对数、待复核内容审核数与当前死信积压数，按渠道类型过滤。
+    """
+    operator = build_operator(current_user, request)
+    result = await use_cases.dashboard_view.getDashboardTodos(
+        channel_type=channel_type,
+        operator=operator,
+    )
+    raiseOnControlFailure(result)
+    return {"success": True, "data": serialize_control_data(result.data)}
+
+
 @dashboard_router.get("/dashboard/realtime", response_model=dict)
 async def get_dashboard_realtime(
     request: Request,

backend/server/routers/channels/directory_router.py

@@ -27,10 +27,14 @@ from __future__ import annotations
 
 from typing import Any
 
-from fastapi import APIRouter, Depends, Query, Request
+from fastapi import APIRouter, Body, Depends, Query, Request
-
+from pydantic import BaseModel, ConfigDict, Field
 from yuxi.channels.contract.dtos.channel import ChannelType
-from yuxi.channels.contract.dtos.directory import ClearDirectoryCacheCmd, DirectoryQuery
+from yuxi.channels.contract.dtos.directory import (
+    BatchDirectoryProfilesQuery,
+    ClearDirectoryCacheCmd,
+    DirectoryQuery,
+)
 from yuxi.storage.postgres.models_business import User
 
 from server.routers.channels import (
@@ -143,6 +147,52 @@ async def get_directory_user_profile(
     return {"success": True, "data": serialize_control_data(result.data)}
 
 
+class BatchProfilesRequest(BaseModel):
+    """批量目录资料查询请求。"""
+
+    model_config = ConfigDict(frozen=True)
+
+    peer_ids: list[str] = Field(
+        ...,
+        min_length=1,
+        description="待查询的对端 ID 列表（可能是用户 ID 或群组 ID）",
+    )
+
+
+@directory_router.post(
+    "/{channel_type}/{account_id}/directory/batch_profiles",
+    response_model=dict,
+)
+async def batch_directory_profiles(
+    channel_type: ChannelType,
+    account_id: str,
+    request: Request,
+    payload: BatchProfilesRequest = Body(...),
+    use_cases=Depends(get_channel_use_cases),
+    current_user: User = Depends(get_admin_user),
+) -> dict[str, Any]:
+    """批量查询对端资料（用户或群组）。
+
+    给定一组 peer_id，逐个尝试查询用户资料；若用户不存在，则尝试查询
+    群组详情。结果统一返回为目录条目（type 区分 user/group），用于路由
+    绑定列表等需要把抽象 ID 翻译为可读名称的场景。5min TTL 缓存。
+
+    对应控制面操作 ``directory/batch_profiles``（由
+    ``ChannelControlService.batchGetDirectoryProfiles`` 内部构造
+    ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。
+    """
+    operator = build_operator(current_user, request)
+    query = BatchDirectoryProfilesQuery(peer_ids=tuple(payload.peer_ids))
+    result = await use_cases.directory.batchGetDirectoryProfiles(
+        channel_type=channel_type,
+        account_id=account_id,
+        query=query,
+        operator=operator,
+    )
+    raiseOnControlFailure(result)
+    return {"success": True, "data": serialize_control_data(result.data)}
+
+
 @directory_router.get(
     "/{channel_type}/{account_id}/directory/groups/{group_id}",
     response_model=dict,

backend/server/routers/channels/doctor_router.py

@@ -29,7 +29,6 @@ from __future__ import annotations
 from typing import Any
 
 from fastapi import APIRouter, Depends, Request
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.storage.postgres.models_business import User
 

backend/server/routers/channels/health_router.py

@@ -41,7 +41,6 @@ from typing import Any, Literal
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.health import (
     DiagnosticsExportRequest,
@@ -110,12 +109,8 @@ async def get_health(
 
 @health_router.get("/health/detail", response_model=dict)
 async def get_health_detail(
-    channel_type: ChannelType | None = Query(
+    channel_type: ChannelType | None = Query(default=None, description="按渠道过滤（None 表示全渠道）"),
-        default=None, description="按渠道过滤（None 表示全渠道）"
+    with_probe: bool = Query(default=False, description="是否绕过缓存触发实时聚合（非调用 probeChannel）"),
-    ),
-    with_probe: bool = Query(
-        default=False, description="是否绕过缓存触发实时聚合（非调用 probeChannel）"
-    ),
     use_cases=Depends(get_channel_use_cases),
     current_user: User = Depends(get_admin_user),
 ) -> dict[str, Any]:
@@ -157,17 +152,11 @@ class DiagnosticsExportPayload(BaseModel):
 
     model_config = ConfigDict(frozen=True)
 
-    channel_filter: ChannelType | None = Field(
+    channel_filter: ChannelType | None = Field(default=None, description="按渠道过滤（None 表示全渠道）")
-        default=None, description="按渠道过滤（None 表示全渠道）"
-    )
     include_audit_logs: bool = Field(default=True, description="是否包含审计日志")
-    audit_log_count: int = Field(
+    audit_log_count: int = Field(default=100, ge=1, le=500, description="审计日志条数上限")
-        default=100, ge=1, le=500, description="审计日志条数上限"
-    )
     include_error_logs: bool = Field(default=True, description="是否包含错误日志")
-    error_log_count: int = Field(
+    error_log_count: int = Field(default=100, ge=1, le=500, description="错误日志条数上限")
-        default=100, ge=1, le=500, description="错误日志条数上限"
-    )
 
 
 # ---------------- Endpoints ----------------
@@ -272,9 +261,7 @@ async def get_single_health(
         channel_type=channel_type,
         account_id=account_id,
     )
-    result = await use_cases.health_check_control.getSingleHealth(
+    result = await use_cases.health_check_control.getSingleHealth(query, operator=operator)
-        query, operator=operator
-    )
     raiseOnControlFailure(result)
     return {"success": True, "data": serialize_control_data(result.data)}
 

backend/server/routers/channels/login_router.py

@@ -42,7 +42,6 @@ from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.login import ForceLogoutCmd
 from yuxi.storage.postgres.models_business import User

backend/server/routers/channels/message_router.py

@@ -37,7 +37,6 @@ from typing import Any, Literal
 
 from fastapi import APIRouter, Depends, File, Form, Header, Query, Request, UploadFile
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.admin import AdminSendCmd
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.common import MessageContent

backend/server/routers/channels/outbox_router.py

@@ -89,7 +89,9 @@ async def list_outbox_messages(
     channel_session_id: str | None = Query(default=None, description="按渠道会话 ID 过滤"),
     created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"),
     created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"),
-    retry_count_min: int | None = Query(default=None, ge=0, description="最小重试次数下界（含），用于筛选已发生重试的条目"),
+    retry_count_min: int | None = Query(
+        default=None, ge=0, description="最小重试次数下界（含），用于筛选已发生重试的条目"
+    ),
     limit: int = Query(default=100, ge=1, le=200, description="分页大小"),
     offset: int = Query(default=0, ge=0, description="分页偏移"),
     use_cases=Depends(get_channel_use_cases),
@@ -124,6 +126,7 @@ async def list_outbox_messages(
 async def get_outbox_stats(
     request: Request,
     channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
+    channel_account_id: str | None = Query(default=None, description="按渠道账户 ID 过滤"),
     use_cases=Depends(get_channel_use_cases),
     current_user: User = Depends(get_admin_user),
 ) -> dict[str, Any]:
@@ -132,6 +135,7 @@ async def get_outbox_stats(
     result = await use_cases.outbox_management.getOutboxStats(
         channel_type,
         operator=operator,
+        channel_account_id=channel_account_id,
     )
     raiseOnControlFailure(result)
     return {"success": True, "data": serialize_control_data(result.data)}
@@ -230,9 +234,7 @@ async def batch_retry_dead_letter(
     """
     operator = build_operator(current_user, request)
     outbox_ids = cmd.outbox_ids if cmd else None
-    result = await use_cases.outbox_management.batchRetryDeadLetter(
+    result = await use_cases.outbox_management.batchRetryDeadLetter(operator=operator, outbox_ids=outbox_ids)
-        operator=operator, outbox_ids=outbox_ids
-    )
     raiseOnControlFailure(result)
     return {"success": True, "data": serialize_control_data(result.data)}
 
@@ -252,9 +254,7 @@ async def batch_delete_dead_letter(
     """
     operator = build_operator(current_user, request)
     outbox_ids = cmd.outbox_ids if cmd else None
-    result = await use_cases.outbox_management.batchDeleteDeadLetter(
+    result = await use_cases.outbox_management.batchDeleteDeadLetter(operator=operator, outbox_ids=outbox_ids)
-        operator=operator, outbox_ids=outbox_ids
-    )
     raiseOnControlFailure(result)
     return {"success": True, "data": serialize_control_data(result.data)}
 
@@ -265,9 +265,7 @@ async def export_dead_letter(
     channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
     created_after: str | None = Query(default=None, description="创建时间下界（ISO 8601）"),
     created_before: str | None = Query(default=None, description="创建时间上界（ISO 8601）"),
-    export_format: Literal["json", "csv"] = Query(
+    export_format: Literal["json", "csv"] = Query(default="json", alias="format", description="导出格式（json / csv）"),
-        default="json", alias="format", description="导出格式（json / csv）"
-    ),
     limit: int = Query(default=10000, ge=1, le=100000, description="导出条目上限，防止大规模死信队列 OOM"),
     use_cases=Depends(get_channel_use_cases),
     current_user: User = Depends(get_admin_user),

backend/server/routers/channels/pairing_router.py

@@ -23,9 +23,8 @@ from __future__ import annotations
 
 from typing import Any
 
-from fastapi import APIRouter, Body, Depends, Path, Query, Request
+from fastapi import APIRouter, Body, Depends, Query, Request
 from pydantic import BaseModel, ConfigDict, Field
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.channels.contract.dtos.pairing import (
     BatchPairingCmd,
@@ -146,6 +145,41 @@ async def list_pairings(
     return {"success": True, "data": serialize_control_data(result.data)}
 
 
+@pairing_router.get("/pairings/count", response_model=dict)
+async def count_pairings(
+    request: Request,
+    channel_type: ChannelType | None = Query(default=None, description="按渠道类型过滤"),
+    account_id: str | None = Query(default=None, description="按账户 ID 过滤"),
+    peer_id: str | None = Query(default=None, description="按对端 ID 过滤"),
+    status: PairingStatus | None = Query(default=None, description="按状态过滤"),
+    created_after: str | None = Query(default=None, description="申请时间下界（ISO 8601，含，按 requested_at 过滤）"),
+    created_before: str | None = Query(default=None, description="申请时间上界（ISO 8601，含，按 requested_at 过滤）"),
+    use_cases=Depends(get_channel_use_cases),
+    current_user: User = Depends(get_admin_user),
+) -> dict[str, Any]:
+    """统计配对审批记录数（FR-33 待办角标）。
+
+    支持按渠道/账户/对端/状态过滤，返回单个整数计数。
+    对应控制面操作 ``pairing/count``。
+    """
+    operator = build_operator(current_user, request)
+    created_after_dt = parse_datetime("created_after", created_after)
+    created_before_dt = parse_datetime("created_before", created_before)
+    query = PairingQuery(
+        channel_type=channel_type,
+        account_id=account_id,
+        peer_id=peer_id,
+        status=status,
+        created_after=created_after_dt,
+        created_before=created_before_dt,
+        limit=1,
+        offset=0,
+    )
+    result = await use_cases.pairing_management.countPairings(query, operator)
+    raiseOnControlFailure(result)
+    return {"success": True, "data": serialize_control_data(result.data)}
+
+
 @pairing_router.post("/pairings", response_model=dict)
 async def create_pairing(
     payload: CreatePairingRequest,

backend/server/routers/channels/plugin_router.py

@@ -102,7 +102,10 @@ class BatchPluginLifecycleRequest(BaseModel):
 @plugin_router.get("", response_model=dict)
 async def list_plugins(
     request: Request,
-    state: PluginStateLiteral | None = Query(default=None, description="按生命周期状态过滤（discovered/resolved/loaded/initialized/started/paused/stopped/unloaded/failed/not_installed/installed）"),
+    state: PluginStateLiteral | None = Query(
+        default=None,
+        description="按生命周期状态过滤（discovered/resolved/loaded/initialized/started/paused/stopped/unloaded/failed/not_installed/installed）",
+    ),
     use_cases=Depends(get_channel_use_cases),
     current_user: User = Depends(get_admin_user),
 ) -> dict[str, Any]:

backend/server/routers/channels/session_router.py

@@ -1,6 +1,6 @@
 """会话资源域 Router。
 
-提供渠道会话管理的 11 个 HTTP 端点，全部由 ``get_admin_user`` 守门，仅管理员
+提供渠道会话管理的 12 个 HTTP 端点，全部由 ``get_admin_user`` 守门，仅管理员
 可访问。Router 不含任何业务逻辑，仅做协议翻译：将 HTTP 请求参数组装为契约层
 命令/参数，调用 ``use_cases.session_management`` 的类型化方法走控制面管道，
 再通过 ``raiseOnControlFailure`` 转译失败、``serialize_control_data`` 序列化
@@ -10,6 +10,7 @@
 端点清单：
     - GET    /sessions                                    SES-QUERY-01  列出渠道会话
     - POST   /sessions/batch-close                        SES-BATCH-CLOSE-01 批量关闭会话
+    - GET    /sessions/events                             SES-EVENTS-01 渠道会话实时事件 SSE
     - GET    /sessions/{session_id}                       SES-QUERY-02  查询会话详情
     - POST   /sessions/{session_id}/merge                 SES-05        合并会话（FR-07）
     - POST   /sessions/{session_id}/transfer              SES-13        转移会话所有者（FR-26）
@@ -23,22 +24,32 @@
 
 from __future__ import annotations
 
+import asyncio
+import json
+from collections.abc import AsyncIterator
 from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, ConfigDict, Field
-
+from yuxi.channels.application.extension.channel_event_broadcaster import (
+    ChannelEventBroadcaster,
+)
 from yuxi.channels.contract.dtos.channel import ChannelType, SessionStatus
 from yuxi.channels.contract.dtos.conversation import MergeConversationCmd
+from yuxi.channels.contract.dtos.plugin import DomainEvent
 from yuxi.channels.contract.dtos.session import (
     BatchCloseSessionsCmd,
     CloseSessionCmd,
     OwnerTransferCmd,
 )
+from yuxi.channels.infrastructure import DependencyInjectionContainer
 from yuxi.storage.postgres.models_business import User
 
 from server.routers.channels import (
     build_operator,
+    dataclass_to_dict,
+    get_channel_di_container,
     get_channel_use_cases,
     parse_datetime,
     raiseOnControlFailure,
@@ -212,9 +223,7 @@ async def batch_close_sessions(
     if payload.filter is not None:
         filter_param = {
             "channel_type": payload.filter.channel_type,
-            "inactive_before": parse_datetime(
+            "inactive_before": parse_datetime("inactive_before", payload.filter.inactive_before),
-                "inactive_before", payload.filter.inactive_before
-            ),
             "status": payload.filter.status,
         }
     cmd = BatchCloseSessionsCmd(
@@ -229,6 +238,66 @@ async def batch_close_sessions(
     return {"success": True, "data": serialize_control_data(result.data)}
 
 
+@session_router.get("/sessions/events")
+async def stream_session_events(
+    request: Request,
+    di_container: DependencyInjectionContainer = Depends(get_channel_di_container),
+    current_user: User = Depends(get_admin_user),
+) -> StreamingResponse:
+    """渠道会话/消息实时事件 SSE 端点（BE-14）。
+
+    返回 ``text/event-stream`` 长连接，将 ``ChannelSessionUpdated`` /
+    ``ChannelMessageReceived`` / ``ChannelMessageSent`` 三类事件以 SSE 帧
+    形式推送给已认证管理员。每 15 秒无事件时发送心跳注释，客户端断开时
+    自动取消订阅。
+
+    本端点静态路径先于 ``/sessions/{session_id}`` 声明，避免被动态参数捕获。
+    """
+    broadcaster = di_container.resolve(ChannelEventBroadcaster)
+
+    async def event_stream() -> AsyncIterator[str]:
+        """SSE 事件流异步生成器。"""
+        queue, subscription_id = await broadcaster.subscribe()
+        try:
+            while not await request.is_disconnected():
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=15.0)
+                except TimeoutError:
+                    if await request.is_disconnected():
+                        break
+                    yield ":heartbeat\n\n"
+                    continue
+                if await request.is_disconnected():
+                    break
+                yield _format_sse_event(event)
+        finally:
+            await broadcaster.unsubscribe(subscription_id)
+
+    return StreamingResponse(
+        event_stream(),
+        media_type="text/event-stream; charset=utf-8",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+def _format_sse_event(event: DomainEvent) -> str:
+    """将 ``DomainEvent`` 格式化为 SSE 帧。
+
+    输出格式：
+        event: {event.event_type}\n
+        id: {event.event_id}\n
+        data: {json.dumps(event_dict)}\n\n
+
+    其中 ``event_dict`` 经 ``dataclass_to_dict`` 递归序列化为 JSON 兼容结构。
+    """
+    event_dict = dataclass_to_dict(event)
+    return f"event: {event.event_type}\nid: {event.event_id}\ndata: {json.dumps(event_dict, ensure_ascii=False)}\n\n"
+
+
 @session_router.get("/sessions/{session_id}", response_model=dict)
 async def get_session(
     session_id: str,

backend/server/routers/channels/wizard_router.py

@@ -29,7 +29,6 @@ from urllib.parse import urlparse
 
 from fastapi import APIRouter, Depends, Request
 from pydantic import BaseModel, ConfigDict, Field, field_validator
-
 from yuxi.channels.contract.dtos.channel import ChannelType
 from yuxi.storage.postgres.models_business import User
 
@@ -142,6 +141,15 @@ class OAuthInitiateRequest(BaseModel):
         return v
 
 
+class QrLoginWaitRequest(BaseModel):
+    """扫码登录轮询请求。字段对齐 ``WizardPort.waitWizardQrLogin`` 入参。"""
+
+    model_config = ConfigDict(frozen=True)
+
+    session_id: str = Field(..., min_length=1, description="扫码会话 ID（由 qr/initiate 返回）")
+    current_qr_data_url: str | None = Field(None, description="当前二维码数据 URL（用于刷新场景）")
+
+
 # ---------------- Endpoints ----------------
 
 
@@ -310,3 +318,51 @@ async def handle_wizard_oauth_callback(
     )
     raiseOnControlFailure(result)
     return {"success": True, "data": serialize_control_data(result.data)}
+
+
+@wizard_router.post("/{channel_type}/wizard/qr/initiate", response_model=dict)
+async def initiate_wizard_qr_login(
+    channel_type: ChannelType,
+    request: Request,
+    use_cases=Depends(get_channel_use_cases),
+    current_user: User = Depends(get_admin_user),
+) -> dict[str, Any]:
+    """发起向导扫码登录（WIZ-07）。
+
+    对应控制面操作 ``wizard/qr-initiate``（由 ``ChannelControlService.initiateWizardQrLogin``
+    内部构造 ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。
+    在账户尚未创建时复用 ``LoginAdapter`` 获取二维码，返回 session_id 与
+    qr_data_url 供前端展示并轮询。
+    """
+    operator = build_operator(current_user, request)
+    result = await use_cases.wizard.initiateWizardQrLogin(
+        channel_type=channel_type,
+        operator=operator,
+    )
+    raiseOnControlFailure(result)
+    return {"success": True, "data": serialize_control_data(result.data)}
+
+
+@wizard_router.post("/{channel_type}/wizard/qr/wait", response_model=dict)
+async def wait_wizard_qr_login(
+    channel_type: ChannelType,
+    payload: QrLoginWaitRequest,
+    request: Request,
+    use_cases=Depends(get_channel_use_cases),
+    current_user: User = Depends(get_admin_user),
+) -> dict[str, Any]:
+    """轮询向导扫码登录结果（WIZ-08）。
+
+    对应控制面操作 ``wizard/qr-wait``（由 ``ChannelControlService.waitWizardQrLogin``
+    内部构造 ``ControlCmd`` 并委托 ``_executeControl`` 执行控制面管道）。
+    扫码成功后返回 credentials，由前端填入当前向导步骤并通过 ``apply`` 提交。
+    """
+    operator = build_operator(current_user, request)
+    result = await use_cases.wizard.waitWizardQrLogin(
+        channel_type=channel_type,
+        session_id=payload.session_id,
+        current_qr_data_url=payload.current_qr_data_url,
+        operator=operator,
+    )
+    raiseOnControlFailure(result)
+    return {"success": True, "data": serialize_control_data(result.data)}