测试

SDK 为智能体工作流、沙盒会话和 Realtime 会话提供确定性、与提供方无关的测试替身。它们在内存中运行，不会向模型、沙盒提供方或 Realtime API 发送请求，并会记录 SDK 负责的规范化交互。

所需方案索引

我想要…	使用	跳转至
返回固定的最终答案	`ScriptedModel` 配合 `assistantMessage()`	返回固定响应
测试多轮工具循环	`functionCall()` 后接助手响应	测试工具工作流
根据请求选择响应	`modelResponder()`	根据请求生成响应
断言运行器发送给模型的内容	`calls`、`firstCall` 或 `lastCall`	检查模型调用
测试流式运行	常规响应步骤，或使用 `modelStream()` 指定精确事件	测试流式传输
测试错误或重试决策	`modelError()`	注入模型故障
检测意外的工作流变化	精确步骤配合 `assertComplete()`	检测工作流偏移
在不启动沙盒的情况下测试 `SandboxAgent`	`scriptedSandboxSession()` 配合 `ScriptedModel`	测试沙盒智能体工作流
匹配沙盒调用或根据调用生成结果	沙盒步骤上的 `match` 或 `respond`	配置沙盒步骤
注入沙盒操作故障	沙盒步骤上的 `error`	配置沙盒步骤
测试 `RealtimeAgent` 的函数工具	发出 `function_call` 并预期 `sendFunctionCallOutput`	测试语音智能体工具工作流
按顺序匹配 Realtime 操作	`runScenario()` 和 `expectCall()`	测试 Realtime 会话
发出入站 Realtime 事件	场景中的 `emit()`	测试 Realtime 会话
测试中断请求和通知	`expectCall('interrupt')` 配合 `emit('audio_interrupted')`	测试中断
使下一次 Realtime 操作失败	`failNextCall()`	注入 Realtime 故障
测试远端断开连接	`disconnect(error?)`	测试远端断开连接
测试提供方请求序列化或线上传输载荷	真实提供方适配器配合模拟网络传输	选择正确的边界
测试沙盒提供方生命周期或隔离	集成测试中的真实沙盒提供方	选择正确的边界

导入

大多数应用可以使用便捷子路径：

边界	推荐导入路径	标准包导入路径
智能体、`Model` 和沙盒工作流	`@openai/agents/testing`	`@openai/agents-core/testing`
Realtime 传输	`@openai/agents/realtime/testing`	`@openai/agents-realtime/testing`

便捷子路径公开的测试 API 与标准包相同。测试符号不会从主要运行时入口点导出。

智能体工作流方案

返回固定响应

为每次预期的模型调用传入一个规范化输出项数组。输出数组简写形式会为一次请求提供确定性的响应 ID 和用量。

import { Agent, Runner } from '@openai/agents';
import { ScriptedModel, assistantMessage } from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('returns a deterministic final answer', async () => {
  const model = new ScriptedModel([
    [assistantMessage('Paris is the capital of France.')],
  ]);
  const agent = new Agent({
    name: 'Geography assistant',
    model,
  });

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'What is the capital of France?');

  assert.equal(result.finalOutput, 'Paris is the capital of France.');
  assert.equal(model.calls.length, 1);
  // Also fail if a later agent change stops before using the whole script.
  model.assertComplete();
});

测试结束时务必调用 model.assertComplete()。它可以捕获工作流在使用完所有已配置步骤之前就停止的情况。

测试工具工作流

通过脚本配置两个模型响应：第一个调用工具，第二个生成最终答案。真实的 SDK 工具流水线会在这两次模型调用之间执行。

import { Agent, Runner, tool } from '@openai/agents';
import {
  ScriptedModel,
  assistantMessage,
  functionCall,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';
import { z } from 'zod';

test('runs a multi-turn tool workflow', async () => {
  const weather = tool({
    name: 'get_weather',
    description: 'Gets the weather for a city.',
    parameters: z.object({ city: z.string() }),
    execute: async ({ city }) => `${city}: sunny`,
  });
  const model = new ScriptedModel([
    // The first model turn enters the real SDK tool execution pipeline.
    [functionCall('get_weather', { city: 'Tokyo' }, { callId: 'call_1' })],
    // The second turn sees the tool result and finishes the workflow.
    [assistantMessage('It is sunny in Tokyo.')],
  ]);
  const agent = new Agent({
    name: 'Weather assistant',
    model,
    tools: [weather],
  });

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'What is the weather in Tokyo?');

  assert.equal(result.finalOutput, 'It is sunny in Tokyo.');
  assert.equal(model.calls.length, 2);
  const lastInput = model.lastCall?.request.input;
  assert(Array.isArray(lastInput));
  assert(lastInput.some((item) => item.type === 'function_call_result'));
  model.assertComplete();
});

这种模式会测试工具输入验证、执行、结果转换、钩子、护栏以及下一轮模型调用。它不会通过直接调用工具函数来绕过工具运行时。

根据请求生成响应

当响应取决于规范化请求，或断言应位于模型边界时，使用 modelResponder()。回调可以是同步或异步的。

import { Agent, Runner } from '@openai/agents';
import {
  ScriptedModel,
  assistantMessage,
  modelResponder,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('derives a response from the recorded request', async () => {
  const model = new ScriptedModel([
    modelResponder((call) => {
      // The responder receives the normalized request at the Model boundary.
      assert.equal(call.index, 0);
      assert.equal(call.streamed, false);
      assert.deepEqual(call.request.input, [
        {
          type: 'message',
          role: 'user',
          content: 'Summarize this',
        },
      ]);
      return [assistantMessage(`Handled model call ${call.index}.`)];
    }),
  ]);
  const agent = new Agent({ name: 'Assistant', model });

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'Summarize this');

  assert.equal(result.finalOutput, 'Handled model call 0.');
  model.assertComplete();
});

对于确实依赖请求的行为，使用响应器。如果简单的序列就足够，优先使用固定步骤；固定脚本更便于诊断意外轮次。

检查模型调用

ScriptedModel 会在完成所选步骤或抛出错误之前记录调用。

成员	内容
`calls`	按调用顺序排列的所有 `RecordedModelCall`
`firstCall`	第一次调用，或 `undefined`
`lastCall`	最近一次调用，或 `undefined`
`remainingSteps`	尚未使用的已配置步骤数

每次调用都有一个从零开始的 index，与它在 calls 中的位置一致，还包含一个 streamed 标志和规范化的 request。记录请求中的可变数据容器会在调用时复制为独立快照。读取或修改某个返回的快照不会改写模型保留的历史记录。回调、类实例和 AbortSignal 等运行时对象则保持对象身份不变。

常见断言包括：

使用 call.request.input 检查用户输入、工具调用和工具结果
使用 call.request.modelSettings 检查发送给模型的实际设置
使用 call.request.tools 和 call.request.handoffs 检查已准备好的能力
使用 call.streamed 区分流式传输和非流式传输路径

测试流式传输

常规响应步骤既适用于流式运行，也适用于非流式运行。当 run(..., { stream: true }) 调用模型时，ScriptedModel 会发出规范化的开始事件、助手文本增量，以及包含完整输出和用量的终止响应事件。

自动生成的流事件是独立快照。修改已产出的增量不会改变后续的终止响应事件。

仅当精确的规范化 StreamEvent 序列本身属于被测行为时，才使用 modelStream(events)。当该序列依赖于记录的调用时，使用 modelStreamResponder(callback)。这些事件是 SDK 规范化的事件，而不是 Responses 或 Chat Completions 的线上传输分块。

如果非流式调用使用了原始流步骤，就会抛出 InvalidScriptedModelStepError，其中包含 reason: 'incompatible_call'。

注入模型故障

使用 modelError(error, retryAdvice?) 使一次模型调用失败。重试建议可以是固定的 ModelRetryAdvice 值，也可以是回调。运行器执行的重试是另一次模型调用，会使用脚本中的下一个步骤。

modelError(error)
modelError(error, { suggested: true, replaySafety: 'safe' })
modelError(error, ({ attempt }) => ({ suggested: attempt === 1 }))

上文展示的是签名而非 TypeScript 示例，因为重试建议是否会触发再次尝试，取决于外围运行器的重试策略。

已中止的请求不会使用任何步骤。记录的调用到达中止检查点后，如果请求已中止，就会抛出名为 AbortError 的错误，其中包含 callIndex。

自定义代码运行期间，取消采用协作方式。ScriptedModel 不会让正在执行的 modelResponder() 或 modelStreamResponder() 回调，或异步迭代器尚未完成的 next()，与请求信号进行竞速。可能无限等待的代码必须自行监听 call.request.signal，或在迭代器内部监听同一个信号，并随后完成执行。当回调或迭代器交还控制权后，ScriptedModel 会在接受响应或发出下一个事件之前再次检查信号。

检测工作流偏移

将脚本中的轮次视为预期的工作流结构。下面的测试在两轮订单查询时通过，并在清理阶段执行 assertComplete()，因此即使其他断言或工作流本身失败，也会检查脚本。

例如，如果应用代码后来将 lookup_order 改为需要审批，运行会在第一个脚本响应之后停止。清理阶段随后会报告一个未使用的步骤。如果编排发生变化，导致在预期步骤之后又发出一次模型请求，模型会在该请求处失败。因此，现有回归测试无需让真实模型推断新路径，就能捕获智能体控制流的变化。

import { Agent, type Model, Runner, tool } from '@openai/agents';
import {
  ScriptedModel,
  assistantMessage,
  functionCall,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';
import { z } from 'zod';

function createOrderAgent(model: Model) {
  const lookupOrder = tool({
    name: 'lookup_order',
    description: 'Looks up an order.',
    parameters: z.object({ orderId: z.string() }),
    execute: async ({ orderId }) => `${orderId}: shipped`,
  });

  return new Agent({
    name: 'Order assistant',
    model,
    tools: [lookupOrder],
  });
}

test('preserves the order lookup workflow contract', async (t) => {
  const model = new ScriptedModel([
    [
      functionCall(
        'lookup_order',
        { orderId: 'order_123' },
        { callId: 'call_1' },
      ),
    ],
    [assistantMessage('Order 123 has shipped.')],
  ]);
  // Keep this assertion in teardown so an early workflow exit is reported
  // even when another assertion fails first.
  t.after(() => model.assertComplete());
  const agent = createOrderAgent(model);

  // ScriptedModel replaces model I/O; disable tracing separately so this test
  // makes no network requests.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'Where is order 123?');

  assert.equal(result.finalOutput, 'Order 123 has shipped.');
  assert.equal(model.calls.length, 2);
});

不要在常规回归测试中捕获这些不匹配错误。应让测试失败，并使用错误类型和字段诊断预期工作流的哪一侧发生了变化。

错误	结构化字段	含义
`UnexpectedModelCallError`	`callIndex`、`streamed`	脚本结束后，工作流又进行了一次模型调用
`UnconsumedModelStepsError`	`remainingSteps`	工作流在使用完所有步骤之前结束
`InvalidScriptedModelStepError`	`reason`、`inputIndex?`、`callIndex?`、`stepType?`	步骤格式有误，或与调用模式不兼容
`ScriptedModelRequestAbortedError`	`callIndex?`	请求在脚本化操作之前或期间被中止

构造函数或 enqueue() 会在模型调用使用步骤之前拒绝格式有误的步骤封装。这样，配置错误就会在靠近测试定义的位置暴露，而不是在后续工作流中才失败。

沙盒智能体方案

测试沙盒智能体工作流

将 ScriptedModel 与 scriptedSandboxSession() 结合使用，即可在不创建 Docker 容器或远程沙盒的情况下测试真实的 SandboxAgent 运行时。模型脚本选择能力工具，沙盒脚本则定义对应 SandboxSession 方法的返回内容。

import { Runner } from '@openai/agents';
import { SandboxAgent, shell } from '@openai/agents/sandbox';
import {
  ScriptedModel,
  assistantMessage,
  functionCall,
  scriptedSandboxSession,
} from '@openai/agents/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('runs a SandboxAgent shell workflow without a real sandbox', async (t) => {
  const sandbox = scriptedSandboxSession([
    {
      method: 'execCommand',
      match: ({ cmd }) => {
        // Assert the command at the sandbox boundary, after the shell tool has
        // validated and converted the model's function-call arguments.
        assert.equal(cmd, 'pwd');
      },
      result: '/workspace\n',
    },
  ]);
  const model = new ScriptedModel([
    [functionCall('exec_command', { cmd: 'pwd' }, { callId: 'call_1' })],
    [assistantMessage('The workspace is /workspace.')],
  ]);
  const agent = new SandboxAgent({
    name: 'Workspace assistant',
    model,
    capabilities: [shell()],
  });

  // Teardown assertions also report unused steps if the workflow exits early.
  t.after(() => {
    sandbox.assertComplete();
    model.assertComplete();
  });

  // The scripted boundaries do not disable the separate tracing exporter.
  const runner = new Runner({ tracingDisabled: true });
  const result = await runner.run(agent, 'Which directory are you in?', {
    // The real SandboxAgent runtime receives the in-memory session instead of
    // creating a Docker or remote sandbox.
    sandbox: { session: sandbox },
  });

  assert.equal(result.finalOutput, 'The workspace is /workspace.');
  assert.deepEqual(
    sandbox.calls.map((call) => call.method),
    ['execCommand'],
  );
  assert.equal(model.calls.length, 2);
});

此测试跨越 SDK 负责的两个规范化边界。ScriptedModel 驱动面向模型的轮次，脚本化沙盒接收真实 shell 能力生成的 execCommand 参数。因此，该测试涵盖工具参数验证、能力路由、沙盒会话调用、向下一轮模型调用传递工具结果，以及最终输出处理。

它不测试真实模型是否会选择该命令、沙盒提供方是否能成功启动，或该提供方如何执行命令。请使用评估来检验模型决策质量，并使用真实沙盒提供方的集成测试来检验提供方生命周期、文件系统、进程和隔离行为。

配置沙盒步骤

每个步骤对应一次 SandboxSession 方法调用。请只选择一种结果，并且仅在参数重要时添加匹配器：

步骤成员	使用场景
`result`	方法应返回固定值
`respond(call)`	结果取决于带类型的调用参数或调用索引
`error`	方法应抛出指定错误，或以该错误拒绝
`match(...args)`	测试应在产生结果之前拒绝非预期的方法参数

沙盒方法按一个全局顺序匹配，包括 supportsPty() 等同步方法和 execCommand() 等异步方法。返回的会话只会安装脚本中指定的方法。这可以保留能力检测行为：例如，当工作流应公开交互式 shell 支持时，需要同时为 supportsPty 和 writeStdin 编写脚本。

sandbox.calls 记录调用时的独立参数快照，包含从零开始的 index（与每次调用在数组中的位置一致）、method 和位置参数 args。诊断失败时可使用 sandbox.remainingSteps，并在测试结束时调用 sandbox.assertComplete()，以免工作流提前退出而悄然留下未使用的预期操作。

错误	结构化字段	含义
`UnexpectedSandboxCallError`	`callIndex`、`actualMethod`、`expectedMethod`、`remainingSteps`	工作流调用了错误的方法，或在脚本结束后继续调用
`SandboxCallMatcherError`	`callIndex`、`method`	某个步骤的匹配器返回了 `false`
`UnconsumedSandboxStepsError`	`remainingSteps`、`pendingMethods`	工作流在使用完所有步骤之前结束
`InvalidScriptedSandboxStepError`	`reason`、`inputIndex`、`method?`	步骤格式有误，或指定了不受支持的方法

语音智能体与会话方案

测试语音智能体工具工作流

将真实函数工具附加到 RealtimeAgent，然后通过脚本化传输层发出规范化的 function_call。RealtimeSession 会从该智能体解析工具，通过 SDK 工具流水线验证并执行工具，再通过 sendFunctionCallOutput 发送结果。

import { RealtimeAgent, RealtimeSession, tool } from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';
import { z } from 'zod';

test('executes a RealtimeAgent tool call', async () => {
  let lookedUpOrderId: string | undefined;
  const lookupOrder = tool({
    name: 'lookup_order',
    description: 'Looks up an order by ID.',
    parameters: z.object({ orderId: z.string() }),
    execute: async ({ orderId }) => {
      lookedUpOrderId = orderId;
      return `Order ${orderId} has shipped.`;
    },
  });
  const agent = new RealtimeAgent({
    name: 'Order assistant',
    instructions: 'Help customers track their orders.',
    tools: [lookupOrder],
  });
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(agent, { transport });
  let markToolOutputReturned!: () => void;
  let rejectToolOutput!: (reason?: unknown) => void;
  const toolOutputReturned = new Promise<void>((resolve, reject) => {
    markToolOutputReturned = resolve;
    rejectToolOutput = reject;
  });
  // Observe a scenario failure that happens before exercise reaches its await.
  // Awaiting the original promise still receives the same rejection.
  void toolOutputReturned.catch(() => undefined);

  await transport.runScenario({
    // Tool execution is asynchronous. Bound the scenario so a missing tool
    // output fails here instead of waiting for the test runner's timeout.
    signal: AbortSignal.timeout(5_000),
    scenario: async ({ expectCall, emit }) => {
      try {
        await expectCall('connect');
        await expectCall('sendMessage', (call) => {
          assert.equal(call.message, 'Where is order 123?');
        });

        // Script the model choosing the agent's tool. This tests how the SDK
        // handles that choice, not whether a real model would make the choice.
        emit('turn_started', {
          type: 'response_started',
          providerData: { response: { id: 'response_1' } },
        });
        emit('function_call', {
          type: 'function_call',
          name: 'lookup_order',
          callId: 'call_1',
          arguments: JSON.stringify({ orderId: 'order_123' }),
          responseId: 'response_1',
        });

        // RealtimeSession runs the actual tool and sends its result back through
        // the transport so the model could continue the response.
        await expectCall('sendFunctionCallOutput', (call) => {
          assert.equal(call.toolCall.callId, 'call_1');
          assert.equal(call.output, 'Order order_123 has shipped.');
          assert.equal(call.startResponse, true);
        });
        markToolOutputReturned();
        await expectCall('close');
      } catch (error) {
        // Unblock exercise so its finally block can close the session.
        rejectToolOutput(error);
        throw error;
      }
    },
    exercise: async () => {
      try {
        await session.connect({ apiKey: 'test' });
        session.sendMessage('Where is order 123?');
        await toolOutputReturned;
      } finally {
        // Clean up even when the scenario fails or reaches its time limit.
        session.close();
      }
    },
  });

  assert.equal(lookedUpOrderId, 'order_123');
  transport.assertComplete();
  transport.assertClosed();
});

这是一个智能体测试，因为工具及其业务逻辑属于 RealtimeAgent。脚本化的 function_call 代表模型决定使用该工具。它不能证明真实模型会遵循智能体的指令或选择该工具；请通过评估或集成测试来检验模型决策质量。

测试操作会等待异步工具输出后再关闭会话，因此此代码示例传入了一个 AbortSignal 来限制等待时间。有序调用不匹配时仍会立即失败；该信号用于处理回归导致预期工具输出根本未发送的情况。

使用相同模式测试智能体负责的交接、工具审批和工具护栏：发出用于启动该行为的规范化输入，然后断言真实 SDK 流水线产生的会话事件和出站传输调用。

测试 Realtime 会话

将 ScriptedRealtimeTransport 传入真实的 RealtimeSession。使用 runScenario() 同时运行预期的传输交互和会话测试操作，无需依赖测试运行器的超时。下面的代码示例模拟两轮用户交互，并验证传输层发出的规范化项会更新真实会话历史。

import {
  RealtimeAgent,
  type RealtimeItem,
  RealtimeSession,
} from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

function createScenarioGate() {
  let release!: () => void;
  let fail!: (reason: unknown) => void;
  const promise = new Promise<void>((resolve, reject) => {
    release = resolve;
    fail = reject;
  });
  // Observe a scenario failure that happens before exercise reaches its await.
  // Awaiting the original promise still receives the same rejection.
  void promise.catch(() => undefined);
  return { promise, release, fail };
}

test('runs a two-turn Realtime conversation', async () => {
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(
    new RealtimeAgent({ name: 'Assistant' }),
    { transport },
  );
  const conversation: RealtimeItem[] = [
    {
      itemId: 'user_1',
      type: 'message',
      role: 'user',
      status: 'completed',
      content: [{ type: 'input_text', text: 'Hello' }],
    },
    {
      itemId: 'assistant_1',
      type: 'message',
      role: 'assistant',
      status: 'completed',
      content: [{ type: 'output_text', text: 'Hi! How can I help?' }],
    },
    {
      itemId: 'user_2',
      type: 'message',
      role: 'user',
      status: 'completed',
      content: [{ type: 'input_text', text: 'What did I just say?' }],
    },
    {
      itemId: 'assistant_2',
      type: 'message',
      role: 'assistant',
      status: 'completed',
      content: [{ type: 'output_text', text: 'You said hello.' }],
    },
  ];
  const signal = AbortSignal.timeout(5_000);
  const firstTurnDelivered = createScenarioGate();
  let secondTurnDelivered: ReturnType<typeof createScenarioGate> | undefined;

  await transport.runScenario({
    // The application waits for each scripted turn. Bound that coordination
    // so a missing outbound call cannot leave both callbacks waiting forever.
    signal,
    // This side scripts the normalized transport-facing interaction.
    scenario: async ({ expectCall, emit }) => {
      try {
        await expectCall('connect');

        await expectCall('sendMessage', (call) => {
          assert.equal(call.message, 'Hello');
        });
        // Emit normalized history items to exercise RealtimeSession history.
        emit('item_update', conversation[0]);
        emit('item_update', conversation[1]);
        // Create the next gate before releasing the application to send again.
        secondTurnDelivered = createScenarioGate();
        firstTurnDelivered.release();

        await expectCall('sendMessage', (call) => {
          assert.equal(call.message, 'What did I just say?');
        });
        emit('item_update', conversation[2]);
        emit('item_update', conversation[3]);
        secondTurnDelivered.release();

        await expectCall('close');
      } catch (error) {
        // Forward scenario failures to whichever application wait is active.
        (secondTurnDelivered ?? firstTurnDelivered).fail(error);
        throw error;
      }
    },
    // This side drives the application through the public session API.
    exercise: async () => {
      try {
        await session.connect({ apiKey: 'test' });
        session.sendMessage('Hello');
        await firstTurnDelivered.promise;

        session.sendMessage('What did I just say?');
        assert(secondTurnDelivered);
        await secondTurnDelivered.promise;
      } finally {
        // Close the session after success, timeout, or scenario failure.
        session.close();
      }
    },
  });

  assert.deepEqual(session.history, conversation);
  transport.assertComplete();
  transport.assertClosed();
});

scenario 回调为面向传输层的一侧编写脚本：匹配出站调用并发出规范化的入站事件。exercise 回调像应用代码一样操作，驱动公开的会话 API。代码示例中的注释明确标出了这一边界。此测试检验的是 RealtimeSession 行为，而不是提供方服务器帧或转换行为。

应用会等待每个脚本化轮次完成后再发送下一条消息，因此此代码示例向 runScenario() 传入了一个有时间限制的 AbortSignal。如果信号到期，runScenario() 会拒绝尚未完成的预期，场景会将该失败转发给当前启用的、仅供测试使用的协调关卡，而测试操作会执行会话清理，避免两个回调都陷入阻塞。

expectCall(method, matcher?) 的类型由方法决定，并按一个全局顺序匹配调用。匹配器可以包含断言，也可以返回 false 来拒绝调用。如果调用恰好发生在对应预期注册之前，传输层会保留该调用，供下一个预期匹配。

任一侧失败、AbortSignal 取消场景，或测试操作结束但仍有未完成的预期时，runScenario() 都会终止。测试操作结束时如果仍有剩余预期，就会产生 IncompleteRealtimeScenarioError。当相互等待的回调达到明确指定的信号截止时间时，如上面的代码示例所示，runScenario() 会产生 ScriptedRealtimeScenarioCancelledError，并报告尚未完成的预期。

测试中断

从与应用相关的方向测试中断：

要测试停止按钮或类似的应用操作，先发出助手音频响应的开始事件，等待会话观察到该事件，然后调用 session.interrupt()，并预期出站 interrupt 调用。
要测试应用如何处理中断通知，从传输层发出 audio_interrupted，并对会话监听器进行断言。

此代码示例将两者保留为独立测试，以免暗示其中一个事件会自动触发另一个。

import { RealtimeAgent, RealtimeSession } from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

function createScenarioGate() {
  let release!: () => void;
  let fail!: (reason: unknown) => void;
  const promise = new Promise<void>((resolve, reject) => {
    release = resolve;
    fail = reject;
  });
  // Observe a scenario failure that happens before exercise reaches its await.
  // Awaiting the original promise still receives the same rejection.
  void promise.catch(() => undefined);
  return { promise, release, fail };
}

function createSession() {
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(
    new RealtimeAgent({ name: 'Assistant' }),
    { transport },
  );
  return { session, transport };
}

test('sends an interrupt request through the transport', async () => {
  const { session, transport } = createSession();
  const signal = AbortSignal.timeout(5_000);
  const readyToInterrupt = createScenarioGate();
  let sessionObservedAudio = false;
  session.once('audio_start', () => {
    sessionObservedAudio = true;
  });

  await transport.runScenario({
    signal,
    scenario: async ({ expectCall, emit }) => {
      try {
        await expectCall('connect');
        await expectCall('sendMessage');

        // Simulate the beginning of an assistant utterance. RealtimeSession
        // emits audio_start when it receives the first audio chunk.
        emit('turn_started', {
          type: 'response_started',
          providerData: { response: { id: 'response_1' } },
        });
        emit('audio', {
          type: 'audio',
          responseId: 'response_1',
          data: new Uint8Array([1, 2, 3]).buffer,
        });
        // emit() delivers events synchronously. Verify that RealtimeSession saw
        // the chunk before releasing the concurrently running application side.
        assert.equal(sessionObservedAudio, true);
        readyToInterrupt.release();

        await expectCall('interrupt');
        await expectCall('close');
      } catch (error) {
        readyToInterrupt.fail(error);
        throw error;
      }
    },
    exercise: async () => {
      try {
        await session.connect({ apiKey: 'test' });
        session.sendMessage('Tell me a long story');
        await readyToInterrupt.promise;

        // Application code, such as a stop button, interrupts active playback.
        session.interrupt();
      } finally {
        session.close();
      }
    },
  });

  transport.assertComplete();
  transport.assertClosed();
});

test('forwards an interruption notification to the application', async () => {
  const { session, transport } = createSession();
  let interruptionObserved = false;
  session.once('audio_interrupted', () => {
    interruptionObserved = true;
  });

  const connectCall = transport.expectCall('connect');
  await session.connect({ apiKey: 'test' });
  await connectCall;

  // Independently, the transport can notify the session of an interruption.
  transport.emit('audio_interrupted');
  assert.equal(interruptionObserved, true);

  const closeCall = transport.expectCall('close');
  session.close();
  await closeCall;
  transport.assertComplete();
  transport.assertClosed();
});

注入 Realtime 故障

failNextCall(method, error) 会使该方法的下一次调用抛出所提供的值。该次尝试仍会被记录并参与匹配，因此测试可以同时断言操作和失败。

import { RealtimeAgent, RealtimeSession } from '@openai/agents/realtime';
import { ScriptedRealtimeTransport } from '@openai/agents/realtime/testing';
import assert from 'node:assert/strict';
import test from 'node:test';

test('injects a transport failure into the next matching call', async () => {
  const transport = new ScriptedRealtimeTransport();
  const session = new RealtimeSession(
    new RealtimeAgent({ name: 'Assistant' }),
    { transport },
  );
  const failure = new Error('connection failed');

  // Failure injection and call matching are independent: the failed attempt
  // is still recorded and must satisfy the expectation.
  transport.failNextCall('connect', failure);
  const connectCall = transport.expectCall('connect');

  await assert.rejects(session.connect({ apiKey: 'test' }), failure);
  assert.equal((await connectCall).method, 'connect');
  assert.equal(transport.status, 'disconnected');
  transport.assertComplete();
  transport.assertClosed();
});

为同一方法注册的多个故障会按注册顺序使用。如果仍有未使用的注入故障，assertComplete() 就会失败。

测试远端断开连接

调用 transport.disconnect(error?) 来模拟远端关闭连接。它会将传输层状态切换为 disconnected，发出 connection_change，并可选地发出传输层 error。

对于连接丢失和远端关闭，使用 disconnect()。当应用发起清理，且测试应预期出站 close 调用时，使用 session.close()。

检查 Realtime 调用

transport.calls 按调用顺序包含独立快照。可变数组、普通对象和缓冲区会在调用边界处复制，而运行时类实例保持对象身份不变。记录的 connect 调用包含 apiKeyProvided，但绝不会存储 API 密钥。

常见方法名称包括 connect、sendEvent、requestResponse、sendMessage、addImage、sendAudio、updateSessionConfig、close、mute、sendFunctionCallOutput、interrupt、resetHistory 和 sendMcpResponse。导出的方法和调用类型派生自 RealtimeTransportLayer，因此新增的传输方法不会悄然从带类型的测试接口中遗漏。

验证完成状态和清理

Realtime 测试结束时应执行以下两个断言：

断言	检查内容
`transport.assertComplete()`	不存在未匹配的调用、尚未完成的预期或未使用的注入故障
`transport.assertClosed()`	生命周期状态为 `disconnected`

Realtime 错误还会公开结构化状态：

错误	结构化字段
`UnexpectedRealtimeCallError`	`callIndex`、`expectedMethod`、`actualMethod`
`RealtimeCallMatcherError`	`callIndex`、`actualMethod`
`IncompleteRealtimeScenarioError`	`unconsumedCalls`、`pendingExpectations`、`pendingFailures`
`ScriptedRealtimeScenarioCancelledError`	`pendingExpectations`
`RealtimeTransportNotClosedError`	`status`
`ScriptedRealtimeConnectionSupersededError`	`operation`

API 速查表

模型步骤工厂

工厂	使用场景
`modelResponse(response)`	需要显式的响应步骤；也可以直接传入输出数组
`modelResponder(callback)`	响应取决于记录的调用
`modelError(error, retryAdvice?)`	某次模型调用应失败
`modelStream(events)`	需要精确的规范化流
`modelStreamResponder(callback)`	精确的流取决于记录的调用

常用输出辅助函数

辅助函数	产物
`assistantMessage(text, options?)`	一条已完成的助手文本消息
`functionCall(name, arguments, options)`	一次已完成的函数调用；`options.callId` 为必填项

对于推理、托管工具、图像、音频，或故意构造的异常协议场景，请直接传入规范化输出项。辅助函数集合刻意保持精简，以免测试模块变成第二个提供方转换层。

沙盒会话控制项

成员	用途
`scriptedSandboxSession(steps)`	创建内存中的 `SandboxSession`，包含脚本中指定的方法
`match(...args)`	验证一次预期方法调用的带类型参数
`result`	返回固定的方法结果
`respond(call)`	根据带类型的已记录调用计算结果
`error`	抛出注入的错误，或以该错误拒绝
`calls`	按调用顺序检查独立调用快照
`assertComplete()`	查找工作流未使用的沙盒步骤

Realtime 控制项

方法	用途
`runScenario({ scenario, exercise, signal? })`	同时运行两侧；当仅供测试使用的协调机制可能让双方都陷入等待时，传入 `signal`
`expectCall(method, matcher?)`	匹配下一次出站传输调用
`emit(event, ...args)`	同步传递带类型的入站事件
`failNextCall(method, error)`	使某个方法的下一次调用失败
`disconnect(error?)`	模拟终止性的远端断开连接
`assertComplete()`	查找未匹配的调用、预期或故障
`assertClosed()`	验证生命周期清理

正确边界的选择

当测试需要在不依赖模型提供方的情况下检验 SDK 运行循环、工具、交接、护栏、会话、重试或流式传输时，使用 ScriptedModel。

当测试需要在不启动沙盒提供方的情况下检验 SandboxAgent 的能力与编排时，将 scriptedSandboxSession() 与 ScriptedModel 配合使用。提供方创建、进程执行、文件系统保真度、持久化和隔离检查应放在针对真实提供方的集成测试中。

当测试需要在不建立 WebRTC 或 WebSocket 连接的情况下检验 RealtimeSession 行为，或 RealtimeAgent 的工具与交接编排时，使用 ScriptedRealtimeTransport。将智能体业务逻辑保留在附加到 RealtimeAgent 的工具、交接和护栏中；将连接、历史记录、音频和中断断言保留在会话与传输边界。

不要使用这些测试替身来测试提供方请求转换、HTTP 或 WebSocket 载荷、身份验证标头、提供方特有的流式传输分块，或沙盒提供方生命周期与隔离。模型线上协议测试应保留真实模型适配器并配合模拟网络传输，沙盒集成测试则应使用真实沙盒提供方。

最终检查清单

仅为规范化模型、沙盒会话或 Realtime 传输边界负责的交互编写脚本。
断言重要的请求字段，而不是运行器的私有状态。
优先使用固定响应步骤；仅对依赖请求的行为使用响应器。
优先使用自动流式传输；仅在事件级行为重要时使用精确的流。
模型测试结束时调用 model.assertComplete()。
沙盒测试结束时调用 sandbox.assertComplete()。
Realtime 测试结束时调用 transport.assertComplete() 和 transport.assertClosed()。
断言结构化错误字段，而不是解析错误消息。
提供方线上协议测试应使用真实适配器配合模拟网络传输。

运行智能体介绍 ScriptedModel 所测试的运行循环。
模型说明 Model 边界和提供方适配器。
快速入门说明沙盒能力、会话和提供方生命周期。
语音智能体概述介绍 RealtimeSession。
传输机制涵盖生产环境传输方式和传输事件。

范围与当前限制

明确排除在测试模块之外的内容

这些工具替换的是 SDK 负责的规范化边界。它们并非用于证明模型或外部提供方负责的行为：

模型质量、指令遵循和工具选择质量应通过评估或使用真实模型的测试来检验。
Responses 和 Chat Completions 请求序列化、身份验证、提供方默认设置、HTTP 行为及提供方流分块，需要使用真实模型适配器配合模拟或受控的网络传输来测试。
沙盒启动、进程执行、文件系统保真度、持久化、资源限制、安全隔离和提供方清理，需要使用真实沙盒提供方进行集成测试。
Realtime 服务器行为、原始 WebSocket 或 WebRTC 帧、音频编码与播放、身份验证和网络恢复，需要真实传输层或集成环境。

脚本化工具不会尝试模拟这些层。在这里添加提供方特有的行为，会让工作流测试依赖于提供方协议的第二套不完整实现。

当前未提供的功能

当前测试模块不提供：

面向每一种规范化模型输出项的便捷构建器。请对常见助手消息和函数调用使用这些小型辅助函数，并直接传入其他规范化项。
高层模拟 Realtime 模型。Realtime 测试会显式匹配出站调用，并发出场景所需的规范化入站事件。
无序的沙盒预期。scriptedSandboxSession() 按一个全局顺序使用方法步骤。
特定于测试运行器的匹配器、测试夹具或自动清理。请使用 Node.js、Vitest、Jest 或其他运行器的断言和清理 API，并显式调用所提供的完成状态和清理断言。
用于提供方线上协议测试的模拟传输。在这些测试中，应保留真实提供方适配器，并模拟其网络边界。

这些内容描述的是当前 API 接口范围，并不承诺在未来版本中添加上述缺失功能。