补充调整

HA 的 brightness 量程是 0～255，原样交给模型会被念成"一百五十三"；灯关着时是 null，还会报"亮度是:None"。换算成百分比，和设置亮度用的 brightness_pct 同一个单位，开着的灯最低报 1%（同 HA 前端）；不是数字时不报这一项。附测试。

.github/workflows/test.yml

@@ -20,24 +20,58 @@ jobs:
           python-version: '3.10'
           cache: 'pip'
           cache-dependency-path: 'main/xiaozhi-server/requirements.txt'
+      - name: Install system dependencies
+        run: |
+          # opuslib_next 是 ctypes 绑定，pip 装上的是纯 Python 包，真正的编解码
+          # 靠系统里的 libopus.so。runner 上默认没有，import 会失败，音频相关的
+          # 用例直接报错。正式镜像 Dockerfile-server-base 里也是装 libopus0 的
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends libopus0
       - name: Install dependencies
         working-directory: main/xiaozhi-server
         run: |
           pip install -r requirements.txt
           pip install pytest pytest-asyncio freezegun
-      - name: Run smoke test
+      - name: Prepare test config
         working-directory: main/xiaozhi-server
         run: |
-          if [ -f tests/test_smoke.py ]; then
+          # config/settings.py 在 import 期就要这个文件，缺了整套用例连收集都过不去。
-            pytest tests/test_smoke.py -x -q
+          # data/ 在 .gitignore 里，只能由这一步现造
-          else
+          mkdir -p data
-            echo "::warning::Slice 2 not done yet — tests/test_smoke.py missing"
+          printf '{}\n' > data/.config.yaml
-            exit 0
+      - name: Run tests
-          fi
+        working-directory: main/xiaozhi-server
+        run: |
+          # Edge TTS 那条要连微软公网，runner 的出口 IP 会被风控挡掉，先排除。
+          # 根治办法是给它打 live 标记，让 mock 模式下自动不收（另开一条改）
+          pytest tests/ -v -k "not test_tts_text_to_speak_writes_file"
 
   java:
     name: Java (manager-api)
     runs-on: ubuntu-latest
+    # 三个 @SpringBootTest 要真连 MySQL 和 Redis，空库即可，Liquibase 会自己建表
+    services:
+      mysql:
+        image: mysql:8.0
+        env:
+          MYSQL_ROOT_PASSWORD: 123456
+          MYSQL_DATABASE: xiaozhi_esp32_server
+        ports:
+          - 3306:3306
+        options: >-
+          --health-cmd="mysqladmin ping -h 127.0.0.1 -uroot -p123456"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=20
+      redis:
+        image: redis:8.0
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd="redis-cli ping"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=20
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-java@v4
@@ -47,7 +81,12 @@ jobs:
           cache: 'maven'
       - name: Run tests
         working-directory: main/manager-api
-        run: mvn -B -q test -DfailIfNoTests=false
+        run: mvn -B test -DskipTests=false -DfailIfNoTests=false
+      - name: Check the integration tests really ran
+        working-directory: main/manager-api
+        # 守卫在依赖不齐时会把整类集成测试跳过，这里盯一下它们有没有真跑，
+        # 免得门禁被悄悄掏空
+        run: bash scripts/check-integration-tests.sh
 
   web:
     name: Vue (manager-web)
@@ -63,4 +102,6 @@ jobs:
         working-directory: main/manager-web
         run: |
           npm ci --no-audit --no-fund
+          npm run test:contract
           npm run test:unit
+          npm run build

docs/device-call-guide.md

@@ -144,5 +144,7 @@
 ### Q: 我叫他呼叫"张山"，但是他老是识别成"张三"，怎么办？
 - 可以查阅你使用的asr服务的文档，确认是否支持热词识别。
 - 如果你用的是`FunASRServer`,可以在容器里的`热词文件`里添加"张山"，然后重启容器。
-- 如果你用的是`火山引擎`的服务，可以在`火山引擎的控制台`里添加`热词文件`，然后回到智控台的`模型配置页面`，把`热词文件名称`配置在`火山引擎的tts`上去。
+- 如果你用的是`火山引擎`的服务，可以在`火山引擎的控制台`里添加`热词文件`，然后回到智控台的`模型配置`页面，在`语音识别`里找到模型名称为`豆包语音识别(流式)`的那一条，点`修改`，弹窗里的`供应器`是`火山引擎语音识别(流式)`，在弹窗的`调用信息`区域把`热词文件名称`填到对应的输入框，点`确认保存`。
+- `豆包语音识别2.0(流式)`（供应器是`火山引擎语音识别(流式)`）和`豆包语音识别`（供应器是`火山引擎语音识别`）这两条也都有`热词文件名称`这个字段，用法一样，配置时注意别选错行。
+- 改完要让设备重新连接才会生效：设备连上时才会把最新的私有配置拉下来，一直连着的设备不会自动变。
 

docs/web-search-integration.md

@@ -2,14 +2,15 @@
 
 ## 功能简介
 
-联网搜索插件 `web_search` 支持在对话过程中实时联网搜索信息并返回结果。插件支持三个搜索源：秘塔（Metaso）、Tavily和Serply，用户可根据需要选择其中一个。
+联网搜索插件 `web_search` 支持在对话过程中实时联网搜索信息并返回结果。插件支持四个搜索源：秘塔（Metaso）、Tavily、Serply和Outcrawl，用户可根据需要选择其中一个。
 
 ## API Key申请指南
 
-目前我们适配了`秘塔搜索`、`Tavily搜索`和`Serply搜索`。
+目前我们适配了`秘塔搜索`、`Tavily搜索`、`Serply搜索`和`Outcrawl搜索`。
 - Tavily搜索：每个月1000次免费额度。
 - 秘塔搜索：拥有较为优质的国内数据源。
 - Serply搜索：返回谷歌搜索结果，注册赠送2500次免费额度，额度用完后需要付费。
+- Outcrawl搜索：面向AI智能体的网页搜索接口，每月1000 credits免费额度。
 
 ## API Key申请指南
 
@@ -32,6 +33,13 @@
 - 复制生成的API Key，这是配置所需的关键信息
 - 接口说明见 [Serply文档](https://serply.io/docs)
 
+### 方式四：使用Outcrawl搜索
+
+- 访问 [Outcrawl](https://outcrawl.ai)，注册并登录账号
+- 在控制台中创建API Key
+- 复制生成的API Key，这是配置所需的关键信息
+- 接口说明见 [Outcrawl文档](https://outcrawl.ai/docs/search/overview)
+
 ## 配置方式
 
 ### 方式1. 使用智控台部署（推荐）
@@ -40,14 +48,14 @@
 - 进入"配置角色"页面，选择要配置的智能体
 - 点击"编辑功能"按钮，在右侧参数配置区域找到"联网搜索"插件
 - 勾选"联网搜索"
-- 填入搜索源（`metaso`、`tavily`或`serply`），并将对应的`API Key`填入配置项
+- 填入搜索源（`metaso`、`tavily`、`serply`或`outcrawl`），并将对应的`API Key`填入配置项
 - 保存配置，再保存智能体配置
 
 ### 方式2. 单模块xiaozhi-server部署
 
 在 `data/.config.yaml` 中配置：
 
-- 将搜索源填入 `provider`，可选值为 `metaso`、`tavily` 或 `serply`
+- 将搜索源填入 `provider`，可选值为 `metaso`、`tavily`、`serply` 或 `outcrawl`
 - 将申请到的API Key填入 `api_key`
 
 ```yaml

main/manager-api/README.md

@@ -16,3 +16,28 @@ Vue 3.x
 # 接口文档
 启动后打开：http://localhost:8002/xiaozhi/doc.html
 
+# 跑测试
+测试分两种：纯单元测试，和要真连 MySQL、Redis 的集成测试（`DeviceTest`、`loginControllerTest`、`ModelConfigPersistenceTest`）。
+
+集成测试需要先起好 MySQL 8 和 Redis 8。没起的时候这类测试整类跳过，不会让构建变红，日志里会写明缺的是哪个依赖。CI 上不一样：依赖连不上就直接让构建失败，不许跳过，免得门禁被悄悄掏空。
+
+本地用 Docker 起这两个依赖：
+
+```bash
+docker run -d --name xiaozhi-mysql -p 3306:3306 \
+  -e MYSQL_ROOT_PASSWORD=123456 -e MYSQL_DATABASE=xiaozhi_esp32_server mysql:8.0
+docker run -d --name xiaozhi-redis -p 6379:6379 redis:8.0
+```
+
+跑全部测试：
+
+```bash
+mvn -B test -DskipTests=false
+```
+
+只想编译、不跑测试：
+
+```bash
+mvn -B package -DskipTests=true
+```
+

main/manager-api/pom.xml

@@ -32,7 +32,7 @@
         <liquibase-core.version>4.20.0</liquibase-core.version>
         <aliyun-sms-version>4.1.0</aliyun-sms-version>
         <okio-version>3.4.0</okio-version>
-        <skipTests>true</skipTests>
+        <skipTests>false</skipTests>
         <argLine></argLine>
     </properties>
 

main/manager-api/scripts/check-integration-tests.sh

@@ -0,0 +1,44 @@
+#!/usr/bin/env bash
+# 检查那几个要连 MySQL 和 Redis 的集成测试是不是真的跑了。
+# 守卫要是哪天手滑把它们全跳过了，CI 必须红，不能安安静静少跑一半门禁。
+set -uo pipefail
+
+REPORT_DIR="${1:-target/surefire-reports}"
+
+WATCHED=(
+  "xiaozhi.modules.device.DeviceTest"
+  "xiaozhi.modules.model.service.impl.ModelConfigPersistenceTest"
+  "xiaozhi.modules.sys.loginControllerTest"
+)
+
+if [ ! -d "$REPORT_DIR" ]; then
+  echo "找不到测试报告目录：$REPORT_DIR"
+  exit 1
+fi
+
+missing=()
+for class in "${WATCHED[@]}"; do
+  count=0
+  for report in "$REPORT_DIR"/TEST-*.xml; do
+    [ -e "$report" ] || continue
+    name=$(sed -n 's/.*<testsuite[^>]*name="\([^"]*\)".*/\1/p' "$report" | head -1)
+    if [ "$name" = "$class" ]; then
+      count=$(sed -n 's/.*<testsuite[^>]*tests="\([0-9]*\)".*/\1/p' "$report" | head -1)
+      count=${count:-0}
+      break
+    fi
+  done
+  if [ "$count" -eq 0 ]; then
+    missing+=("$class")
+  else
+    echo "OK  $class 跑了 $count 条"
+  fi
+done
+
+if [ ${#missing[@]} -ne 0 ]; then
+  echo "这几个集成测试一条都没跑：${missing[*]}"
+  echo "守卫把门禁掏空了，构建必须失败"
+  exit 1
+fi
+
+echo "集成测试都跑过了"

main/manager-api/src/main/resources/db/changelog/202610302051.sql

@@ -0,0 +1,22 @@
+-- PaddleSpeechTTS：智控台协议收敛为仅 websocket（与 xiaozhi-server 实现一致）
+
+UPDATE `ai_model_provider`
+SET `fields` = JSON_ARRAY(
+    JSON_OBJECT('key', 'protocol', 'label', '协议类型', 'type', 'string', 'options', JSON_ARRAY('websocket')),
+    JSON_OBJECT('key', 'url', 'label', '服务地址', 'type', 'string'),
+    JSON_OBJECT('key', 'spk_id', 'label', '音色', 'type', 'int'),
+    JSON_OBJECT('key', 'sample_rate', 'label', '采样率', 'type', 'float'),
+    JSON_OBJECT('key', 'speed', 'label', '语速', 'type', 'float'),
+    JSON_OBJECT('key', 'volume', 'label', '音量', 'type', 'float'),
+    JSON_OBJECT('key', 'save_path', 'label', '保存路径', 'type', 'string')
+)
+WHERE `id` = 'SYSTEM_TTS_PaddleSpeechTTS';
+
+UPDATE `ai_model_config`
+SET `remark` = 'PaddleSpeechTTS 配置说明：
+1. PaddleSpeech 是百度飞桨开源的语音合成工具，支持本地离线部署和模型训练。paddlepaddle百度飞浆框架地址：https://www.paddlepaddle.org.cn/
+2. 当前仅支持 WebSocket 协议进行流式传输（参考部署文档：https://github.com/xinnan-tech/xiaozhi-esp32-server/blob/main/docs/paddlespeech-deploy.md）。
+3. 使用前要在本地部署 paddlespeech 服务，服务默认运行在 ws://127.0.0.1:8092/paddlespeech/tts/streaming
+4. 支持自定义发音人、语速、音量和采样率。
+'
+WHERE `id` = 'TTS_PaddleSpeechTTS';

main/manager-api/src/main/resources/db/changelog/202611021200.sql

@@ -0,0 +1,9 @@
+-- 联网搜索插件新增Outcrawl搜索源
+
+UPDATE `ai_model_provider`
+SET `fields` = JSON_SET(
+    `fields`,
+    '$[0].label',
+    '搜索源：metaso / tavily / serply / outcrawl'
+)
+WHERE `id` = 'SYSTEM_PLUGIN_WEB_SEARCH';

main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -753,3 +753,17 @@ databaseChangeLog:
         - sqlFile:
             encoding: utf8
             path: classpath:db/changelog/202609172100.sql
+  - changeSet:
+      id: 202610302051
+      author: shinjiyu
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202610302051.sql
+  - changeSet:
+      id: 202611021200
+      author: heydryft
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202611021200.sql

main/manager-api/src/test/java/xiaozhi/modules/correctword/service/impl/CorrectWordFileServiceImplTest.java

@@ -14,9 +14,14 @@ import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+
 import xiaozhi.modules.agent.dao.AgentCorrectWordMappingDao;
 import xiaozhi.modules.agent.entity.AgentCorrectWordMappingEntity;
 import xiaozhi.modules.correctword.dao.CorrectWordFileDao;
@@ -37,6 +42,20 @@ class CorrectWordFileServiceImplTest {
     private final CorrectWordFileServiceImpl service =
             new CorrectWordFileServiceImpl(fileDao, itemDao, mappingDao);
 
+    @BeforeAll
+    static void registerEntityMetadata() {
+        // 这条用例不起 Spring 上下文，MyBatis-Plus 的实体元数据没人注册，
+        // LambdaQueryWrapper 拼条件时找不到 lambda 缓存就抛异常。
+        // 全量跑的时候是前面的上下文测试顺带注册好的，单独跑就露馅。
+        // 这里自己补上，单独跑和跟着全量跑结果一样。
+        TableInfoHelper.initTableInfo(
+                new MapperBuilderAssistant(new MybatisConfiguration(), ""),
+                CorrectWordItemEntity.class);
+        TableInfoHelper.initTableInfo(
+                new MapperBuilderAssistant(new MybatisConfiguration(), ""),
+                CorrectWordFileEntity.class);
+    }
+
     @Test
     void getAllItemsByAgentId_returns_empty_list_when_no_mappings() {
         when(mappingDao.selectByAgentId(AGENT_ID)).thenReturn(Collections.emptyList());

main/manager-api/src/test/java/xiaozhi/modules/device/DeviceTest.java

@@ -3,6 +3,7 @@ package xiaozhi.modules.device;
 import java.util.HashMap;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -15,6 +16,7 @@ import xiaozhi.common.exception.RenException;
 import xiaozhi.common.redis.RedisUtils;
 import xiaozhi.modules.sys.dto.SysUserDTO;
 import xiaozhi.modules.sys.service.SysUserService;
+import xiaozhi.testsupport.IntegrationTestEnvironment;
 
 @Slf4j
 @SpringBootTest
@@ -22,6 +24,11 @@ import xiaozhi.modules.sys.service.SysUserService;
 @DisplayName("设备测试")
 public class DeviceTest {
 
+    @BeforeAll
+    static void checkDependencies() {
+        IntegrationTestEnvironment.requireDatabaseAndRedis();
+    }
+
     @Autowired
     private RedisUtils redisUtils;
     @Autowired

main/manager-api/src/test/java/xiaozhi/modules/model/service/impl/ModelConfigPersistenceTest.java

@@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
 
 import java.util.UUID;
 
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -19,12 +20,18 @@ import xiaozhi.common.redis.RedisUtils;
 import xiaozhi.modules.model.dto.ModelConfigBodyDTO;
 import xiaozhi.modules.model.dto.ModelConfigDTO;
 import xiaozhi.modules.model.service.ModelConfigService;
+import xiaozhi.testsupport.IntegrationTestEnvironment;
 
 @SpringBootTest
 @ActiveProfiles("dev")
 @Transactional
 class ModelConfigPersistenceTest {
 
+    @BeforeAll
+    static void checkDependencies() {
+        IntegrationTestEnvironment.requireDatabaseAndRedis();
+    }
+
     @Autowired
     private ModelConfigService service;
 

main/manager-api/src/test/java/xiaozhi/modules/sys/loginControllerTest.java

@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.when;
 
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -18,12 +19,18 @@ import xiaozhi.modules.security.dto.LoginDTO;
 import xiaozhi.modules.security.dto.SmsVerificationDTO;
 import xiaozhi.modules.sys.dto.RetrievePasswordDTO;
 import xiaozhi.modules.sys.service.SysUserService;
+import xiaozhi.testsupport.IntegrationTestEnvironment;
 
 @Slf4j
 @SpringBootTest
 @ActiveProfiles("dev")
 class loginControllerTest {
 
+    @BeforeAll
+    static void checkDependencies() {
+        IntegrationTestEnvironment.requireDatabaseAndRedis();
+    }
+
     @Autowired
     LoginController loginController;
 

main/manager-api/src/test/java/xiaozhi/testsupport/IntegrationTestEnvironment.java

@@ -0,0 +1,149 @@
+package xiaozhi.testsupport;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.time.Duration;
+
+import org.junit.jupiter.api.Assumptions;
+import org.springframework.boot.env.YamlPropertySourceLoader;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.PropertySource;
+import org.springframework.core.env.PropertySourcesPropertyResolver;
+import org.springframework.core.env.StandardEnvironment;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.StatefulRedisConnection;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 几个 @SpringBootTest 要真连 MySQL 和 Redis。本地没起这两个依赖时，
+ * 整类测试跳过，别让 Spring 上下文起不来变成一堆看不懂的报错；
+ * CI 里依赖必须真的在，探不通就直接让构建红。
+ */
+@Slf4j
+public final class IntegrationTestEnvironment {
+
+    /** 集成测试用的配置文件，跟应用本身读的是同一套 */
+    private static final String[] CONFIG_FILES = {"application.yml", "application-dev.yml"};
+
+    private static Boolean available;
+    private static String problem;
+
+    private IntegrationTestEnvironment() {
+    }
+
+    /**
+     * 放在要连库的测试类的 @BeforeAll 里。
+     * 本地依赖不齐就整类跳过，CI 里依赖不齐就直接失败。
+     */
+    public static void requireDatabaseAndRedis() {
+        String issue = probe();
+        if (issue == null) {
+            return;
+        }
+        if (isCi()) {
+            throw new IllegalStateException(
+                    "CI 环境里" + issue + "，这几个集成测试必须真的跑起来，不能跳过");
+        }
+        log.warn("跳过依赖 MySQL 和 Redis 的集成测试：{}", issue);
+        Assumptions.assumeTrue(false, issue);
+    }
+
+    /** GitHub Actions 上会带 CI=true */
+    public static boolean isCi() {
+        String value = System.getenv("CI");
+        if (value == null) {
+            return false;
+        }
+        String trimmed = value.trim();
+        return !trimmed.isEmpty() && !"false".equalsIgnoreCase(trimmed);
+    }
+
+    /** 依赖都齐返回 null，不齐返回缺什么 */
+    static synchronized String probe() {
+        if (available != null) {
+            return problem;
+        }
+        PropertySourcesPropertyResolver resolver = resolver();
+        StringBuilder issues = new StringBuilder();
+
+        String url = firstNonBlank(resolver.getProperty("spring.datasource.druid.url", ""),
+                resolver.getProperty("spring.datasource.url", ""));
+        String username = firstNonBlank(resolver.getProperty("spring.datasource.druid.username", ""),
+                resolver.getProperty("spring.datasource.username", ""));
+        String password = firstNonBlank(resolver.getProperty("spring.datasource.druid.password", ""),
+                resolver.getProperty("spring.datasource.password", ""));
+        if (url.isBlank()) {
+            issues.append("配置里找不到数据源地址；");
+        } else {
+            try {
+                DriverManager.setLoginTimeout(5);
+                try (Connection connection = DriverManager.getConnection(url, username, password)) {
+                    connection.isValid(3);
+                }
+            } catch (Exception e) {
+                issues.append("连不上 MySQL（").append(e.getMessage()).append("）；");
+            }
+        }
+
+        String redisHost = resolver.getProperty("spring.data.redis.host", "localhost");
+        int redisPort = resolver.getProperty("spring.data.redis.port", Integer.class, 6379);
+        String redisPassword = resolver.getProperty("spring.data.redis.password", "");
+        try {
+            RedisURI.Builder builder = RedisURI.builder()
+                    .withHost(redisHost)
+                    .withPort(redisPort)
+                    .withTimeout(Duration.ofSeconds(3));
+            if (redisPassword != null && !redisPassword.isBlank()) {
+                builder.withPassword(redisPassword.toCharArray());
+            }
+            RedisClient client = RedisClient.create(builder.build());
+            try {
+                StatefulRedisConnection<String, String> connection = client.connect();
+                connection.sync().ping();
+                connection.close();
+            } finally {
+                client.shutdown();
+            }
+        } catch (Exception e) {
+            issues.append("连不上 Redis（").append(e.getMessage()).append("）；");
+        }
+
+        if (issues.length() == 0) {
+            available = Boolean.TRUE;
+            problem = null;
+        } else {
+            available = Boolean.FALSE;
+            problem = issues.toString();
+        }
+        return problem;
+    }
+
+    /** 读应用自己那份配置，命令行上的 -D 参数优先级更高，跟应用保持一致 */
+    private static PropertySourcesPropertyResolver resolver() {
+        ConfigurableEnvironment environment = new StandardEnvironment();
+        YamlPropertySourceLoader loader = new YamlPropertySourceLoader();
+        for (String name : CONFIG_FILES) {
+            Resource resource = new ClassPathResource(name);
+            if (!resource.exists()) {
+                continue;
+            }
+            try {
+                for (PropertySource<?> source : loader.load(name, resource)) {
+                    environment.getPropertySources().addLast(source);
+                }
+            } catch (IOException e) {
+                log.warn("读配置 {} 失败：{}", name, e.getMessage());
+            }
+        }
+        return new PropertySourcesPropertyResolver(environment.getPropertySources());
+    }
+
+    private static String firstNonBlank(String first, String second) {
+        return first != null && !first.isBlank() ? first : (second == null ? "" : second);
+    }
+}

main/xiaozhi-server/config.yaml

@@ -172,18 +172,19 @@ plugins:
     # ragflow知识库id
     dataset_ids: ["123456789"]
   # 联网搜索插件配置
-  # 支持秘塔(metaso)、Tavily和Serply三个搜索源
+  # 支持秘塔(metaso)、Tavily、Serply和Outcrawl四个搜索源
   # 秘塔API Key申请：https://metaso.cn/search-api/api-keys
   # Tavily API Key申请：https://app.tavily.com/home
   # Serply API Key申请：https://serply.io
+  # Outcrawl API Key申请：https://outcrawl.ai
   web_search:
-    # 搜索源：metaso / tavily / serply
+    # 搜索源：metaso / tavily / serply / outcrawl
     provider: "metaso"
     # 工具描述，用于告知LLM何时调用此工具，可根据需要自定义
     description: "联网搜索工具。当用户明确需要联网搜索问题时使用此工具。"
     # 返回结果数量（默认5）
     max_results: 5
-    # 秘塔API Key（mk-前缀） / Tavily API Key（tvly-前缀） / Serply API Key
+    # 秘塔API Key（mk-前缀） / Tavily API Key（tvly-前缀） / Serply API Key / Outcrawl API Key
     api_key: "mk-xxx"
 # 声纹识别配置
 voiceprint:
@@ -398,8 +399,8 @@ ASR:
     # DoubaoASR和DoubaoStreamASR的区别是：DoubaoASR是按次收费，DoubaoStreamASR是按时收费
     # 一般来说按次收费的更便宜，但是DoubaoStreamASR使用了大模型技术，效果更好
     type: doubao
-    appid: 你的火山引擎语音合成服务appid
+    appid: 你的火山引擎语音识别服务appid
-    access_token: 你的火山引擎语音合成服务access_token
+    access_token: 你的火山引擎语音识别服务access_token
     cluster: volcengine_input_common
     # 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738
     boosting_table_name: （选填）你的热词文件名称
@@ -412,8 +413,8 @@ ASR:
     # 开通地址https://console.volcengine.com/speech/service/10011
     # 一般来说按次收费的更便宜，但是DoubaoStreamASR使用了大模型技术，效果更好
     type: doubao_stream
-    appid: 你的火山引擎语音合成服务appid
+    appid: 你的火山引擎语音识别服务appid
-    access_token: 你的火山引擎语音合成服务access_token
+    access_token: 你的火山引擎语音识别服务access_token
     # 资源ID：小时版 volc.bigasr.sauc.duration，并发版 volc.bigasr.sauc.concurrent
     resource_id: volc.bigasr.sauc.duration
     # 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738
@@ -432,8 +433,8 @@ ASR:
     # 开通地址：https://console.volcengine.com/speech/service/10038
     # 价格更为便宜，建议在高并发场景下使用
     type: doubao_stream
-    appid: 你的火山引擎语音合成服务appid
+    appid: 你的火山引擎语音识别服务appid
-    access_token: 你的火山引擎语音合成服务access_token
+    access_token: 你的火山引擎语音识别服务access_token
     # 资源ID：小时版 volc.seedasr.sauc.duration，并发版 volc.seedasr.sauc.concurrent
     resource_id: volc.seedasr.sauc.duration
     # 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738
@@ -451,9 +452,9 @@ ASR:
     # token申请地址：https://console.cloud.tencent.com/cam/capi
     # 免费领取资源：https://console.cloud.tencent.com/asr/resourcebundle
     type: tencent
-    appid: 你的腾讯语音合成服务appid
+    appid: 你的腾讯语音识别服务appid
-    secret_id: 你的腾讯语音合成服务secret_id
+    secret_id: 你的腾讯语音识别服务secret_id
-    secret_key: 你的腾讯语音合成服务secret_key
+    secret_key: 你的腾讯语音识别服务secret_key
     output_dir: tmp/
   AliyunASR:
     # 阿里云智能语音交互服务，需要先在阿里云平台开通服务，然后获取验证信息
@@ -619,7 +620,7 @@ LLM:
     # 可在这里找到你的 api_key https://bailian.console.aliyun.com/?apiKey=1#/api-key
     base_url: https://dashscope.aliyuncs.com/compatible-mode/v1
     model_name: qwen-flash
-    api_key: 你的deepseek web key
+    api_key: 你的阿里云百炼API密钥
     temperature: 0.7  # 温度值
     max_tokens: 500   # 最大生成token数
     top_p: 1
@@ -763,10 +764,10 @@ VLLM:
     # 先新建应用，在下面的地址
     # 开通应用地址：https://console.xfyun.cn/app/myapp
     # 有免费额度，但也要开通服务，才能获取api_key
-    # 每一个模型都需要单独开通，每一个模型的api_password都不同，例如Lite模型在https://console.xfyun.cn/services/cbm 开通
+    # 每一个模型都需要单独开通，每一个模型的 api_key 都不同，例如 Lite 模型在 https://console.xfyun.cn/services/cbm 开通
     base_url: https://ark.cn-beijing.volces.com/api/v3
     model_name: lite
-    api_key: 你的api_password
+    api_key: 你的api_key
 TTS:
   # 当前支持的type为edge、doubao，可自行适配
   EdgeTTS:
@@ -1117,8 +1118,8 @@ TTS:
     #SpeechServerDemo https://github.com/PaddlePaddle/PaddleSpeech/tree/develop/demos/speech_server
     #流式传输请参考 https://github.com/PaddlePaddle/PaddleSpeech/wiki/PaddleSpeech-Server-WebSocket-API
     type: paddle_speech
-    protocol: websocket # protocol choices = ['websocket', 'http']
+    protocol: websocket # 当前仅实现 websocket；填其他值会在启动时报错
-    url: ws://127.0.0.1:8092/paddlespeech/tts/streaming  # TTS 服务的 URL 地址，指向本地服务器 [websocket默认ws://127.0.0.1:8092/paddlespeech/tts/streaming，http默认http://127.0.0.1:8090/paddlespeech/tts]
+    url: ws://127.0.0.1:8092/paddlespeech/tts/streaming  # TTS 服务地址（未配置时代码兜底同此本机地址）
     spk_id: 0  # 发音人 ID，0 通常表示默认的发音人
     speed: 1.0  # 语速，1.0 表示正常语速，>1 表示加快，<1 表示减慢
     volume: 1.0  # 音量，1.0 表示正常音量，>1 表示增大，<1 表示减小

main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py

@@ -1,6 +1,8 @@
 """服务端插件工具执行器"""
 
 import asyncio
+import functools
+from concurrent.futures import ThreadPoolExecutor
 from typing import Dict, Any, TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -8,6 +10,10 @@ if TYPE_CHECKING:
 from ..base import ToolType, ToolDefinition, ToolExecutor
 from plugins_func.register import all_function_registry, module_func_map, Action, ActionResponse
 
+# 插件专用线程池：不用 asyncio 默认池（DNS 解析、ASR、initialize_modules 等都在用），
+# 避免慢插件打满默认池后拖垮所有协程的 DNS 解析和连接建立。
+_PLUGIN_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="plugin")
+
 
 class ServerPluginExecutor(ToolExecutor):
     """服务端插件工具执行器"""
@@ -28,23 +34,21 @@ class ServerPluginExecutor(ToolExecutor):
 
         try:
             # 根据工具类型决定如何调用
-            if hasattr(func_item, "type"):
+            func_type = getattr(func_item, "type", None)
-                func_type = func_item.type
+            if func_type is not None and func_type.code in [3, 4, 5]:
-                if func_type.code in [4, 5]:  # SYSTEM_CTL, IOT_CTL (需要conn参数)
+                # CHANGE_SYS_PROMPT, SYSTEM_CTL, IOT_CTL (需要conn参数)
-                    result = func_item.func(conn, **arguments)
+                call = functools.partial(func_item.func, conn, **arguments)
-                elif func_type.code == 2:  # WAIT
+            else:  # WAIT 等不传conn（含 type 未指定的插件）
-                    result = func_item.func(**arguments)
+                call = functools.partial(func_item.func, **arguments)
-                elif func_type.code == 3:  # CHANGE_SYS_PROMPT
-                    result = func_item.func(conn, **arguments)
-                else:
-                    result = func_item.func(**arguments)
-            else:
-                # 默认不传conn参数
-                result = func_item.func(**arguments)
 
-            # 兼容 async def 工具函数
+            if asyncio.iscoroutinefunction(func_item.func):
-            if asyncio.iscoroutine(result):
+                result = await call()
-                result = await result
+            else:
+                # 同步函数在 loop 线程上直接执行会卡住整个事件循环，放专用线程池。
+                result = await asyncio.get_running_loop().run_in_executor(_PLUGIN_POOL, call)
+                # 兼容同步函数返回协程的写法
+                if asyncio.iscoroutine(result):
+                    result = await result
 
             return result
 

main/xiaozhi-server/core/providers/tools/unified_tool_manager.py

@@ -47,9 +47,12 @@ class ToolManager:
         return all_tools
 
     def get_function_descriptions(self) -> List[Dict[str, Any]]:
-        """获取所有工具的函数描述（OpenAI格式）"""
+        """获取所有工具的函数描述（OpenAI格式）
+
+        返回缓存的浅拷贝，防止调用方就地修改污染缓存。
+        """
         if self._cached_function_descriptions is not None:
-            return self._cached_function_descriptions
+            return list(self._cached_function_descriptions)
 
         descriptions = []
         tools = self.get_all_tools()
@@ -57,7 +60,7 @@ class ToolManager:
             descriptions.append(tool_definition.description)
 
         self._cached_function_descriptions = descriptions
-        return descriptions
+        return list(descriptions)
 
     def has_tool(self, tool_name: str) -> bool:
         """检查是否存在指定工具"""

main/xiaozhi-server/core/providers/tts/paddle_speech.py

@@ -23,9 +23,11 @@ class TTSProvider(TTSProviderBase):
 
     def __init__(self, config, delete_audio_file):
         super().__init__(config, delete_audio_file)
-        self.url = config.get("url", "ws://192.168.1.10:8092/paddlespeech/tts/streaming")
+        self.url = config.get("url", "ws://127.0.0.1:8092/paddlespeech/tts/streaming")
         self.protocol = config.get("protocol", "websocket")
-        
+        if self.protocol != "websocket":
+            raise ValueError("Unsupported protocol. Please use 'websocket'.")
+
         if config.get("private_voice"):
             self.spk_id = int(config.get("private_voice"))
         else:
@@ -82,7 +84,7 @@ class TTSProvider(TTSProviderBase):
         if self.protocol == "websocket":
             return await self.text_streaming(text, output_file)
         else:
-            raise ValueError("Unsupported protocol. Please use 'websocket' or 'http'.")
+            raise ValueError("Unsupported protocol. Please use 'websocket'.")
 
     async def text_streaming(self, text, output_file):
         try:

main/xiaozhi-server/core/utils/util.py

@@ -135,7 +135,7 @@ def get_ip_info(ip_addr, logger):
         if is_private_ip(ip_addr):
             ip_addr = ""
         url = f"https://whois.pconline.com.cn/ipJson.jsp?json=true&ip={ip_addr}"
-        resp = requests.get(url).json()
+        resp = requests.get(url, timeout=5).json()
         ip_info = {"city": resp.get("city")}
 
         # 存入缓存

main/xiaozhi-server/plugins_func/functions/get_weather.py

@@ -268,7 +268,8 @@ async def get_weather(conn: "ConnectionHandler", location: str = None, lang: str
                 location = cached_ip_info.get("city")
             else:
                 # 缓存未命中，调用API获取
-                ip_info = get_ip_info(client_ip, logger)
+                # get_ip_info 是同步 requests 调用，放线程池，避免卡住主事件循环
+                ip_info = await asyncio.to_thread(get_ip_info, client_ip, logger)
                 if ip_info:
                     cache_manager.set(CacheType.IP_INFO, client_ip, ip_info)
                     location = ip_info.get("city")

main/xiaozhi-server/plugins_func/functions/hass_state.py

@@ -121,13 +121,10 @@ async def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
                 + str(response.json()["attributes"]["media_title"])
                 + " "
             )
-        if "volume_level" in response.json()["attributes"]:
+        volume_level = response.json()["attributes"].get("volume_level")
-            responsetext = (
+        if isinstance(volume_level, (int, float)):
-                responsetext
+            # volume_level 量程 0～1，换算成百分比（同 volume_set 的 1～100）
-                + "音量是:"
+            responsetext = responsetext + "音量是:" + f"{round(volume_level * 100)}%" + " "
-                + str(response.json()["attributes"]["volume_level"])
-                + " "
-            )
         if "color_temp_kelvin" in response.json()["attributes"]:
             responsetext = (
                 responsetext
@@ -142,13 +139,11 @@ async def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
                 + str(response.json()["attributes"]["rgb_color"])
                 + " "
             )
-        if "brightness" in response.json()["attributes"]:
+        brightness = response.json()["attributes"].get("brightness")
-            responsetext = (
+        if isinstance(brightness, (int, float)):
-                responsetext
+            # brightness 量程 0～255，换算成百分比（同 brightness_pct 的 1～100），开灯最低报 1%
-                + "亮度是:"
+            percent = max(1, round(brightness * 100 / 255)) if brightness > 0 else 0
-                + str(response.json()["attributes"]["brightness"])
+            responsetext = responsetext + "亮度是:" + f"{percent}%" + " "
-                + " "
-            )
         logger.bind(tag=TAG).info(f"查询返回内容: {responsetext}")
         return responsetext
     else:

main/xiaozhi-server/plugins_func/functions/web_search.py

@@ -149,6 +149,54 @@ async def _search_serply(api_key: str, query: str, max_results: int) -> str:
     return "\n".join(lines)
 
 
+async def _search_outcrawl(api_key: str, query: str, max_results: int) -> str:
+    """调用Outcrawl搜索API"""
+    url = "https://api.outcrawl.ai/v1/search"
+    headers = {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+        "User-Agent": "xiaozhi-esp32-server",
+    }
+    # Outcrawl单次最多返回20条
+    payload = {"query": query, "limit": max(1, min(max_results, 20))}
+    logger.bind(tag=TAG).debug(f"Outcrawl搜索请求 | URL: {url} | payload: {payload}")
+    async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=3.0)) as client:
+        response = await client.post(url, json=payload, headers=headers)
+    data = response.json()
+    logger.bind(tag=TAG).debug(f"Outcrawl搜索响应 | status: {response.status_code}")
+
+    if response.status_code != 200:
+        # 错误信封为 {"code": "...", "message": "...", "details": {}}，
+        # code 是官方承诺的封闭枚举
+        code = data.get("code", "") if isinstance(data, dict) else ""
+        detail = data.get("message", "") if isinstance(data, dict) else ""
+        logger.bind(tag=TAG).error(
+            f"Outcrawl搜索失败 | status: {response.status_code} | code: {code} | detail: {detail}"
+        )
+        if code == "quota_exceeded":
+            # 该code覆盖两种情况：每分钟请求数限流，或月度credits用尽
+            return "联网搜索被限流或额度已用完，请稍后重试或检查Outcrawl账户余额。"
+        if code == "unauthorized":
+            return "联网搜索请求失败，请检查API Key是否正确。"
+        # 其余code（bad_request、no_capacity、search_refused、internal等）
+        # 多为输入问题或服务端暂时性故障，统一引导重试
+        return "联网搜索请求失败，请稍后重试。"
+
+    hits = data.get("hits", [])
+    if not hits:
+        return "未找到相关搜索结果。"
+
+    lines = ["【联网搜索结果】"]
+    for i, item in enumerate(hits[:max_results], 1):
+        title = item.get("title") or "无标题"
+        snippet = item.get("snippet", "")
+        lines.append(f"{i}. 标题：{title}")
+        if snippet:
+            lines.append(f"   摘要：{snippet}")
+
+    return "\n".join(lines)
+
+
 @register_function("web_search", WEB_SEARCH_FUNCTION_DESC, ToolType.SYSTEM_CTL)
 async def web_search(conn: "ConnectionHandler", query: str = None):
     logger.bind(tag=TAG).info(f"web_search 被调用 | query={query}")
@@ -175,6 +223,8 @@ async def web_search(conn: "ConnectionHandler", query: str = None):
             result_text = await _search_tavily(api_key, query, max_results)
         elif provider == "serply":
             result_text = await _search_serply(api_key, query, max_results)
+        elif provider == "outcrawl":
+            result_text = await _search_outcrawl(api_key, query, max_results)
         else:
             return ActionResponse(
                 Action.REQLLM,

main/xiaozhi-server/plugins_func/register.py

@@ -83,7 +83,11 @@ module_func_map = {}
 
 
 def register_function(name, desc, type=None):
-    """注册函数到函数注册字典的装饰器"""
+    """注册函数到函数注册字典的装饰器
+
+    同步函数在工作线程中执行，函数体内不可调用 asyncio 事件循环 API；
+    需要异步（httpx、await）请定义为 async def。
+    """
 
     def decorator(func):
         all_function_registry[name] = FunctionItem(name, desc, func, type)

main/xiaozhi-server/tests/conftest.py

@@ -47,10 +47,29 @@ if str(_PROJECT_ROOT) not in sys.path:
 # ---------------------------------------------------------------------------
 try:
     import opuslib_next  # noqa: F401
-except Exception:
+except Exception as _opus_import_error:
+    # opuslib_next 是 ctypes 绑定，pip 装上的是纯 Python 包，真正的编解码靠系统里的
+    # libopus.so。runner 上没装 libopus 时 import 会失败，这里换桩顶上。
+    # 桩只能撑住 import 链路，真要编解码得先把 libopus 装上
+    # （apt-get install libopus0，正式镜像 Dockerfile-server-base 里也是这么装的）。
+    print(
+        f"[conftest] opuslib_next 不可用，改用桩：{_opus_import_error}",
+        flush=True,
+    )
+
+    def _opus_unavailable(_self, *_args, **_kwargs):
+        raise RuntimeError(
+            "libopus 没装，Opus 编解码跑不了。"
+            "装法：apt-get install libopus0（正式镜像 Dockerfile-server-base 里也是这么装的）"
+        )
+
     _stub = types.ModuleType("opuslib_next")
-    _stub.Encoder = type("Encoder", (), {"__init__": lambda *a, **kw: None})
+    _stub.Encoder = type(
-    _stub.Decoder = type("Decoder", (), {"__init__": lambda *a, **kw: None})
+        "Encoder", (), {"__init__": lambda *a, **kw: None, "encode": _opus_unavailable}
+    )
+    _stub.Decoder = type(
+        "Decoder", (), {"__init__": lambda *a, **kw: None, "decode": _opus_unavailable}
+    )
     _stub.APPLICATION_AUDIO = 0
     _stub.OpusError = type("OpusError", (Exception,), {})
     _stub.constants = types.SimpleNamespace()
@@ -323,35 +342,25 @@ def _install_mock_layer() -> None:
 
     asyncio.sleep = _fast_sleep  # type: ignore[assignment]
 
-    # 网络层 mock：如果真实 httpx 已加载，把它的 AsyncClient.send 替换成
+    # 网络层 mock：如果真实 httpx 已加载，把它的 send 替换成
     # 立即抛 ConnectError 的桩，触发 ``_should_retry`` 走到 ``raise``，
     # 但避免真发请求。Live 模式完全不安装这层。
     if not LIVE_API_TESTS:
         try:
-            import httpx as _httpx  # noqa: F401
+            import httpx as _httpx
 
-            class _FakeAsyncClient:
+            async def _fake_async_send(self, request, **kwargs):
-                def __init__(self, *args, **kwargs):
+                raise _httpx.ConnectError("MOCK: connect refused")
-                    pass
 
-                async def send(self, *args, **kwargs):
+            def _fake_send(self, request, **kwargs):
-                    raise _httpx.ConnectError("MOCK: connect refused")
+                raise _httpx.ConnectError("MOCK: connect refused")
 
-                async def aclose(self):
+            # 只换 send 这一个方法，不换整个类。有些库（openai 2.x 的
-                    return None
+            # SyncHttpxClientWrapper）继承 httpx.Client 再自己拼请求，
-
+            # 整个类被换掉的话它们继承到的是假类，拼请求时直接
-            class _FakeClient:
+            # AttributeError，报错还完全看不出是 mock 层的问题
-                def __init__(self, *args, **kwargs):
+            _httpx.AsyncClient.send = _fake_async_send  # type: ignore[assignment]
-                    pass
+            _httpx.Client.send = _fake_send  # type: ignore[assignment]
-
-                def send(self, *args, **kwargs):
-                    raise _httpx.ConnectError("MOCK: connect refused")
-
-                def close(self):
-                    return None
-
-            _httpx.AsyncClient = _FakeAsyncClient  # type: ignore[misc]
-            _httpx.Client = _FakeClient  # type: ignore[misc]
         except ImportError:
             pass
 

main/xiaozhi-server/tests/core/providers/tools/test_plugin_executor_offload.py

@@ -0,0 +1,134 @@
+"""服务端插件执行器：同步插件函数在线程池里跑，不卡事件循环；async 插件照常 await；异常变成 ERROR 响应。"""
+from __future__ import annotations
+
+import asyncio
+import time
+from types import SimpleNamespace
+
+from core.providers.tools.server_plugins.plugin_executor import ServerPluginExecutor
+from plugins_func.register import Action, ActionResponse, ToolType, all_function_registry
+
+
+def _register(name, func, tool_type):
+    item = SimpleNamespace(name=name, func=func, type=tool_type, description={})
+    all_function_registry[name] = item
+    return item
+
+
+def test_sync_plugin_runs_off_loop_and_async_plugin_is_awaited():
+    conn = SimpleNamespace(config={})
+
+    async def paced_sleep(delay):
+        # 不能用 asyncio.sleep 计时：conftest 在 mock 模式会把它换成不让出循环的桩，
+        # 这里放到线程池里睡，保证 ticker 每拍都真正让出事件循环。
+        await asyncio.get_running_loop().run_in_executor(None, time.sleep, delay)
+
+    def slow_sync(conn_arg, entity_id):
+        time.sleep(0.3)  # 模拟阻塞 IO
+        return ActionResponse(Action.RESPONSE, None, f"{entity_id} 已打开")
+
+    async def fast_async(question):
+        await asyncio.sleep(0)
+        return ActionResponse(Action.REQLLM, f"答：{question}", None)
+
+    def boom(conn_arg):
+        raise RuntimeError("连不上")
+
+    _register("t_slow_sync", slow_sync, ToolType.SYSTEM_CTL)
+    _register("t_fast_async", fast_async, ToolType.WAIT)
+    _register("t_boom", boom, ToolType.SYSTEM_CTL)
+    try:
+        async def go():
+            ex = ServerPluginExecutor(conn)
+            ticks = []
+
+            async def ticker():
+                while True:
+                    ticks.append(time.monotonic())
+                    await paced_sleep(0.02)
+
+            t = asyncio.create_task(ticker())
+            r = await ex.execute(conn, "t_slow_sync", {"entity_id": "light.living"})
+            t.cancel()
+            assert r.action == Action.RESPONSE and r.response == "light.living 已打开"
+            assert len(ticks) >= 5                      # 300 ms 里事件循环一直在转（没被 time.sleep 卡住）
+            r2 = await ex.execute(conn, "t_fast_async", {"question": "几点"})
+            assert r2.action == Action.REQLLM and r2.result == "答：几点"
+            r3 = await ex.execute(conn, "t_boom", {})
+            assert r3.action == Action.ERROR and "连不上" in r3.response
+            r4 = await ex.execute(conn, "t_missing", {})
+            assert r4.action == Action.NOTFOUND
+        asyncio.run(go())
+    finally:
+        for n in ("t_slow_sync", "t_fast_async", "t_boom"):
+            all_function_registry.pop(n, None)
+
+
+def test_conn_passing_rules_and_compat_branches():
+    """code 3/4/5 传 conn；WAIT/type=None 不传；同步函数返回协程会被兜底 await。"""
+    conn = SimpleNamespace(config={})
+    seen = {}
+
+    def prompt_fn(conn_arg):
+        seen["prompt"] = conn_arg
+        return ActionResponse(Action.RESPONSE, None, "切角色")
+
+    def wait_fn():
+        seen["wait"] = True
+        return ActionResponse(Action.RESPONSE, None, "等")
+
+    def no_type_fn():
+        seen["no_type"] = True
+        return ActionResponse(Action.RESPONSE, None, "无类型")
+
+    def sync_returns_coro(conn_arg):
+        seen["coro"] = conn_arg
+
+        async def inner():
+            return ActionResponse(Action.RESPONSE, None, "协程兜底")
+
+        return inner()
+
+    _register("t_prompt", prompt_fn, ToolType.CHANGE_SYS_PROMPT)
+    _register("t_wait", wait_fn, ToolType.WAIT)
+    _register("t_no_type", no_type_fn, None)
+    _register("t_sync_coro", sync_returns_coro, ToolType.SYSTEM_CTL)
+    try:
+        async def go():
+            ex = ServerPluginExecutor(conn)
+            r1 = await ex.execute(conn, "t_prompt", {})
+            assert seen["prompt"] is conn and r1.response == "切角色"
+            r2 = await ex.execute(conn, "t_wait", {})
+            assert "wait" in seen and r2.response == "等"
+            r3 = await ex.execute(conn, "t_no_type", {})
+            assert "no_type" in seen and r3.response == "无类型"
+            r4 = await ex.execute(conn, "t_sync_coro", {})
+            assert seen["coro"] is conn and r4.response == "协程兜底"
+        asyncio.run(go())
+    finally:
+        for n in ("t_prompt", "t_wait", "t_no_type", "t_sync_coro"):
+            all_function_registry.pop(n, None)
+
+
+def test_sync_plugin_uses_dedicated_pool():
+    """同步插件必须跑在专用池上，不能用 asyncio 默认池（DNS 解析等共用默认池）。"""
+    import threading
+
+    conn = SimpleNamespace(config={})
+    seen = {}
+
+    def sync_fn():
+        seen["thread"] = threading.current_thread().name
+        return ActionResponse(Action.RESPONSE, None, "ok")
+
+    _register("t_pool", sync_fn, ToolType.WAIT)
+    try:
+        async def go():
+            ex = ServerPluginExecutor(conn)
+            r = await ex.execute(conn, "t_pool", {})
+            assert r.response == "ok"
+
+        asyncio.run(go())
+        assert seen["thread"].startswith("plugin"), seen["thread"]  # 专用池线程名前缀
+    finally:
+        all_function_registry.pop("t_pool", None)

main/xiaozhi-server/tests/plugins_func/test_web_search_outcrawl.py

@@ -0,0 +1,122 @@
+"""Tests for the Outcrawl source of the web_search plugin.
+
+The HTTP layer is replaced with a fake client and every request is recorded.
+Nothing here touches the network.
+"""
+import httpx
+import pytest
+
+from plugins_func.functions import web_search as web_search_module
+from plugins_func.register import Action
+
+
+class FakeResponse:
+    def __init__(self, status_code=200, payload=None):
+        self.status_code = status_code
+        self.payload = payload
+
+    def json(self):
+        return self.payload
+
+
+class FakeAsyncClient:
+    """Stands in for httpx.AsyncClient and records what was requested."""
+
+    requests = []
+    response = None
+
+    def __init__(self, *args, **kwargs):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc_info):
+        return False
+
+    async def post(self, url, json=None, headers=None):
+        FakeAsyncClient.requests.append({"url": url, "json": json, "headers": headers})
+        return FakeAsyncClient.response
+
+
+@pytest.fixture
+def fake_http(monkeypatch):
+    """Replace httpx.AsyncClient with the recording fake."""
+    FakeAsyncClient.requests = []
+    FakeAsyncClient.response = FakeResponse(payload={"hits": []})
+    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
+    return FakeAsyncClient
+
+
+class FakeConn:
+    def __init__(self, web_search_config):
+        self.config = {"plugins": {"web_search": web_search_config}}
+
+
+async def test_outcrawl_sends_bearer_key_and_limit(fake_http):
+    await web_search_module._search_outcrawl("oc-key", "上海天气", 5)
+
+    [request] = fake_http.requests
+    assert request["url"] == "https://api.outcrawl.ai/v1/search"
+    assert request["json"] == {"query": "上海天气", "limit": 5}
+    assert request["headers"]["Authorization"] == "Bearer oc-key"
+    assert request["headers"]["User-Agent"] == "xiaozhi-esp32-server"
+
+
+@pytest.mark.parametrize(("max_results", "limit"), [(0, 1), (50, 20)])
+async def test_outcrawl_clamps_limit(fake_http, max_results, limit):
+    await web_search_module._search_outcrawl("oc-key", "q", max_results)
+
+    assert fake_http.requests[0]["json"]["limit"] == limit
+
+
+async def test_outcrawl_formats_titles_and_snippets_without_urls(fake_http):
+    fake_http.response = FakeResponse(
+        payload={
+            "hits": [
+                {"url": "https://a.example/1", "title": "标题一", "snippet": "摘要一", "rank": 1},
+                {"url": "https://a.example/2", "title": None, "snippet": "", "rank": 2},
+                {"url": "https://a.example/3", "title": "标题三", "snippet": "摘要三", "rank": 3},
+            ],
+            "usage": {"credits": "1.5"},
+        }
+    )
+
+    result = await web_search_module._search_outcrawl("oc-key", "q", 2)
+
+    assert result == "【联网搜索结果】\n1. 标题：标题一\n   摘要：摘要一\n2. 标题：无标题"
+    assert "https://" not in result
+
+
+async def test_outcrawl_empty_hits(fake_http):
+    fake_http.response = FakeResponse(payload={"hits": []})
+
+    assert await web_search_module._search_outcrawl("oc-key", "q", 3) == "未找到相关搜索结果。"
+
+
+@pytest.mark.parametrize(
+    ("code", "message"),
+    [
+        ("unauthorized", "联网搜索请求失败，请检查API Key是否正确。"),
+        ("quota_exceeded", "联网搜索被限流或额度已用完，请稍后重试或检查Outcrawl账户余额。"),
+        ("bad_request", "联网搜索请求失败，请稍后重试。"),
+        ("internal", "联网搜索请求失败，请稍后重试。"),
+    ],
+)
+async def test_outcrawl_error_code(fake_http, code, message):
+    fake_http.response = FakeResponse(
+        status_code=429, payload={"code": code, "message": "denied", "details": {}}
+    )
+
+    assert await web_search_module._search_outcrawl("oc-key", "q", 3) == message
+
+
+async def test_web_search_dispatches_to_outcrawl(fake_http):
+    fake_http.response = FakeResponse(payload={"hits": [{"url": "https://a.example", "title": "T", "snippet": "S"}]})
+    conn = FakeConn({"provider": "Outcrawl", "api_key": "oc-key", "max_results": 3})
+
+    response = await web_search_module.web_search(conn, "q")
+
+    assert response.action == Action.REQLLM
+    assert response.result == "【联网搜索结果】\n1. 标题：T\n   摘要：S"
+    assert fake_http.requests[0]["json"] == {"query": "q", "limit": 3}

main/xiaozhi-server/tests/test_llm.py

@@ -65,6 +65,13 @@ def _build_provider():
     ptype = get_provider_type("LLM", name)
     if not ptype:
         pytest.skip(f"LLM '{name}' 的 type 字段缺失")
+    if not LIVE_API_TESTS:
+        # mock 模式把网络拦了，这条链路拿不到真实回复，跑下来只是一次连接失败。
+        # 真要验这条链路，用 RUN_LIVE_API_TESTS=1 配真 key 跑
+        pytest.skip(
+            f"mock 模式不发真请求，跳过 LLM '{name}' 的链路验证。"
+            "真请求模式：export RUN_LIVE_API_TESTS=1 并填真 key 后再跑。"
+        )
     from core.utils.llm import create_instance
 
     return create_instance(ptype, cfg)