feat: 配置中心-凭据（邮件推送AI表单/掩码交互/SMTP测试/推送探测端点）

- 后端 POST /tenants/{slug}/email/test：resolve_email_settings 当前配置
  真发测试邮件（主题 [all-kefu] SMTP 测试邮件+租户名+UTC+8 时间）；
  回退后 host/user 仍空 → 400；成败如实返回 {sent, error}，密钥不进响应
- 后端 POST /tenants/{slug}/push/test：连通性探测——与 _do_post 同款
  连接形态 POST 空 messages 批量（红线：绝不携带线索数据）；不可达
  reason 分类 dns/timeout/connect（Windows 裸 ConnectError 文案兜底），
  auth_ok 三态（2xx 真 / 401·403 假 / 其余 null），单次无重试
- email_service 抽 _smtp_send 共用连接逻辑（_send 契约不变）+
  send_test_email（成败+错误串如实返回，不走吞异常 bool）
- 前端 CredsPane：邮件/推送/AI 三组凭据表单（数值/邮箱前端预检 +
  后端 400 双层）；密钥哨兵交互（__SET__ 占位符/不动保存保持库值/
  清除按钮显式清空可撤销）；SMTP 测试与推送探测三态内联展示
- 测试：pytest 新增 18（email/test 7 + push/test 11），全量 595 通过；
  vitest 新增 20（model 哨兵往返/预检边界 + API 调用形态），107 通过

Co-Authored-By: Claude Code <noreply@anthropic.com>

app/engine/services/email_service.py

@@ -1,5 +1,6 @@
 import html
 import logging
+from datetime import datetime, timedelta, timezone
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 
@@ -58,16 +59,23 @@ def _cc_views(contacts: dict, cc_info: dict | None) -> tuple[dict, dict]:
     return body, subj
 
 
+async def _smtp_send(msg: MIMEMultipart, cfg: dict) -> None:
+    """底层连接发送(Phase 3 Task 7c 从 _send 抽出): 直调 aiosmtplib,
+    失败抛原始异常——需要错误串的调用方(配置中心测试邮件 send_test_email)
+    复用同一连接逻辑;_send 在此之上包"吞异常返 bool"契约,行为不变。"""
+    await aiosmtplib.send(
+        msg, hostname=cfg["host"], port=cfg["port"],
+        username=cfg["user"], password=cfg["password"],
+        use_tls=True,
+    )
+
+
 async def _send(msg: MIMEMultipart, cfg: dict) -> bool:
     if not all([cfg["host"], cfg["user"], cfg["notify_emails"]]):
         logger.warning("SMTP not configured, skipping email")
         return False
     try:
-        await aiosmtplib.send(
-            msg, hostname=cfg["host"], port=cfg["port"],
-            username=cfg["user"], password=cfg["password"],
-            use_tls=True,
-        )
+        await _smtp_send(msg, cfg)
         logger.info("Lead email sent to %s", cfg["notify_emails"])
         return True
     except Exception:
@@ -170,3 +178,35 @@ async def send_correction_email(
         f"<table>{_contact_rows(body_contacts)}</table>"
         f"</body></html>", "html", "utf-8"))
     return await _send(msg, cfg)
+
+
+async def send_test_email(
+    to: str, tenant_name: str = "",
+    config_data: dict | None = None,
+) -> tuple[bool, str | None]:
+    """配置中心测试邮件(Phase 3 Task 7c): 按当前生效邮件配置真发一封到
+    指定收件人——主题/正文钉契约("[all-kefu] SMTP 测试邮件"+租户名+时间,
+    时间与 lead_push_service 同款 +08:00 口径)。
+
+    与 _send 的吞异常 bool 契约解耦: 成败与错误串如实返回(探测端点直传
+    前端,凭据对不对让收件箱与错误串说话);连接逻辑复用 _smtp_send。
+    "SMTP 未配置"的 400 判定在端点层(本函数假定 cfg 已可用)。"""
+    cfg = resolve_email_settings(config_data)
+    now = datetime.now(timezone(timedelta(hours=8)))
+    msg = MIMEMultipart("alternative")
+    msg["Subject"] = "[all-kefu] SMTP 测试邮件"
+    msg["From"] = f"【{cfg['sender_name']}】 <{cfg['user']}>"
+    msg["To"] = to
+    msg.attach(MIMEText(
+        f"<html><body>"
+        f"<p>这是一封 SMTP 配置测试邮件——收到本邮件说明该租户的邮件配置"
+        f"可以正常发信。</p>"
+        f"<p>租户：{_esc(tenant_name)}</p>"
+        f"<p>时间：{now.strftime('%Y-%m-%d %H:%M:%S')}（UTC+8）</p>"
+        f"</body></html>", "html", "utf-8"))
+    try:
+        await _smtp_send(msg, cfg)
+        return True, None
+    except Exception as exc:
+        logger.exception("Test email to %s failed", to)
+        return False, repr(exc)

app/platform/routes/config_routes.py

@@ -19,6 +19,10 @@
   缺省用当前墙——逐句调 boundary_filter 纯函数给判定,**不落库不 bump 版本**
   (reset/编辑器的预览入口)。判定优先级与 conversation.py 执行侧一致:
   价格墙 → 零号码墙 → 再索要墙(仅 has_contact);
+- POST /email/test 与 /push/test(凭据测试,Phase 3 Task 7c):分别用
+  resolve_email_settings/resolve_push_settings 的当前配置真发测试邮件 /
+  探测推送连通性——**push 探测绝不携带线索数据**(空 messages 批量),
+  两端点不落库不 bump 版本,成败如实返回(400 只用于"未配置"预检);
 - GET/PUT /api/platform/settings:PlatformSetting CRUD,仅超管。本阶段只落
   存储与接口(value 以 JSON Text 直存,读回还原类型);resolver 的平台默认
   回退链接入归 Phase 3。
@@ -36,12 +40,15 @@ email.smtp.password 及平台设置的密钥名值,所有携带 config/value 的
 """
 import copy
 import json
+import socket
 
+import httpx
 from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.engine.services import boundary_filter
+from app.engine.services.email_service import send_test_email
 from app.engine.wall_schema import WallConfigError, compile_wall
 from app.platform.auth import UserContext, require_super
 from app.platform.config_validation import (
@@ -49,6 +56,10 @@ from app.platform.config_validation import (
     ValidationContext,
     validate_group,
 )
+from app.platform.config_views import (
+    resolve_email_settings,
+    resolve_push_settings,
+)
 from app.platform.db import (
     PlatformSetting,
     Tenant,
@@ -488,6 +499,109 @@ async def test_wall(slug: str, request: Request, user=Depends(_tenant_guard)):
     return {"slug": slug, "wall_source": wall_source, "results": results}
 
 
+# ---- 凭据测试端点(email/push test,Phase 3 Task 7c) -------------------------------
+
+_EMAIL_TEST_KEYS = frozenset({"to"})
+
+
+@router.post("/{slug}/email/test")
+async def test_email(slug: str, request: Request,
+                     user=Depends(_tenant_guard)):
+    """SMTP 测试邮件: 用当前生效邮件配置(resolve_email_settings,租户组
+    优先/空槽回退 env SMTP_*)真发一封到 body.to——主题/正文契约见
+    email_service.send_test_email。SMTP 未配置(回退后 host/user 仍空)
+    → 400;发送失败不 5xx,{sent: false, error} 如实返回。不落库不 bump
+    版本;密钥绝不进响应。"""
+    body = await _json_body(request)
+    unknown = set(body) - _EMAIL_TEST_KEYS
+    if unknown:
+        raise HTTPException(
+            status_code=400,
+            detail=f"未知字段: {', '.join(sorted(unknown))}(合法: to)")
+    to = body.get("to")
+    if not isinstance(to, str) or "@" not in to.strip():
+        raise HTTPException(
+            status_code=400,
+            detail=f"to: 须为含 @ 的收件人邮箱(实际 {to!r})")
+
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        _, snapshot = await _load_current(session, tenant.id, slug)
+        cfg = resolve_email_settings(snapshot)
+    if not cfg["host"] or not cfg["user"]:
+        raise HTTPException(
+            status_code=400,
+            detail="SMTP 未配置(服务器/账号在租户配置与环境回退后均为空)"
+                   "——请先在「配置中心 → 凭据」保存邮件 SMTP")
+    sent, error = await send_test_email(to.strip(), tenant.name, snapshot)
+    return {"slug": slug, "sent": sent, "error": error}
+
+
+def _transport_reason(exc: Exception) -> str:
+    """传输异常 → 不可达原因分类(前端三态展示): dns(域名解析失败)/
+    timeout(TimeoutException 族)/connect(拒绝·不可达·TLS 等其余)。
+
+    dns 判定双通道: 沿 __cause__/__context__ 链找 gaierror(httpx 把解析
+    错误包在 ConnectError 里),并对链上异常做文案匹配——Windows 实测
+    (冒烟 Task 7c)httpx/anyio 会把 errno 11001 包成裸 ConnectError,
+    cause 链不带 gaierror,只能从 "[Errno 11001] getaddrinfo failed"
+    文案识别;分类仅供修复提示,错误原文始终随响应返回兜底。"""
+    node: BaseException | None = exc
+    while node is not None:
+        if isinstance(node, socket.gaierror):
+            return "dns"
+        text = str(node).lower()
+        if "getaddrinfo" in text or "name or service not known" in text:
+            return "dns"
+        node = node.__cause__ or node.__context__
+    if isinstance(exc, httpx.TimeoutException):
+        return "timeout"
+    return "connect"
+
+
+async def _probe_push(cfg: dict) -> dict:
+    """单次连通性探测(无重试): 与 lead_push_service._do_post 同款连接形态
+    (timeout/follow_redirects/X-Api-Key 头),body 恒为空 messages 列表——
+    **绝不携带任何线索数据**(UniService 是批量结构,空批量对方案应为
+    无操作;契约文档不在库内,这是可用的最保守预检形态)。"""
+    try:
+        async with httpx.AsyncClient(timeout=cfg["timeout"],
+                                     follow_redirects=True) as client:
+            resp = await client.post(
+                cfg["url"], json={"messages": []},
+                headers={"X-Api-Key": cfg["api_key"]})
+    except httpx.HTTPError as exc:
+        return {"reachable": False, "auth_ok": None, "status": None,
+                "reason": _transport_reason(exc),
+                "error": str(exc) or repr(exc)}
+    auth_ok = (True if resp.status_code < 300
+               else False if resp.status_code in (401, 403)
+               else None)
+    return {"reachable": True, "auth_ok": auth_ok,
+            "status": resp.status_code, "reason": None, "error": None}
+
+
+@router.post("/{slug}/push/test")
+async def test_push(slug: str, request: Request,
+                    user=Depends(_tenant_guard)):
+    """推送连通性探测: 只验 URL 可达 + 鉴权通过,不发真数据(空 messages
+    批量,见 _probe_push)。URL 回退后仍空 → 400;其余如实返回——不可达
+    (含 dns/timeout/connect 原因分类)/可达但鉴权失败(401/403)/正常(2xx);
+    其余状态码可达但鉴权不定(auth_ok=null)。不落库不 bump 版本。"""
+    engine = request.app.state.platform_engine
+    async with AsyncSession(engine, expire_on_commit=False) as session:
+        tenant = await _get_tenant_or_404(session, slug)
+        _, snapshot = await _load_current(session, tenant.id, slug)
+    cfg = resolve_push_settings(snapshot)
+    if not cfg["url"]:
+        raise HTTPException(
+            status_code=400,
+            detail="推送 URL 未配置(租户配置与环境回退后均为空)"
+                   "——请先在「配置中心 → 凭据」保存推送 URL")
+    return {"slug": slug, **await _probe_push(cfg)}
+
+
 # ---- 平台设置(仅超管;resolver 回退链接入归 Phase 3) -----------------------------