feat: 2026-07-12~13 全量更新 - AI对话链路改造+H5 v4/v5+坐席端v5+上下文感知诊断+知识库迭代3

## H5 员工端 v4 (2026-07-13 00:48 已部署) - 人工按钮三态文案统一为"人工坐席" - 按钮位置移至发送键和语音按钮上方(垂直堆叠) - 点按钮直接调 store.shakeAgent()，删除 CallAgentModal 弹窗动画 - 截图快捷键提示改为"截图->粘贴：Alt+Shift+A-Ctrl+V ---> Ctrl+V" - 移动端隐藏截图提示(CSS 媒体查询) - AI转人工提示改为"已为您呼叫人工坐席，请稍等!" - 坐席接入提示改为"坐席正在查看您的信息，请等待处理回复！" - 删除"摇铃呼叫坐席"入口和文案 - 删除孤儿组件 MessageList.vue + shake 动画 CSS ## H5 员工端 v5 (2026-07-13 02:08 已部署) - RightPanel v2.1：删除"软件安装"和"资源权限"标签页 - 移除标签栏，智能推荐(DynamicRecommend)直接展示 - 删除 SoftwareDownloads/ApprovalLinks 引用和相关 CSS ## AI 对话链路全栈改造 Phase 1-6 (已部署) - Phase 1: Dify JSON输出 + 后端blocking解析 + 双WS推送 + 错误降级 - Phase 2: 关键词收窄(~25强意图词) + 两级分类Prompt + 删除前端checkApprovalIntent - Phase 3: WS扩展(ai_thinking+dynamic_recommend) + ai_structured气泡 + RightPanel v2 + 选项回传 - Phase 4: VisionService接入 + 图片消息融合(5秒窗口) + 降级策略 - Phase 5: 坐席端ai_thinking指示器 + ai_structured/byod_card渲染 + handleNewMessage修复 - Phase 6: diagnosis_stage(6值) + response_time_ms计时 + 慢响应告警(>10s) ## 坐席端 v5 (2026-07-13 01:38 已部署) - ai_structured/byod_card 只读渲染 - AI思考指示器 UI - handleNewMessage 透传 msg_type/extra_data 修复 - 布局优化v2.0: QuickReplyBar L1+L2悬浮 + ReplyBox左右分区 + 右栏260/560px切换 - 键盘快捷键v2.3: 纯数字路由 + ESC分层撤销 + Shift+Space用event.code ## 上下文感知智能诊断闭环 (2026-07-12 已部署) - 三层诊断(API→Script→AI) + 三段排队(VIP→info_locked→not locked) - 答题插队 + 五场景关闭 - 迁移052(6表+6列) + queue_service + quiz_service + closing_service - H5前端: QueueWaiting + RightPanel双Tab + InputBar三态 + ResolveConfirmCard - 坐席前端: pending_close结单流程 + 信息锁定(Dify步骤完成+有效回答率≥70%) ## 知识库迭代3 (2026-07-12 已部署) - 分诊交互(H5+坐席+Dify独立应用) - 拓扑预览(ECharts只读) - 代答排除(4种匹配器: keyword/regex/intent/category) - 迁移051 + 44文件43测试通过 ## 后端变更 - 6个Python文件改造(h5_ai_task.py/h5.py/ai_service.py/closing_service.py等) - funny_phrase_service.py: shake/connected/keyword 默认文案更新 - session_service.py: 企微消息文案同步 - 新增: queue.py/quiz.py/triage.py/exclusion_rules.py 等API端点 - 新增: diagnostic.py/quiz.py/triage_session.py 等模型 - 新增: closing_service/queue_service/quiz_service/triage_service 等服务 ## 文档更新 - CHANGELOG.md: 新增 [未发布] 区全部变更记录 - 项目管理主文档 v2.5: 新增v0.7.3版本 + 已完成看板 + 最近搞定 - 版本记录: 新增v0.7.3条目 - AI对话链路实施计划: Phase 1-6 全部标记✅已实施 - 新增架构图/时序图/类图(mermaid) ## 部署路径修正 - 服务器项目根路径: /opt/wecom-it-desk/ - 所有前端dist均为ro bind mount，只能在宿主机源路径操作 - 服务器nginx /h5/ 是静态文件服务(非proxy_pass) - elFinder上传二进制不可靠(MD5不匹配)，改用base64分块上传
2026-07-13 02:17:03 +08:00
parent bea288e414
commit 449c6d4875
176 changed files with 46637 additions and 4805 deletions
@@ -0,0 +1,820 @@
+# -*- coding: utf-8 -*-
+"""代答排除模块回归测试 — T04
+
+测试范围:
+  Matcher单元测试（12个）: keyword(5) + regex(3) + intent(2) + category(2)
+  ExclusionService测试（5个）: no_rules, priority_order, first_hit_stops,
+                               logs_hit, test_match_no_logging
+  API测试（8个）: create, duplicate_name, get_detail, update, delete,
+                  toggle, test_match, get_stats
+  ai_handler集成测试（3个）: exclusion_hit, exclusion_miss, exclusion_error
+
+测试依赖: conftest.py 提供的 client / db_session fixtures
+"""
+
+import uuid
+from contextlib import ExitStack
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.exclusion_rule import ExclusionRule
+from app.models.exclusion_log import ExclusionLog
+from app.models.triage_session import TriageSession
+
+
+# ============================================================================
+# 辅助函数
+# ============================================================================
+
+async def _login_admin(client: AsyncClient, db_session: AsyncSession) -> str:
+    """创建 admin 角色用户并返回 Bearer token。"""
+    from app.models.role import Role
+    from app.models.user_role import UserRole
+
+    admin_id = f"test_admin_{uuid.uuid4().hex[:8]}"
+
+    # 确保 admin 角色存在
+    stmt = select(Role).where(Role.name == "admin")
+    result = await db_session.execute(stmt)
+    admin_role = result.scalars().first()
+    if not admin_role:
+        admin_role = Role(
+            name="admin", display_name="管理员",
+            description="系统管理员", permissions=[],
+        )
+        db_session.add(admin_role)
+        await db_session.flush()
+
+    # 创建 UserRole 关联
+    db_session.add(UserRole(
+        employee_id=admin_id, role_id=admin_role.id,
+        source="manual", assigned_by="test_fixture",
+    ))
+    await db_session.flush()
+
+    # 登录
+    resp = await client.post("/agents/login", json={
+        "user_id": admin_id, "name": "测试管理员",
+    })
+    return resp.json()["data"]["token"]
+
+
+async def _create_exclusion_rule(db_session: AsyncSession, **kwargs) -> ExclusionRule:
+    """在数据库中创建排除规则。
+
+    Args:
+        db_session: 数据库会话
+        **kwargs: 覆盖默认字段值
+
+    Returns:
+        ExclusionRule: 创建的规则对象
+    """
+    defaults = {
+        "rule_name": f"规则-{uuid.uuid4().hex[:8]}",
+        "rule_description": "测试规则",
+        "priority": "P2",
+        "match_type": "keyword",
+        "match_condition": "密码",
+        "match_scope": [],
+        "action_type": "transfer_human",
+        "transfer_message": "已为您转接人工坐席",
+        "status": "enabled",
+        "hit_count": 0,
+        "created_by": "test_admin",
+    }
+    defaults.update(kwargs)
+    rule = ExclusionRule(**defaults)
+    db_session.add(rule)
+    await db_session.flush()
+    return rule
+
+
+# ============================================================================
+# Section A — Matcher 单元测试（12个）
+# ============================================================================
+
+class TestKeywordMatcher:
+    """关键词匹配器测试。"""
+
+    @pytest.mark.asyncio
+    async def test_keyword_match_hit(self):
+        """关键词命中。"""
+        from app.services.matchers.keyword_matcher import KeywordMatcher
+
+        matcher = KeywordMatcher()
+        result = await matcher.match(
+            message="我的密码过期了怎么办",
+            condition="密码过期,账号锁定",
+        )
+        assert result.matched is True
+        assert "密码过期" in result.matched_detail
+
+    @pytest.mark.asyncio
+    async def test_keyword_match_miss(self):
+        """关键词未命中。"""
+        from app.services.matchers.keyword_matcher import KeywordMatcher
+
+        matcher = KeywordMatcher()
+        result = await matcher.match(
+            message="今天天气真好",
+            condition="密码过期,账号锁定",
+        )
+        assert result.matched is False
+
+    @pytest.mark.asyncio
+    async def test_keyword_match_case_insensitive(self):
+        """大小写不敏感匹配。"""
+        from app.services.matchers.keyword_matcher import KeywordMatcher
+
+        matcher = KeywordMatcher()
+        result = await matcher.match(
+            message="VPN connection failed",
+            condition="vpn,password",
+        )
+        assert result.matched is True
+        assert "vpn" in result.matched_detail.lower()
+
+    @pytest.mark.asyncio
+    async def test_keyword_match_multiple_keywords(self):
+        """多关键词逗号分隔，任一命中即匹配。"""
+        from app.services.matchers.keyword_matcher import KeywordMatcher
+
+        matcher = KeywordMatcher()
+        # 第三个关键词命中
+        result = await matcher.match(
+            message="打印机卡纸了",
+            condition="密码过期,账号锁定,打印机",
+        )
+        assert result.matched is True
+        assert "打印机" in result.matched_detail
+
+    @pytest.mark.asyncio
+    async def test_keyword_match_empty_message(self):
+        """空消息返回未命中。"""
+        from app.services.matchers.keyword_matcher import KeywordMatcher
+
+        matcher = KeywordMatcher()
+        result = await matcher.match(message="", condition="密码")
+        assert result.matched is False
+
+        # 空条件也应返回未命中
+        result = await matcher.match(message="密码过期", condition="")
+        assert result.matched is False
+
+
+class TestRegexMatcher:
+    """正则匹配器测试。
+
+    注意: RegexMatcher 使用 signal.SIGALRM 做 ReDoS 超时保护，
+    Windows 不支持 SIGALRM。测试通过 fixture patch signal 模块
+    使正则匹配在 Windows 上正常工作。
+    源码 regex_matcher.py 在 Windows 上存在兼容性问题
+    （SIGALRM 不可用时直接返回未命中，应降级为无超时匹配）。
+    """
+
+    @pytest.fixture(autouse=True)
+    def _patch_signal(self):
+        """Windows 兼容: patch signal.SIGALRM 使正则匹配器正常工作。"""
+        import signal as sig
+        if not hasattr(sig, "SIGALRM"):
+            with ExitStack() as stack:
+                stack.enter_context(patch.object(sig, "SIGALRM", 14, create=True))
+                stack.enter_context(patch.object(sig, "ITIMER_REAL", 0, create=True))
+                if not hasattr(sig, "setitimer"):
+                    stack.enter_context(
+                        patch.object(sig, "setitimer", create=True, return_value=None)
+                    )
+                stack.enter_context(
+                    patch.object(sig, "signal", return_value=sig.SIG_DFL)
+                )
+                yield
+        else:
+            yield
+
+    @pytest.mark.asyncio
+    async def test_regex_match_hit(self):
+        """正则命中。"""
+        from app.services.matchers.regex_matcher import RegexMatcher
+
+        matcher = RegexMatcher()
+        result = await matcher.match(
+            message="我的密码好像过期了",
+            condition="密码.*过期",
+        )
+        assert result.matched is True
+        assert "密码" in result.matched_detail
+
+    @pytest.mark.asyncio
+    async def test_regex_match_miss(self):
+        """正则未命中。"""
+        from app.services.matchers.regex_matcher import RegexMatcher
+
+        matcher = RegexMatcher()
+        result = await matcher.match(
+            message="今天天气真好",
+            condition="密码.*过期",
+        )
+        assert result.matched is False
+
+    @pytest.mark.asyncio
+    async def test_regex_match_invalid_pattern(self):
+        """无效正则返回未命中（不抛异常）。"""
+        from app.services.matchers.regex_matcher import RegexMatcher
+
+        matcher = RegexMatcher()
+        # 无效正则括号不匹配
+        result = await matcher.match(
+            message="测试消息",
+            condition="[unclosed",
+        )
+        assert result.matched is False
+
+
+class TestIntentMatcher:
+    """意图匹配器测试。"""
+
+    @pytest.mark.asyncio
+    async def test_intent_match_dify_unavailable(self):
+        """Dify 不可用时降级返回未命中。"""
+        from app.services.matchers.intent_matcher import IntentMatcher
+
+        matcher = IntentMatcher()
+        # Mock _recognize_intent 返回 None（Dify 不可用）
+        with patch.object(matcher, "_recognize_intent", return_value=None):
+            result = await matcher.match(
+                message="我的密码忘了",
+                condition="password_reset,account_unlock",
+            )
+        assert result.matched is False
+
+    @pytest.mark.asyncio
+    async def test_intent_match_hit(self):
+        """意图命中（mock Dify 返回匹配的意图）。"""
+        from app.services.matchers.intent_matcher import IntentMatcher
+
+        matcher = IntentMatcher()
+        # Mock _recognize_intent 返回 password_reset
+        with patch.object(matcher, "_recognize_intent", return_value="password_reset"):
+            result = await matcher.match(
+                message="我的密码忘了，帮我重置一下",
+                condition="password_reset,account_unlock",
+            )
+        assert result.matched is True
+        assert "password_reset" in result.matched_detail
+
+
+class TestCategoryMatcher:
+    """分类匹配器测试。"""
+
+    @pytest.mark.asyncio
+    async def test_category_match_hit(self, db_session: AsyncSession):
+        """分类命中 — 分诊记录的 problem_category 在排除列表中。"""
+        from app.services.matchers.category_matcher import CategoryMatcher
+
+        conv_id = f"conv-cat-{uuid.uuid4().hex[:8]}"
+        # 创建分诊记录，problem_category = "Outlook"
+        triage = TriageSession(
+            conversation_id=conv_id,
+            user_id="test_user",
+            user_name="测试",
+            request_title="测试",
+            request_content="测试内容",
+            source="wecom_h5",
+            status="routed",
+            urgency="low",
+            problem_category="Outlook",
+        )
+        db_session.add(triage)
+        await db_session.flush()
+
+        matcher = CategoryMatcher()
+        result = await matcher.match(
+            message="测试消息",
+            condition="Outlook,VPN,打印机",
+            context={"conversation_id": conv_id, "db": db_session},
+        )
+        assert result.matched is True
+        assert "Outlook" in result.matched_detail
+
+    @pytest.mark.asyncio
+    async def test_category_match_no_triage(self, db_session: AsyncSession):
+        """无分诊记录返回未命中（软依赖）。"""
+        from app.services.matchers.category_matcher import CategoryMatcher
+
+        matcher = CategoryMatcher()
+        # 使用不存在的 conversation_id
+        result = await matcher.match(
+            message="测试消息",
+            condition="Outlook,VPN",
+            context={"conversation_id": "non-existent-conv", "db": db_session},
+        )
+        assert result.matched is False
+
+
+# ============================================================================
+# Section B — ExclusionService 测试（5个）
+# ============================================================================
+
+class TestExclusionService:
+    """ExclusionService 责任链匹配引擎测试。"""
+
+    @pytest.mark.asyncio
+    async def test_check_exclusions_no_rules(self, db_session: AsyncSession):
+        """无规则时返回未命中。"""
+        from app.services.exclusion_service import ExclusionService
+
+        service = ExclusionService()
+        result = await service.check_exclusions(
+            db=db_session,
+            message="测试消息",
+            conversation_id="conv-001",
+            user_id="user-001",
+        )
+        assert result.matched is False
+
+    @pytest.mark.asyncio
+    async def test_check_exclusions_priority_order(self, db_session: AsyncSession):
+        """优先级排序 P0 > P1 — P0 规则先匹配。"""
+        from app.services.exclusion_service import ExclusionService
+
+        # 创建两条规则，P1 和 P0，消息同时包含两个关键词
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="P1规则-密码",
+            priority="P1",
+            match_condition="密码",
+        )
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="P0规则-宕机",
+            priority="P0",
+            match_condition="宕机",
+        )
+        await db_session.flush()
+
+        service = ExclusionService()
+        result = await service.check_exclusions(
+            db=db_session,
+            message="系统宕机了，密码也忘了",
+            conversation_id="conv-002",
+            user_id="user-002",
+        )
+        assert result.matched is True
+        # P0 规则应先匹配
+        assert result.rule_name == "P0规则-宕机"
+
+    @pytest.mark.asyncio
+    async def test_check_exclusions_first_hit_stops(self, db_session: AsyncSession):
+        """命中即停止 — 只记录一条日志。"""
+        from app.services.exclusion_service import ExclusionService
+
+        # 创建两条都能匹配的规则
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="规则A-密码",
+            priority="P0",
+            match_condition="密码",
+        )
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="规则B-密码",
+            priority="P1",
+            match_condition="密码",
+        )
+        await db_session.flush()
+
+        service = ExclusionService()
+        result = await service.check_exclusions(
+            db=db_session,
+            message="密码过期了",
+            conversation_id="conv-003",
+            user_id="user-003",
+        )
+        assert result.matched is True
+        # 只有 P0 规则应命中
+        assert result.rule_name == "规则A-密码"
+
+        # 验证只创建了一条日志
+        log_result = await db_session.execute(select(ExclusionLog))
+        logs = log_result.scalars().all()
+        assert len(logs) == 1
+        assert logs[0].rule_name == "规则A-密码"
+
+    @pytest.mark.asyncio
+    async def test_check_exclusions_logs_hit(self, db_session: AsyncSession):
+        """命中记录日志 + 更新 hit_count。"""
+        from app.services.exclusion_service import ExclusionService
+
+        rule = await _create_exclusion_rule(
+            db_session,
+            rule_name="日志测试规则",
+            match_condition="密码过期",
+            hit_count=0,
+        )
+        await db_session.flush()
+
+        service = ExclusionService()
+        result = await service.check_exclusions(
+            db=db_session,
+            message="我的密码过期了",
+            conversation_id="conv-004",
+            user_id="user-004",
+        )
+        assert result.matched is True
+        assert result.rule_name == "日志测试规则"
+
+        # 验证 hit_count 已更新
+        await db_session.refresh(rule)
+        assert rule.hit_count == 1
+
+        # 验证日志已创建
+        log_result = await db_session.execute(
+            select(ExclusionLog).where(ExclusionLog.rule_id == rule.id)
+        )
+        logs = log_result.scalars().all()
+        assert len(logs) == 1
+        assert logs[0].action_type == "transfer_human"
+
+    @pytest.mark.asyncio
+    async def test_test_match_no_logging(self, db_session: AsyncSession):
+        """test_match 不记录日志、不更新 hit_count。"""
+        from app.services.exclusion_service import ExclusionService
+
+        rule = await _create_exclusion_rule(
+            db_session,
+            rule_name="测试匹配规则",
+            match_condition="密码过期",
+            hit_count=0,
+        )
+        await db_session.flush()
+
+        service = ExclusionService()
+        result = await service.test_match(
+            db=db_session,
+            message="我的密码过期了",
+            rule_id=rule.id,
+        )
+        assert result.matched is True
+
+        # 验证 hit_count 未更新
+        await db_session.refresh(rule)
+        assert rule.hit_count == 0
+
+        # 验证无日志创建
+        log_result = await db_session.execute(
+            select(ExclusionLog).where(ExclusionLog.rule_id == rule.id)
+        )
+        logs = log_result.scalars().all()
+        assert len(logs) == 0
+
+
+# ============================================================================
+# Section C — API 测试（8个）
+# ============================================================================
+
+class TestExclusionAPI:
+    """代答排除管理 API 测试。"""
+
+    @pytest.mark.asyncio
+    async def test_create_rule_success(self, client, db_session):
+        """新建规则成功。"""
+        token = await _login_admin(client, db_session)
+
+        resp = await client.post(
+            "/admin/exclusion-rules",
+            json={
+                "rule_name": "测试规则-新建",
+                "rule_description": "测试描述",
+                "priority": "P1",
+                "match_type": "keyword",
+                "match_condition": "密码,账号",
+                "match_scope": [],
+                "action_type": "transfer_human",
+                "transfer_message": "已转人工",
+            },
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        rule = data["data"]
+        assert rule["rule_name"] == "测试规则-新建"
+        assert rule["status"] == "enabled"
+        assert rule["hit_count"] == 0
+
+    @pytest.mark.asyncio
+    async def test_create_rule_duplicate_name(self, client, db_session):
+        """规则名重复返回 400。"""
+        token = await _login_admin(client, db_session)
+
+        # 先创建一条规则
+        await _create_exclusion_rule(db_session, rule_name="重复规则名")
+        await db_session.flush()
+
+        # 再用同名创建
+        resp = await client.post(
+            "/admin/exclusion-rules",
+            json={
+                "rule_name": "重复规则名",
+                "priority": "P2",
+                "match_type": "keyword",
+                "match_condition": "测试",
+                "action_type": "transfer_human",
+            },
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 400
+        assert "已存在" in data["message"]
+
+    @pytest.mark.asyncio
+    async def test_get_rule_detail(self, client, db_session):
+        """规则详情。"""
+        token = await _login_admin(client, db_session)
+
+        rule = await _create_exclusion_rule(
+            db_session,
+            rule_name="详情测试规则",
+            match_type="keyword",
+            match_condition="密码",
+        )
+        await db_session.flush()
+
+        resp = await client.get(
+            f"/admin/exclusion-rules/{rule.id}",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["id"] == rule.id
+        assert data["data"]["rule_name"] == "详情测试规则"
+
+    @pytest.mark.asyncio
+    async def test_update_rule(self, client, db_session):
+        """编辑规则。"""
+        token = await _login_admin(client, db_session)
+
+        rule = await _create_exclusion_rule(
+            db_session,
+            rule_name="编辑前规则",
+        )
+        await db_session.flush()
+
+        resp = await client.put(
+            f"/admin/exclusion-rules/{rule.id}",
+            json={
+                "rule_name": "编辑后规则",
+                "match_condition": "VPN,网络",
+            },
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["rule_name"] == "编辑后规则"
+        assert data["data"]["match_condition"] == "VPN,网络"
+
+    @pytest.mark.asyncio
+    async def test_delete_rule(self, client, db_session):
+        """删除规则。"""
+        token = await _login_admin(client, db_session)
+
+        rule = await _create_exclusion_rule(
+            db_session,
+            rule_name="待删除规则",
+        )
+        await db_session.flush()
+        rule_id = rule.id
+
+        resp = await client.delete(
+            f"/admin/exclusion-rules/{rule_id}",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+
+        # 验证已删除
+        resp = await client.get(
+            f"/admin/exclusion-rules/{rule_id}",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        assert resp.json()["code"] == 404
+
+    @pytest.mark.asyncio
+    async def test_toggle_rule(self, client, db_session):
+        """启用/停用规则。"""
+        token = await _login_admin(client, db_session)
+
+        rule = await _create_exclusion_rule(
+            db_session,
+            rule_name="切换状态规则",
+            status="enabled",
+        )
+        await db_session.flush()
+
+        # 停用
+        resp = await client.post(
+            f"/admin/exclusion-rules/{rule.id}/toggle",
+            json={"status": "disabled"},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["status"] == "disabled"
+
+        # 再启用
+        resp = await client.post(
+            f"/admin/exclusion-rules/{rule.id}/toggle",
+            json={"status": "enabled"},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        assert resp.json()["data"]["status"] == "enabled"
+
+    @pytest.mark.asyncio
+    async def test_test_match_api(self, client, db_session):
+        """测试匹配 API。"""
+        token = await _login_admin(client, db_session)
+
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="API测试匹配规则",
+            match_type="keyword",
+            match_condition="密码过期",
+        )
+        await db_session.flush()
+
+        resp = await client.post(
+            "/admin/exclusion-rules/test",
+            json={"message": "我的密码过期了"},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["matched"] is True
+        assert data["data"]["rule_name"] == "API测试匹配规则"
+
+    @pytest.mark.asyncio
+    async def test_get_stats_api(self, client, db_session):
+        """统计概要 API。"""
+        token = await _login_admin(client, db_session)
+
+        # 创建规则
+        await _create_exclusion_rule(db_session, status="enabled")
+        await _create_exclusion_rule(db_session, status="disabled")
+        await db_session.flush()
+
+        resp = await client.get(
+            "/admin/exclusion-rules/stats",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        stats = data["data"]
+        assert "enabled_count" in stats
+        assert "disabled_count" in stats
+        assert "monthly_hits" in stats
+        assert "monthly_transfers" in stats
+        assert stats["enabled_count"] == 1
+        assert stats["disabled_count"] == 1
+
+
+# ============================================================================
+# Section D — ai_handler 集成测试（3个）
+# ============================================================================
+
+class TestAIHandlerExclusion:
+    """AI 回复处理器与代答排除的集成测试。
+
+    验证 ai_handler.handle_message 在 AI 回复前检查排除规则：
+    1. 命中排除规则 → 拦截 AI 回复，返回排除结果
+    2. 未命中 → 正常调用 AI
+    3. 排除检查异常 → 降级继续 AI 回复（不阻断主流程）
+    """
+
+    @pytest.mark.asyncio
+    async def test_ai_handler_exclusion_hit(self, db_session: AsyncSession):
+        """命中排除规则 — AI 回复前拦截，不调用 AI 服务。"""
+        from app.services.ai_handler import AIHandler
+
+        # 创建排除规则（关键词 "密码" → transfer_human）
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="密码排除规则",
+            match_type="keyword",
+            match_condition="密码",
+            action_type="transfer_human",
+            transfer_message="此问题需转人工处理",
+        )
+        await db_session.flush()
+
+        # 创建 AIHandler with mock AIService
+        mock_ai_service = AsyncMock()
+        handler = AIHandler(ai_service=mock_ai_service)
+
+        result = await handler.handle_message(
+            content="我的密码过期了怎么办",
+            conversation_id="conv-ai-001",
+            user_id="user-001",
+            db=db_session,
+        )
+
+        # 验证：命中排除规则，返回 excluded 类型
+        assert result.reply_type == "excluded"
+        assert result.should_transfer is True
+        assert result.should_count is False
+        # AI 服务不应被调用
+        mock_ai_service.get_reply.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_ai_handler_exclusion_miss(self, db_session: AsyncSession):
+        """未命中排除规则 — 正常调用 AI 服务。"""
+        from app.services.ai_handler import AIHandler
+
+        # 创建排除规则（关键词 "密码"）
+        await _create_exclusion_rule(
+            db_session,
+            rule_name="密码排除规则",
+            match_type="keyword",
+            match_condition="密码",
+        )
+        await db_session.flush()
+
+        # 创建 AIHandler with mock AIService
+        mock_ai_service = AsyncMock()
+        mock_ai_service.get_reply.return_value = {
+            "hit": True,
+            "content": "建议您重启电脑试试",
+            "conversation_id": "dify-conv-001",
+        }
+        handler = AIHandler(ai_service=mock_ai_service)
+
+        # 消息不含 "密码" → 不命中排除规则 → 正常 AI 回复
+        result = await handler.handle_message(
+            content="打印机怎么连接",
+            conversation_id="conv-ai-002",
+            user_id="user-002",
+            db=db_session,
+        )
+
+        # 验证：正常 AI 回复
+        assert result.reply_type == "ai_hit"
+        assert result.should_count is True
+        assert "重启电脑" in result.content
+        # AI 服务应被调用
+        mock_ai_service.get_reply.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_ai_handler_exclusion_error(self, db_session: AsyncSession):
+        """排除检查异常不阻断主流程 — 降级继续 AI 回复。"""
+        from app.services.ai_handler import AIHandler
+
+        # 创建 AIHandler with mock AIService
+        mock_ai_service = AsyncMock()
+        mock_ai_service.get_reply.return_value = {
+            "hit": True,
+            "content": "AI降级回复",
+            "conversation_id": "dify-conv-002",
+        }
+        handler = AIHandler(ai_service=mock_ai_service)
+
+        # Patch check_exclusions 抛出异常
+        with patch(
+            "app.services.exclusion_service.get_exclusion_service"
+        ) as mock_get_service:
+            mock_service = AsyncMock()
+            mock_service.check_exclusions.side_effect = Exception("DB connection error")
+            mock_get_service.return_value = mock_service
+
+            result = await handler.handle_message(
+                content="打印机问题",
+                conversation_id="conv-ai-003",
+                user_id="user-003",
+                db=db_session,
+            )
+
+        # 验证：排除检查异常后降级继续 AI 回复
+        assert result.reply_type == "ai_hit"
+        assert result.should_count is True
+        # AI 服务应被调用（降级不阻断）
+        mock_ai_service.get_reply.assert_called_once()
@@ -0,0 +1,601 @@
+# -*- coding: utf-8 -*-
+"""分诊交互模块回归测试 — T02
+
+测试范围:
+  H5端接口（7个）: start_success, start_timeout, start_dify_unavailable,
+                   submit_step, skip_step, transfer, complete
+  坐席端接口（6个）: list_pending, get_stats, get_detail, route_session,
+                   get_history, exclude_options_ws_push
+  Service层（2个）: determine_urgency_keywords, determine_urgency_confidence
+
+测试依赖: conftest.py 提供的 client / db_session fixtures
+"""
+
+import asyncio
+import uuid
+from datetime import datetime
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.triage_session import TriageSession
+from app.services.triage_service import TriageService
+
+
+# ============================================================================
+# 辅助函数
+# ============================================================================
+
+async def _login(client: AsyncClient, db_session: AsyncSession, role: str = "agent") -> str:
+    """登录并返回 Bearer token。
+
+    Args:
+        client: 测试客户端
+        db_session: 数据库会话
+        role: 角色（agent/admin）
+
+    Returns:
+        str: Bearer token
+    """
+    from app.models.role import Role
+    from app.models.user_role import UserRole
+
+    user_id = f"test_{role}_{uuid.uuid4().hex[:8]}"
+
+    # 确保角色存在
+    stmt = select(Role).where(Role.name == role)
+    result = await db_session.execute(stmt)
+    db_role = result.scalars().first()
+    if not db_role:
+        display = "坐席" if role == "agent" else "管理员"
+        db_role = Role(
+            name=role, display_name=display,
+            description=f"{display}角色", permissions=[],
+        )
+        db_session.add(db_role)
+        await db_session.flush()
+
+    # 创建 UserRole 关联
+    db_session.add(UserRole(
+        employee_id=user_id,
+        role_id=db_role.id,
+        source="manual",
+        assigned_by="test_fixture",
+    ))
+    await db_session.flush()
+
+    # 登录
+    resp = await client.post("/agents/login", json={
+        "user_id": user_id,
+        "name": f"测试{role}",
+    })
+    return resp.json()["data"]["token"]
+
+
+async def _create_triage_session(db_session: AsyncSession, **kwargs) -> TriageSession:
+    """在数据库中创建分诊会话记录。
+
+    Args:
+        db_session: 数据库会话
+        **kwargs: 覆盖默认字段值
+
+    Returns:
+        TriageSession: 创建的会话对象
+    """
+    defaults = {
+        "conversation_id": f"conv-{uuid.uuid4().hex[:8]}",
+        "user_id": "test_user_001",
+        "user_name": "测试用户",
+        "user_dept": "技术部",
+        "request_title": "测试问题标题",
+        "request_content": "测试问题内容",
+        "source": "wecom_h5",
+        "status": "pending",
+        "urgency": "medium",
+    }
+    defaults.update(kwargs)
+    session = TriageSession(**defaults)
+    db_session.add(session)
+    await db_session.flush()
+    return session
+
+
+# ============================================================================
+# Fixture: Mock Dify 分诊服务
+# ============================================================================
+
+@pytest.fixture
+def mock_dify_triage():
+    """Mock Dify triage service on the singleton TriageService。
+
+    TriageService 是单例，dify_service 在 __init__ 中赋值。
+    此 fixture 替换 dify_service 为 AsyncMock，测试后恢复原值。
+    """
+    from app.services.triage_service import get_triage_service
+    service = get_triage_service()
+    original = service.dify_service
+    mock = AsyncMock()
+    service.dify_service = mock
+    yield mock
+    service.dify_service = original
+
+
+# ============================================================================
+# Section A — H5 端接口测试（7个）
+# ============================================================================
+
+class TestH5Triage:
+    """H5 端分诊交互接口测试。"""
+
+    @pytest.mark.asyncio
+    async def test_start_triage_success(self, client, db_session, mock_dify_triage):
+        """正常发起分诊 — code=0, 返回 triage_id/steps/confidence/urgency。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.85,
+            "urgency": "medium",
+            "suggested_route": "ai_self",
+            "problem_type": "软件",
+            "problem_category": "Outlook",
+            "matched_knowledge": "FAQ-001",
+            "match_score": 0.92,
+            "context_tags": ["email"],
+            "triage_steps": [
+                {
+                    "question": "您遇到的问题是？",
+                    "options": [
+                        {"label": "无法登录", "probability": 0.7},
+                        {"label": "邮件发不出", "probability": 0.3},
+                    ],
+                }
+            ],
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-test-001",
+            "question": "我的Outlook打不开了",
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        result = data["data"]
+        assert "triage_id" in result
+        assert len(result["steps"]) == 1
+        assert result["confidence"] == 0.85
+        # "打不开" 是中级关键词 → medium
+        assert result["urgency"] == "medium"
+
+    @pytest.mark.asyncio
+    async def test_start_triage_timeout(self, client, db_session, mock_dify_triage):
+        """Dify 超时 — status=timeout, 自动转人工。"""
+        token = await _login(client, db_session)
+
+        # 模拟 Dify 超时（asyncio.wait_for 捕获 TimeoutError）
+        mock_dify_triage.analyze.side_effect = asyncio.TimeoutError()
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-test-002",
+            "question": "密码过期了怎么办",
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["status"] == "timeout"
+        assert "triage_id" in data["data"]
+
+    @pytest.mark.asyncio
+    async def test_start_triage_dify_unavailable(self, client, db_session, mock_dify_triage):
+        """Dify 不可用 — 降级转人工。"""
+        token = await _login(client, db_session)
+
+        # 模拟 Dify 不可用（RuntimeError 触发降级转人工）
+        mock_dify_triage.analyze.side_effect = RuntimeError("Dify unavailable")
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-test-003",
+            "question": "VPN连不上了",
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["status"] == "timeout"
+        assert "triage_id" in data["data"]
+
+    @pytest.mark.asyncio
+    async def test_submit_step_success(self, client, db_session, mock_dify_triage):
+        """提交步骤选择 — 返回 next_step 和 collected_context。"""
+        token = await _login(client, db_session)
+
+        # 先发起分诊
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.8,
+            "urgency": "low",
+            "triage_steps": [
+                {"question": "问题1", "options": [{"label": "选项A", "probability": 0.6}]},
+                {"question": "问题2", "options": [{"label": "选项B", "probability": 0.5}]},
+            ],
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-step-001",
+            "question": "打印机问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        # 提交步骤0的选择
+        resp = await client.post("/h5/triage/step", json={
+            "triage_id": triage_id,
+            "step_index": 0,
+            "selected_label": "选项A",
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["next_step"] is not None
+        assert data["data"]["next_step"]["question"] == "问题2"
+        assert "选项A" in data["data"]["collected_context"]
+
+    @pytest.mark.asyncio
+    async def test_skip_step_success(self, client, db_session, mock_dify_triage):
+        """跳过步骤 — 返回 next_step。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.7,
+            "urgency": "low",
+            "triage_steps": [
+                {"question": "问题1", "options": [{"label": "选项A", "probability": 0.6}]},
+                {"question": "问题2", "options": [{"label": "选项B", "probability": 0.5}]},
+            ],
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-skip-001",
+            "question": "网络问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        resp = await client.post("/h5/triage/skip", json={
+            "triage_id": triage_id,
+            "step_index": 0,
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["next_step"] is not None
+        assert data["data"]["next_step"]["question"] == "问题2"
+
+    @pytest.mark.asyncio
+    async def test_transfer_to_human_success(self, client, db_session, mock_dify_triage):
+        """转人工 — status=waiting_agent。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.6,
+            "urgency": "low",
+            "triage_steps": [{"question": "问题1", "options": []}],
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-transfer-001",
+            "question": "硬件问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        resp = await client.post("/h5/triage/transfer", json={
+            "triage_id": triage_id,
+            "context": ["用户选择的上下文"],
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["status"] == "waiting_agent"
+        assert "conversation_id" in data["data"]
+
+    @pytest.mark.asyncio
+    async def test_complete_triage_success(self, client, db_session, mock_dify_triage):
+        """分诊完成 — 返回 reply 和 confidence。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.9,
+            "urgency": "low",
+            "triage_steps": [
+                {"question": "问题1", "options": [{"label": "选项A", "probability": 0.8}]}
+            ],
+        }
+        mock_dify_triage.generate_reply.return_value = {
+            "reply": "建议您重启Outlook客户端。",
+            "confidence": 0.88,
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-complete-001",
+            "question": "软件使用问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        resp = await client.post("/h5/triage/complete", json={
+            "triage_id": triage_id,
+            "context": ["选项A"],
+        }, headers={"Authorization": f"Bearer {token}"})
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert "reply" in data["data"]
+        assert data["data"]["confidence"] == 0.88
+
+
+# ============================================================================
+# Section B — 坐席端接口测试（6个）
+# ============================================================================
+
+class TestAgentTriage:
+    """坐席端分诊看板接口测试。"""
+
+    @pytest.mark.asyncio
+    async def test_list_pending_sorted_by_urgency(self, client, db_session):
+        """待分诊列表按紧急度排序 high > medium > low。"""
+        token = await _login(client, db_session)
+
+        # 创建3条不同紧急度的待分诊记录（创建顺序故意打乱）
+        await _create_triage_session(db_session, urgency="low", request_title="低优先级")
+        await _create_triage_session(db_session, urgency="high", request_title="高优先级")
+        await _create_triage_session(db_session, urgency="medium", request_title="中优先级")
+        await db_session.flush()
+
+        resp = await client.get(
+            "/agent/triage/pending",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        items = data["data"]["items"]
+        assert len(items) == 3
+        # high 应排在最前
+        assert items[0]["urgency"] == "high"
+        assert items[1]["urgency"] == "medium"
+        assert items[2]["urgency"] == "low"
+
+    @pytest.mark.asyncio
+    async def test_get_stats(self, client, db_session):
+        """统计概要返回6项指标。"""
+        token = await _login(client, db_session)
+
+        # 创建测试数据
+        await _create_triage_session(db_session, status="pending", urgency="high")
+        await _create_triage_session(db_session, status="triaging", urgency="medium")
+        await _create_triage_session(
+            db_session, status="routed", route_action="ai_self",
+            operated_at=datetime.now(),
+        )
+        await db_session.flush()
+
+        resp = await client.get(
+            "/agent/triage/stats",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        stats = data["data"]
+        # 验证6项指标字段都存在
+        assert "pending_total" in stats
+        assert "today_triaged" in stats
+        assert "ai_self_count" in stats
+        assert "human_count" in stats
+        assert "auto_approval_count" in stats
+        assert "avg_duration_sec" in stats
+        # 验证待分诊数（1 pending + 1 triaging = 2）
+        assert stats["pending_total"] == 2
+
+    @pytest.mark.asyncio
+    async def test_get_detail(self, client, db_session, mock_dify_triage):
+        """获取分诊详情。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.85,
+            "urgency": "medium",
+            "triage_steps": [{"question": "问题1", "options": []}],
+            "problem_type": "软件",
+            "problem_category": "Outlook",
+        }
+
+        # 发起分诊创建会话
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-detail-001",
+            "question": "Outlook问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        # 获取详情
+        resp = await client.get(
+            f"/agent/triage/{triage_id}",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        detail = data["data"]
+        assert detail["id"] == triage_id
+        assert detail["problem_category"] == "Outlook"
+        assert detail["confidence"] == 0.85
+
+    @pytest.mark.asyncio
+    async def test_route_session(self, client, db_session, mock_dify_triage):
+        """坐席路由操作覆盖 AI 建议。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.7,
+            "urgency": "low",
+            "triage_steps": [{"question": "问题1", "options": []}],
+            "suggested_route": "ai_self",
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-route-001",
+            "question": "一般问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        # 坐席路由为转人工（覆盖AI建议的ai_self）
+        resp = await client.post(
+            f"/agent/triage/{triage_id}/route",
+            json={
+                "route_action": "human",
+                "route_note": "需要人工排查",
+            },
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        assert data["data"]["route_action"] == "human"
+        assert data["data"]["status"] == "routed"
+        assert data["data"]["route_note"] == "需要人工排查"
+
+    @pytest.mark.asyncio
+    async def test_get_history(self, client, db_session):
+        """历史列表返回 routed/skipped/timeout 状态的记录。"""
+        token = await _login(client, db_session)
+
+        # 创建历史记录
+        await _create_triage_session(db_session, status="routed", route_action="ai_self")
+        await _create_triage_session(db_session, status="routed", route_action="human")
+        await _create_triage_session(db_session, status="skipped")
+        # pending 不应出现在历史中
+        await _create_triage_session(db_session, status="pending")
+        await db_session.flush()
+
+        resp = await client.get(
+            "/agent/triage/history",
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["code"] == 0
+        items = data["data"]["items"]
+        assert len(items) == 3  # 只有 routed/skipped
+        for item in items:
+            assert item["status"] in ("routed", "skipped", "timeout")
+
+    @pytest.mark.asyncio
+    async def test_exclude_options_ws_push(self, client, db_session, mock_dify_triage):
+        """排除选项通过 WS 推送到 H5。"""
+        token = await _login(client, db_session)
+
+        mock_dify_triage.analyze.return_value = {
+            "confidence": 0.8,
+            "urgency": "low",
+            "triage_steps": [{"question": "问题1", "options": [
+                {"label": "选项A", "probability": 0.5},
+                {"label": "选项B", "probability": 0.3},
+            ]}],
+        }
+
+        resp = await client.post("/h5/triage/start", json={
+            "conversation_id": "conv-exclude-001",
+            "question": "测试问题",
+        }, headers={"Authorization": f"Bearer {token}"})
+        triage_id = resp.json()["data"]["triage_id"]
+
+        # Mock WS manager 的 send_to_employee 方法
+        with patch(
+            "app.services.ws_manager.manager.send_to_employee",
+            new_callable=AsyncMock,
+        ) as mock_ws:
+            resp = await client.post(
+                f"/agent/triage/{triage_id}/exclude-options",
+                json={
+                    "excluded_labels": ["选项A"],
+                    "recommended_label": "选项B",
+                },
+                headers={"Authorization": f"Bearer {token}"},
+            )
+
+            assert resp.status_code == 200
+            data = resp.json()
+            assert data["code"] == 0
+            assert data["data"]["excluded"] is True
+            # 验证 WS 推送被调用
+            mock_ws.assert_called_once()
+            # 验证推送数据格式
+            call_args = mock_ws.call_args
+            ws_data = call_args[0][1]  # 第二个位置参数
+            assert ws_data["type"] == "triage_exclude"
+            assert "选项A" in ws_data["data"]["excluded_labels"]
+            assert ws_data["data"]["recommended_label"] == "选项B"
+
+
+# ============================================================================
+# Section C — Service 层测试（2个）
+# ============================================================================
+
+class TestTriageService:
+    """TriageService 业务逻辑测试。"""
+
+    def test_determine_urgency_keywords(self):
+        """紧急度判断关键词规则。
+
+        规则：
+        - 高级关键词（紧急/宕机/崩溃等）→ high
+        - 中级关键词（报错/失败/连不上等）→ medium
+        - 无关键词 → low
+        """
+        # 高级关键词 → high
+        assert TriageService.determine_urgency("系统宕机了") == "high"
+        assert TriageService.determine_urgency("紧急！密码过期") == "high"
+        assert TriageService.determine_urgency("电脑蓝屏了") == "high"
+        assert TriageService.determine_urgency("系统崩溃了") == "high"
+
+        # 中级关键词 → medium
+        assert TriageService.determine_urgency("VPN连不上") == "medium"
+        assert TriageService.determine_urgency("打印机报错") == "medium"
+        assert TriageService.determine_urgency("登录失败") == "medium"
+        assert TriageService.determine_urgency("页面打不开") == "medium"
+
+        # 无关键词 → low
+        assert TriageService.determine_urgency("我想查一下工资条") == "low"
+        assert TriageService.determine_urgency("请问年假怎么申请") == "low"
+
+    def test_determine_urgency_confidence(self):
+        """置信度低于 0.5 为 high。
+
+        规则：
+        - confidence < 0.5 → high（即使没有关键词）
+        - 高级关键词始终优先于置信度
+        - 置信度优先于中级关键词
+        """
+        # 置信度 < 0.5 → high（即使没有关键词）
+        assert TriageService.determine_urgency("一般问题", confidence=0.3) == "high"
+        assert TriageService.determine_urgency("普通咨询", confidence=0.49) == "high"
+
+        # 置信度 >= 0.5 且无关键词 → low
+        assert TriageService.determine_urgency("一般问题", confidence=0.5) == "low"
+        assert TriageService.determine_urgency("普通咨询", confidence=0.9) == "low"
+
+        # 置信度 < 0.5 但有中级关键词 → high（置信度优先于中级关键词）
+        assert TriageService.determine_urgency("VPN连不上", confidence=0.3) == "high"
+
+        # 高级关键词始终优先（即使置信度很高）
+        assert TriageService.determine_urgency("宕机", confidence=0.9) == "high"
+        assert TriageService.determine_urgency("宕机", confidence=0.1) == "high"