テスト

SDK は、エージェントワークフロー、サンドボックスセッション、Realtime セッション向けに、決定的でプロバイダーに依存しないテストダブルを提供します。これらはメモリ内で動作し、モデル、サンドボックスプロバイダー、Realtime APIへのリクエストを行わず、SDK が管理する正規化されたやり取りを記録します。

完全にオフラインのテストでのトレーシングの無効化

スクリプトで定義するユーティリティは、モデル、サンドボックスプロバイダー、Realtime APIの I/O を置き換えます。 トレーシングは SDK の別の境界であり、テスト環境で無効にしない限り、 実行データがエクスポートされる可能性があります。このページのコード例では、new Runner(...) に tracingDisabled: true を渡しています。代わりに、NODE_ENV=test や OPENAI_AGENTS_DISABLE_TRACING=1 を設定するか、テストのセットアップ時に setTracingDisabled(true) を一度呼び出すこともできます。

必要なレシピの選択

やりたいこと	使用するもの	参照先
固定の最終回答を返す	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 を公開します。テスト用シンボルは、メインのランタイムエントリーポイントには含まれません。

エージェントワークフローのレシピ

固定レスポンスの返却

想定されるモデル呼び出しごとに、正規化された出力項目の配列を渡します。出力配列を直接指定する省略記法では、1 回のリクエストに対して決定的なレスポンス 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() で終了してください。これにより、設定したすべてのステップを消費する前にワークフローが停止したケースを検出できます。

ツールワークフローのテスト

ツールを呼び出すモデルレスポンスと、最終回答を生成する 2 番目のレスポンスをスクリプトで定義します。これらのモデル呼び出しの間には、実際の 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	設定されたステップのうち、まだ消費されていない数

各呼び出しには、calls 内の位置に対応するゼロ始まりの index、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の通信チャンクではありません。

ほとんどのワークフローテストでは自動ストリーミングを優先

通常の assistantMessage() レスポンスを使うと、テストが低レベルの イベント構築に依存しなくなります。順序、部分的な出力、 キャンセル、不正なストリームの振る舞いがテスト対象の場合にのみ、厳密なストリームを使用してください。

元のストリームステップを非ストリーミング呼び出しで使用すると、reason: 'incompatible_call' を持つ InvalidScriptedModelStepError がスローされます。

モデルの失敗の注入

modelError(error, retryAdvice?) を使用すると、モデル呼び出しを 1 回失敗させることができます。リトライアドバイスには、固定の ModelRetryAdvice 値またはコールバックを指定できます。ランナーによるリトライは別のモデル呼び出しとなり、スクリプトで定義された次のステップを消費します。

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

上記が TypeScript のコード例ではなくシグネチャを示しているのは、アドバイスによって再試行が行われるかどうかは、周囲のランナーのリトライポリシーで決まるためです。

すでに中止されたリクエストは、ステップを消費しません。記録された呼び出しが中止チェックポイントに到達した後、リクエストが中止されている場合は、callIndex を含む AbortError という名前のエラーがスローされます。

カスタムコードの実行中は、キャンセルは協調的に行われます。ScriptedModel は、実行中の modelResponder() や modelStreamResponder() のコールバック、または非同期イテレーターの保留中の next() と、リクエストシグナルとの間で競合処理を行いません。無期限に待機する可能性のあるコードは、call.request.signal を自身で監視するか、イテレーター内で同じシグナルを監視して、処理を完了させる必要があります。コールバックまたはイテレーターが制御を返すと、ScriptedModel はレスポンスを受け入れる前、または次のイベントを発行する前に、シグナルを再度確認します。

ワークフローの意図しない変化の検出

スクリプトで定義したターンを、期待されるワークフローの形として扱います。次のテストは、2 ターンの注文検索で成功し、ティアダウン中に assertComplete() を実行するため、別のアサーションやワークフロー自体が失敗した場合でも、スクリプトを確認します。

たとえば、後からアプリケーションコードが変更され、lookup_order に承認が必要になると、実行は最初のスクリプトレスポンスの後で停止します。その後、ティアダウンで未消費のステップが 1 つあることが報告されます。オーケストレーションの変更によって、期待されたステップの後に追加のモデルリクエストが行われると、モデルはそのリクエストで失敗します。そのため、既存の回帰テストは、実際のモデルに新しい経路を推論させることなく、エージェントの制御フローの変更を検出できます。

期待するワークフローの形の固定

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 メソッドが返す内容を定義します。

SandboxAgent のシェルワークフローのテスト

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 が管理する 2 つの正規化された境界をまたぎます。ScriptedModel はモデル側のターンを駆動し、スクリプトで定義したサンドボックスは、実際のシェル機能が生成した execCommand の引数を受け取ります。そのため、このテストではツール引数の検証、機能のルーティング、サンドボックスセッションの呼び出し、次のモデルターンへのツール結果の受け渡し、最終出力の処理をカバーします。

実際のモデルがそのコマンドを選択するか、サンドボックスプロバイダーが正常に起動するか、そのプロバイダーがどのようにコマンドを実行するかはテストしません。モデルの判断品質には評価を使用し、プロバイダーのライフサイクル、ファイルシステム、プロセス、分離の振る舞いには、実際のサンドボックスプロバイダーを使う統合テストを使用してください。

サンドボックスステップの設定

各ステップは、SandboxSession メソッドへの呼び出しを 1 回消費します。結果を 1 つだけ選択し、引数が重要な場合にのみマッチャーを追加します。

ステップのメンバー	使用する場面
result	メソッドが固定値を返す必要がある
respond(call)	結果が型付きの呼び出し引数や呼び出しインデックスに依存する
error	メソッドが特定の失敗をスローするか、その失敗で拒否する必要がある
match(...args)	結果を生成する前に、想定外のメソッド引数をテストで拒否する必要がある

サンドボックスのメソッドは、supportsPty() などの同期メソッドと execCommand() などの非同期メソッドを含め、全体で 1 つの順序で照合されます。返されるセッションには、スクリプトで名前を指定したメソッドだけが実装されます。これにより、機能検出が維持されます。たとえば、ワークフローで対話型シェルのサポートを公開する必要がある場合は、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 セッションのテスト

実際の RealtimeSession に ScriptedRealtimeTransport を渡します。runScenario() を使用すると、テストランナーのタイムアウトに依存せず、期待されるトランスポートのやり取りとセッションのテスト処理を一緒に実行できます。次のコード例は、ユーザーの 2 ターンをシミュレートし、トランスポートが発行した正規化された項目によって、実際のセッション履歴が更新されることを検証します。

2 ターンの Realtime 会話のテスト

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 の振る舞いをテストするものであり、プロバイダーのサーバーフレームや変換をテストするものではありません。

アプリケーションは、次のメッセージを送信する前にスクリプトで定義された各ターンを待つため、このコード例では期限付きの AbortSignal を runScenario() に渡しています。シグナルの期限が切れると、runScenario() は保留中の期待を拒否し、シナリオはその失敗をアクティブなテスト専用の同期ゲートに転送します。そして、両方のコールバックをブロックしたままにする代わりに、テスト処理がセッションのクリーンアップを実行します。

expectCall(method, matcher?) はメソッドに応じて型付けされ、全体で 1 つの順序で呼び出しを照合します。マッチャーにはアサーションを含めることができ、false を返して呼び出しを拒否できます。期待が登録される直前に呼び出しが発生した場合、トランスポートは次の期待に備えてその呼び出しを保持します。

runScenario() は、どちらか一方が失敗した場合、AbortSignal がシナリオをキャンセルした場合、または未完了の期待を残してテスト処理が終了した場合に終了します。期待が残ったままテスト処理が終了すると、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();
});

コード例で明示的なシグナルを使用する理由

scenario と exercise は並行して実行されます。readyToInterrupt の Promise は テストの同期用であり、SDK イベントではありません。emit() は同期的であるため、 シナリオはまず実際の RealtimeSession が audio_start を発行したことをアサートし、 その後でテスト処理の待機を解除して session.interrupt() を呼び出せるようにします。これにより、 早すぎる中断と、セッションが音声を検知しなかった場合にテストが ハングすることの両方を防ぎます。audio_interrupted、item_update、item_deleted などの受信通知が テスト対象の振る舞いに含まれる場合は、それらを個別に発行してください。

Realtime の失敗の注入

failNextCall(method, error) は、そのメソッドの次の呼び出しで、指定された値をスローさせます。この呼び出しの試行も記録・照合されるため、テストでは操作と失敗の両方をアサートできます。

次の Realtime 操作の失敗

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();
});

1 つのメソッドに対して複数の失敗を指定した場合、登録順に消費されます。注入した失敗が未使用のまま残ると、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 から導出されるため、新しいトランスポートメソッドが型付きのテスト用 API から気付かれないまま欠落することはありません。

完了とクリーンアップの検証

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?)	モデル呼び出しを 1 回失敗させる必要がある
modelStream(events)	厳密な正規化されたストリームが必要
modelStreamResponder(callback)	厳密なストリームが記録された呼び出しに依存する

共通の出力ヘルパー

ヘルパー	生成するもの
assistantMessage(text, options?)	完了済みのアシスタントのテキストメッセージ
functionCall(name, arguments, options)	完了済みの関数呼び出し。options.callId が必須

推論、組み込みツール（Hosted）、画像、音声、意図的に不正な形式にしたプロトコルのケースについては、正規化された出力項目を直接渡してください。テストモジュールが 2 つ目のプロバイダー変換レイヤーにならないよう、ヘルパーは意図的に少数にとどめています。

サンドボックスセッションの制御

メンバー	目的
scriptedSandboxSession(steps)	スクリプトで指定されたメソッドを持つ、メモリ内の SandboxSession を作成する
match(...args)	期待される 1 回のメソッド呼び出しの型付き引数を検証する
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 フレーム、音声のエンコードと再生、認証、ネットワークの復旧には、実際のトランスポートまたは統合環境が必要です。

スクリプトで定義するユーティリティは、これらのレイヤーのエミュレーションを試みません。ここにプロバイダー固有の振る舞いを追加すると、ワークフローテストが、プロバイダープロトコルの 2 つ目の不完全な実装に依存することになります。

現在提供されていないもの

現在のテストモジュールでは、次のものは提供されていません。

• 正規化されたすべてのモデル出力項目に対応する便利なビルダー。一般的なアシスタントメッセージと関数呼び出しには小さなヘルパーを使用し、それ以外の正規化された項目は直接渡してください。
• 高レベルのシミュレーション用 Realtime モデル。Realtime テストでは、送信呼び出しを明示的に照合し、シナリオに必要な正規化された受信イベントを発行します。
• 順序を問わないサンドボックスの期待。scriptedSandboxSession() は、メソッドのステップを全体で 1 つの順序で消費します。
• テストランナー固有のマッチャー、フィクスチャー、自動ティアダウン。Node.js、Vitest、Jest、その他のランナーのアサーション API とティアダウン API を使用し、提供されている完了とクリーンアップのアサーションを明示的に呼び出してください。
• プロバイダーの通信テスト向けのモックトランスポート。これらのテストでは、実際のプロバイダーアダプターを使用し、そのネットワーク境界をモック化してください。

これらは現在の API の提供範囲を説明するものであり、省略されている機能を将来のリリースで追加するという約束ではありません。