LobeHub Agent Runtime E2E 测试指南：最小 Mock 策略、内存态管理器与 OpenAI 流式响应仿真

本文基于 LobeHub 仓库内置的测试技能文档 Agent Runtime E2E Testing Guide，系统讲解如何对 Agent 运行时（AgentRuntimeService / AgentRuntimeCoordinator）编写端到端测试：只 Mock 数据库与 Redis 两类外部依赖，用 vi.spyOn 拦截 fetch 伪造 OpenAI SSE 流式响应，让真实的模型配置、工具引擎与运行时调度链在测试中完整跑通。读完本文，你可以照仓库既有模式为 Agent 运行时的新场景（文本回复、工具调用、中断等）补写可稳定复现的 E2E 测试。

一、核心原则：最小 Mock（Minimal Mock Principle）

该指南最重要的判断标准是：只 Mock 三个外部依赖，其余全部走真实实现。

依赖	Mock 方案	说明
数据库	PGLite	内存数据库，来自 @lobechat/database/test-utils 的 getTestDB()
Redis（Agent 状态）	InMemoryAgentStateManager	内存实现
Redis（流事件）	InMemoryStreamEventManager	内存实现

明确不 Mock 的部分：

• model-bank —— 使用真实模型配置（保证模型元数据链路真实可用）；
• Mecha（AgentToolsEngine、ContextEngineering）—— 工具引擎与上下文工程走真实实现；
• AgentRuntimeService 与 AgentRuntimeCoordinator —— 被测对象本身。

这个"边界 Mock、内部真实"的策略意味着测试验证的是整条 Agent 执行链的真实行为：从运行时拿到真实模型配置、调用工具引擎、逐步推进状态机，直到产生流式事件与最终状态。由于只有网络层（LLM 的 fetch 调用）被 spy 拦截，任何内部 wiring 的回归（例如事件漏发、状态机步骤错乱）都会在断言中暴露，而不是被 Mock 吞掉。

这一点与 LobeHub Testing Guide 中的总体原则一致：优先 vi.spyOn 而非 vi.mock，在边界（DB、网络、外部服务）打 Mock，而不是在内部模块之间打。

为什么用 vi.spyOn 而不是 vi.mock

指南给出的理由很直接：不同的测试需要不同的 LLM 响应。vi.mock 是模块级、作用域过宽的静态替换；而 vi.spyOn(globalThis, 'fetch') 配合 mockResolvedValueOnce 可以：

• 按测试用例灵活指定返回值（每个 it 独立控制）；
• 方便测试不同场景（纯文本回复、工具调用、错误等）；
• 提供更好的测试隔离（配合 beforeEach 中 vi.clearAllMocks()）。

默认模型：gpt-5

指南指定 E2E 测试默认使用 gpt-5：它始终存在于 model-bank 中，且在模型目录更新时保持稳定。这样测试断言可以硬编码模型名（如流式 chunk 中的 model: 'gpt-5'），不会因为默认模型轮换导致测试漂移。

二、技术实现：四步搭起一套 Agent Runtime E2E 测试

以下四个片段完整继承自指南原文，是搭建一套可运行 E2E 测试的标准骨架。

2.1 数据库准备：getTestDB()（PGLite 内存库）

import { LobeChatDatabase } from '@lobechat/database';
import { getTestDB } from '@lobechat/database/test-utils';

let testDB: LobeChatDatabase;

beforeEach(async () => {
  testDB = await getTestDB();
});

getTestDB() 的实现位于数据库包内（见 getTestDB.ts），基于 PGLite 提供进程内 PostgreSQL。每个测试用例 beforeEach 重新取一个干净库，天然保证用例间的数据隔离——这也是指南 Notes 部分"测试隔离"要求的具体落点之一。

2.2 OpenAI 流式响应助手：createOpenAIStreamResponse

Agent 运行时的 LLM 调用走 OpenAI 兼容的 SSE 协议（data: {...}\n\n 分片，data: [DONE] 收尾）。指南给出的助手函数把"内容 / 工具调用 / 结束原因"三个维度收敛成一个工厂，产出标准 Response（content-type: text/event-stream）：

export const createOpenAIStreamResponse = (options: {
  content?: string;
  toolCalls?: Array<{ id: string; name: string; arguments: string }>;
  finishReason?: 'stop' | 'tool_calls';
}) => {
  const { content, toolCalls, finishReason = 'stop' } = options;

  return new Response(
    new ReadableStream({
      start(controller) {
        const encoder = new TextEncoder();

        if (content) {
          const chunk = {
            id: 'chatcmpl-mock',
            object: 'chat.completion.chunk',
            model: 'gpt-5',
            choices: [{ index: 0, delta: { content }, finish_reason: null }],
          };
          controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`));
        }

        // ... tool_calls handling
        // ... finish chunk
        controller.enqueue(encoder.encode('data: [DONE]\n\n'));
        controller.close();
      },
    }),
    { headers: { 'content-type': 'text/event-stream' } },
  );
};

设计要点：

• 返回的是真正的 Web Response + ReadableStream，与运行时内部 SSE 解析器消费的数据形状完全一致，不需要额外适配；
• finishReason 默认 'stop'，需要驱动"模型要求继续执行工具"的场景时显式传 'tool_calls'；
• 工具调用参数以 JSON.stringify 后的字符串形式放入 arguments 字段，符合 OpenAI 流式协议中 tool_calls 的编码方式。

这个助手正是"Mock 边界在协议层而非模块层"的体现：运行时拿到的是一个格式合法的 SSE 响应，后续解析、状态机推进、事件发布全部是真实代码。

2.3 状态管理：注入内存态管理器

import {
  InMemoryAgentStateManager,
  InMemoryStreamEventManager,
} from '@/server/modules/AgentRuntime';

const stateManager = new InMemoryAgentStateManager();
const streamEventManager = new InMemoryStreamEventManager();

const service = new AgentRuntimeService(serverDB, userId, {
  coordinatorOptions: { stateManager, streamEventManager },
  queueService: null,
  streamEventManager,
});

AgentRuntimeService 的构造参数允许注入 coordinatorOptions.stateManager 与 streamEventManager，把生产环境依赖的 Redis 状态管理/流事件管理替换为内存实现。这两个类的真实位置在 InMemoryAgentStateManager.ts 与 InMemoryStreamEventManager.ts，文件头注释均标明其定位："In-memory implementation for testing and local development environments"。

2.4 Mock OpenAI API：spy 全局 fetch

const fetchSpy = vi.spyOn(globalThis, 'fetch');

it('should handle text response', async () => {
  fetchSpy.mockResolvedValueOnce(createOpenAIStreamResponse({ content: 'Response text' }));
  // ... execute test
});

it('should handle tool calls', async () => {
  fetchSpy.mockResolvedValueOnce(
    createOpenAIStreamResponse({
      toolCalls: [
        {
          id: 'call_123',
          name: 'lobe-web-browsing____search',
          arguments: JSON.stringify({ query: 'weather' }),
        },
      ],
      finishReason: 'tool_calls',
    }),
  );
  // ... execute test
});

mockResolvedValueOnce 是"不同测试不同 LLM 响应"的关键：每个用例按自己的剧本喂一次流式响应，互不干扰。工具名示例 lobe-web-browsing____search 展示了工具调用在 LLM 侧的命名约定（插件前缀 + 下划线 + 工具名），测试可直接断言工具引擎按该名字路由。

三、源码级解析：两个内存态管理器到底做了什么

指南的 Notes 第 1 条要求"每个测试后清理 InMemoryAgentStateManager 与 InMemoryStreamEventManager"。这两个类为此专门暴露了测试向 API，读懂它们的内部结构，才能写出可断言、可清理的测试。

3.1 InMemoryAgentStateManager：状态、步骤历史与执行锁

从 InMemoryAgentStateManager.ts 的源码看，该类实现了 IAgentStateManager 接口，内部维护五个 Map：

• states：operationId -> AgentState，保存/读取时都用 structuredClone 做深拷贝，防止运行时持有引用后继续修改内部副本；saveStepResult 特意改用 JSON 往返序列化，注释说明是为了安全处理不可 structuredClone 的对象（例如 Neon DB 抛出的 DOM ErrorEvent）；
• steps / events：每步的执行历史与事件序列，均只保留最近 200 条（unshift 插入、超长截断）；
• metadata：操作元数据（状态、累计成本、步数、lastActiveAt 等），由 createOperationMetadata 初始化；
• stepLocks：步骤执行锁。tryClaimStep 默认 TTL 为 35 秒，模拟生产环境的分布式抢占语义，refreshStepLock / releaseStepLock 按 ownerId 校验归属。

测试向的 API 有三个，直接对应指南的隔离与断言需求：

方法	用途
clear()	清空全部五个 Map，放在 afterEach 中即可满足"每个测试后清理"的要求
getEventHistory(operationId)	返回每步的事件序列，用于逐步断言事件内容
getStats()	按 running / waiting_for_human / done / error / interrupted 统计操作状态，可用于终态断言

其单元测试见 InMemoryAgentStateManager.test.ts，其中 beforeEach 直接 new InMemoryAgentStateManager()，与指南建议的"每测试一个新实例"等价。

3.2 InMemoryStreamEventManager：事件流、断言助手与形状对齐

InMemoryStreamEventManager.ts 实现了 IStreamEventManager，核心结构与行为：

• 事件缓冲：streams 按 operationId 存储事件数组，上限 1000 条（超出 shift() 丢弃最旧），防止内存溢出；
• 事件形状与生产一致：publishStreamEvent 内部调用 stripFinalStateInEventData 对 data 做与 Redis 版管理器相同的"收口剥离"，源码注释明确指出这是为了让内存版事件形状与生产线上 wire 格式完全一致——否则测试会掩盖剥离逻辑的回归。这是一个值得借鉴的细节：Mock 实现不只是"能跑"，还要"形状对"；
• 长轮询原语：readEventsOnce(operationId, lastEventId = '$', blockMs) 是非阻塞实现——'_' 之外的游标按 id 定位后返回其后事件，'_' 返回当前尾游标；blockMs 被忽略，注释说明真正的阻塞等待由 Redis 版提供；
• 测试断言三件套：
  • getAllEvents(operationId)：取全部事件做顺序/内容断言；
  • waitForEvent(operationId, eventType, timeout = 5000)：订阅 + 回放已有事件，等待指定类型事件（如 agent_runtime_end），默认 5 秒超时；
  • clear()：清空 streams、subscribers 与事件 id 计数器。

事件 id 采用 ${Date.now()}-${counter} 生成，getStreamHistory 返回倒序最近 N 条（默认 100），这些细节与 InMemoryStreamEventManager.test.ts 中的覆盖用例相互印证。

3.3 工厂装配：in-memory 分支如何进入生产代码路径

除了测试直接 new 之外，运行时模块的工厂也感知内存实现：factory.test.ts 中 mock 了 inMemoryAgentStateManager 单例并断言 createAgentStateManager() 在 in-memory 配置下返回它。从源码结构看，两个 InMemory 类文件末尾都导出了对应单例（inMemoryAgentStateManager / inMemoryStreamEventManager），供本地开发环境与工厂的 in-memory 分支复用——也就是说，E2E 测试用的组件与本地开发用的是同一份代码，测试通过即代表该路径在真实装配下可用。

四、运行与调试：超时、日志与既有测试参考

运行命令

仓库测试技能指南 SKILL.md 给出的标准命令是：

# 运行指定测试文件
bunx vitest run --silent='passed-only' '[file-path]'

指南 Notes 第 2 条提醒：E2E 测试可能需要更长的超时（完整的运行时循环涉及多次 LLM"调用"与工具执行，waitForEvent 默认 5 秒，多步用例应整体放宽 it 超时），编写用例时按步骤数评估并显式设置。

调试日志

Notes 第 3 条建议用 DEBUG=lobe-server:* 打开详细日志。两个内存管理器恰好注册了独立的 debug 命名空间（见源码中的 debug() 调用）：

• lobe-server:agent-runtime:in-memory-state-manager：每次保存状态、步骤结果、创建元数据、清理操作都会打日志（含 operationId 与步数）；
• lobe-server:agent-runtime:in-memory-stream-event-manager：发布事件、清理操作、断开连接均有日志。

排查"事件没出现"类问题时，先确认是运行时没发布，还是断言查错了 operationId。

既有的相关测试

以下测试文件可作为新用例的写法参照，位于 tests 目录：

• InMemoryAgentStateManager.test.ts：状态保存/读取、步骤历史、执行锁语义；
• InMemoryStreamEventManager.test.ts：事件发布、游标读取、清理；
• AgentRuntimeCoordinator.test.ts：协调器层的 E2E 级覆盖；
• factory.test.ts：in-memory 装配分支。

五、要点回顾

1. 只 Mock 三个外部依赖（PGLite 数据库、两个内存态管理器），model-bank、Mecha（工具引擎/上下文工程）、AgentRuntimeService、AgentRuntimeCoordinator 全部真实运行——这是该指南与一般"到处打 mock"式测试的分水岭；
2. vi.spyOn(globalThis, 'fetch') + mockResolvedValueOnce 按用例喂入不同 OpenAI SSE 响应，配合 createOpenAIStreamResponse 工厂覆盖文本、工具调用、结束等剧本；
3. 内存实现自带测试 API：clear() 保证隔离，getAllEvents / getEventHistory / waitForEvent / getStats 提供断言抓手，且事件形状与生产 Redis 版刻意对齐；
4. 默认模型固定 gpt-5、按需放大超时、DEBUG=lobe-server:* 排障，构成稳定的日常调试闭环。

按上述骨架补齐自己的用例后，运行 bunx vitest run --silent='passed-only' <file> 验证，即可将新的 Agent 运行时行为纳入这套最小 Mock 的 E2E 保护网。