Merge branch 'main' into realtime-agent-validation

seratch · web-flow · commit 37ea6db2fd69 · 2025-06-06T06:38:02.000+09:00
diff --git a/.changeset/fix-schema-version-serialization.md b/.changeset/fix-schema-version-serialization.md
@@ -0,0 +1,5 @@
+---
+'@openai/agents-core': patch
+---
+
+Export CURRENT_SCHEMA_VERSION constant and use it when serializing run state.
diff --git a/.changeset/wicked-masks-tell.md b/.changeset/wicked-masks-tell.md
@@ -0,0 +1,5 @@
+---
+'@openai/agents-openai': minor
+---
+
+Fix incorrect handling of chat completions mode for handoff
diff --git a/.changeset/wise-results-mate.md b/.changeset/wise-results-mate.md
@@ -0,0 +1,5 @@
+---
+'@openai/agents-realtime': patch
+---
+
+fix: avoid realtime guardrail race condition and detect ongoing response
diff --git a/examples/docs/voice-agents/delegationAgent.ts b/examples/docs/voice-agents/delegationAgent.ts
@@ -27,4 +27,5 @@ const agent = new RealtimeAgent({
   name: 'Customer Support',
   instructions:
     'You are a customer support agent. If you receive any requests for refunds, you need to delegate to your supervisor.',
+  tools: [refundSupervisor],
 });
diff --git a/examples/realtime-next/src/app/websocket/page.tsx b/examples/realtime-next/src/app/websocket/page.tsx
@@ -7,6 +7,7 @@ import {
   TransportEvent,
   RealtimeItem,
   OutputGuardrailTripwireTriggered,
+  RealtimeOutputGuardrail,
 } from '@openai/agents/realtime';
 import { useEffect, useRef, useState } from 'react';
 import { z } from 'zod';
@@ -26,6 +27,21 @@ const refundBackchannel = tool({
   },
 });
 
+const guardrails: RealtimeOutputGuardrail[] = [
+  {
+    name: 'No mention of Dom',
+    execute: async ({ agentOutput }) => {
+      const domInOutput = agentOutput.includes('Dom');
+      return {
+        tripwireTriggered: domInOutput,
+        outputInfo: {
+          domInOutput,
+        },
+      };
+    },
+  },
+];
+
 const agent = new RealtimeAgent({
   name: 'Greeter',
   instructions:
@@ -48,6 +64,7 @@ export default function Home() {
   useEffect(() => {
     session.current = new RealtimeSession(agent, {
       transport: 'websocket',
+      outputGuardrails: guardrails,
     });
     recorder.current = new WavRecorder({ sampleRate: 24000 });
     player.current = new WavStreamPlayer({ sampleRate: 24000 });
@@ -87,6 +104,7 @@ export default function Home() {
   async function connect() {
     if (isConnected) {
       await session.current?.close();
+      await player.current?.interrupt();
       await recorder.current?.end();
       setIsConnected(false);
     } else {
diff --git a/packages/agents-core/src/runState.ts b/packages/agents-core/src/runState.ts
@@ -37,7 +37,7 @@ import { safeExecute } from './utils/safeExecute';
  * run state is compatible with the current version of the SDK.
  * If anything in this schema changes, the version will have to be incremented.
  */
-const CURRENT_SCHEMA_VERSION = '1.0' as const;
+export const CURRENT_SCHEMA_VERSION = '1.0' as const;
 const $schemaVersion = z.literal(CURRENT_SCHEMA_VERSION);
 
 const serializedAgentSchema = z.object({
@@ -349,7 +349,7 @@ export class RunState<TContext, TAgent extends Agent<any, any>> {
    */
   toJSON(): z.infer<typeof SerializedRunState> {
     const output = {
-      $schemaVersion: '1.0',
+      $schemaVersion: CURRENT_SCHEMA_VERSION,
       currentTurn: this._currentTurn,
       currentAgent: {
         name: this._currentAgent.name,
diff --git a/packages/agents-core/test/runState.test.ts b/packages/agents-core/test/runState.test.ts
@@ -4,6 +4,7 @@ import {
   buildAgentMap,
   deserializeModelResponse,
   deserializeItem,
+  CURRENT_SCHEMA_VERSION,
 } from '../src/runState';
 import { RunContext } from '../src/runContext';
 import { Agent } from '../src/agent';
@@ -35,7 +36,7 @@ describe('RunState', () => {
     const agent = new Agent({ name: 'Agent1' });
     const state = new RunState(context, 'input1', agent, 2);
     const json = state.toJSON();
-    expect(json.$schemaVersion).toBe('1.0');
+    expect(json.$schemaVersion).toBe(CURRENT_SCHEMA_VERSION);
     expect(json.currentTurn).toBe(0);
     expect(json.currentAgent).toEqual({ name: 'Agent1' });
     expect(json.originalInput).toEqual('input1');
@@ -65,7 +66,7 @@ describe('RunState', () => {
     expect(
       async () => await RunState.fromString(agent, JSON.stringify(jsonVersion)),
     ).rejects.toThrow(
-      'Run state schema version 0.1 is not supported. Please use version 1.0',
+      `Run state schema version 0.1 is not supported. Please use version ${CURRENT_SCHEMA_VERSION}`,
     );
   });
 
diff --git a/packages/agents-openai/src/openaiChatCompletionsModel.ts b/packages/agents-openai/src/openaiChatCompletionsModel.ts
@@ -67,7 +67,7 @@ export class OpenAIChatCompletionsModel implements Model {
     const output: protocol.OutputModelItem[] = [];
     if (response.choices && response.choices[0]) {
       const message = response.choices[0].message;
-      if (message.content !== undefined) {
+      if (message.content !== undefined && message.content !== null) {
         const { content, ...rest } = message;
         output.push({
           id: response.id,
diff --git a/packages/agents-realtime/src/openaiRealtimeBase.ts b/packages/agents-realtime/src/openaiRealtimeBase.ts
@@ -256,6 +256,7 @@ export abstract class OpenAIRealtimeBase
           type: 'transcript_delta',
           delta: parsed.delta,
           itemId: parsed.item_id,
+          responseId: parsed.response_id,
         });
       }
       // no support for partial transcripts yet.
diff --git a/packages/agents-realtime/src/openaiRealtimeEvents.ts b/packages/agents-realtime/src/openaiRealtimeEvents.ts
@@ -6,15 +6,18 @@ import type { MessageEvent as WebSocketMessageEvent } from 'ws';
 // provide better runtime validation when parsing events from the server.
 
 export const realtimeResponse = z.object({
-  id: z.string().optional(),
-  conversation_id: z.string().optional(),
-  max_output_tokens: z.number().or(z.literal('inf')).optional(),
+  id: z.string().optional().nullable(),
+  conversation_id: z.string().optional().nullable(),
+  max_output_tokens: z.number().or(z.literal('inf')).optional().nullable(),
   metadata: z.record(z.string(), z.any()).optional().nullable(),
-  modalities: z.array(z.string()).optional(),
-  object: z.literal('realtime.response').optional(),
-  output: z.array(z.any()).optional(),
-  output_audio_format: z.string().optional(),
-  status: z.enum(['completed', 'incomplete', 'failed', 'cancelled']).optional(),
+  modalities: z.array(z.string()).optional().nullable(),
+  object: z.literal('realtime.response').optional().nullable(),
+  output: z.array(z.any()).optional().nullable(),
+  output_audio_format: z.string().optional().nullable(),
+  status: z
+    .enum(['completed', 'incomplete', 'failed', 'cancelled', 'in_progress'])
+    .optional()
+    .nullable(),
   status_details: z.record(z.string(), z.any()).optional().nullable(),
   usage: z
     .object({
@@ -26,8 +29,9 @@ export const realtimeResponse = z.object({
         .optional()
         .nullable(),
     })
-    .optional(),
-  voice: z.string().optional(),
+    .optional()
+    .nullable(),
+  voice: z.string().optional().nullable(),
 });
 
 // Basic content schema used by ConversationItem.
@@ -315,7 +319,6 @@ export const responseDoneEventSchema = z.object({
   type: z.literal('response.done'),
   event_id: z.string(),
   response: realtimeResponse,
-  test: z.boolean(),
 });
 
 export const responseFunctionCallArgumentsDeltaEventSchema = z.object({
diff --git a/packages/agents-realtime/src/openaiRealtimeWebRtc.ts b/packages/agents-realtime/src/openaiRealtimeWebRtc.ts
@@ -196,6 +196,7 @@ export class OpenAIRealtimeWebRTC
           if (!parsed || isGeneric) {
             return;
           }
+
           if (parsed.type === 'response.created') {
             this.#ongoingResponse = true;
           } else if (parsed.type === 'response.done') {
@@ -334,6 +335,7 @@ export class OpenAIRealtimeWebRTC
       this.sendEvent({
         type: 'response.cancel',
       });
+      this.#ongoingResponse = false;
     }
 
     this.sendEvent({
diff --git a/packages/agents-realtime/src/openaiRealtimeWebsocket.ts b/packages/agents-realtime/src/openaiRealtimeWebsocket.ts
@@ -328,6 +328,7 @@ export class OpenAIRealtimeWebSocket
       this.sendEvent({
         type: 'response.cancel',
       });
+      this.#ongoingResponse = false;
     }
   }
 
@@ -367,8 +368,6 @@ export class OpenAIRealtimeWebSocket
     this._cancelResponse();
 
     const elapsedTime = Date.now() - this._firstAudioTimestamp;
-    console.log(`Interrupting response after ${elapsedTime}ms`);
-    console.log(`Audio length: ${this._audioLengthMs}ms`);
     if (elapsedTime >= 0 && elapsedTime < this._audioLengthMs) {
       this._interrupt(elapsedTime);
     }
diff --git a/packages/agents-realtime/src/realtimeSession.ts b/packages/agents-realtime/src/realtimeSession.ts
@@ -180,6 +180,7 @@ export class RealtimeSession<
   #transcribedTextDeltas: Record<string, string> = {};
   #history: RealtimeItem[] = [];
   #shouldIncludeAudioData: boolean;
+  #interruptedByGuardrail: Record<string, boolean> = {};
 
   constructor(
     public readonly initialAgent:
@@ -446,7 +447,7 @@ export class RealtimeSession<
     }
   }
 
-  async #runOutputGuardrails(output: string) {
+  async #runOutputGuardrails(output: string, responseId: string) {
     if (this.#outputGuardrails.length === 0) {
       return;
     }
@@ -460,24 +461,28 @@ export class RealtimeSession<
       this.#outputGuardrails.map((guardrail) => guardrail.run(guardrailArgs)),
     );
 
-    for (const result of results) {
-      if (result.output.tripwireTriggered) {
-        const error = new OutputGuardrailTripwireTriggered(
-          `Output guardrail triggered: ${JSON.stringify(result.output.outputInfo)}`,
-          result,
-        );
-        this.emit(
-          'guardrail_tripped',
-          this.#context,
-          this.#currentAgent,
-          error,
-        );
-        this.interrupt();
-
-        const feedbackText = getRealtimeGuardrailFeedbackMessage(result);
-        this.sendMessage(feedbackText);
-        break;
+    const firstTripwireTriggered = results.find(
+      (result) => result.output.tripwireTriggered,
+    );
+    if (firstTripwireTriggered) {
+      // this ensures that if one guardrail already trips and we are in the middle of another
+      // guardrail run, we don't trip again
+      if (this.#interruptedByGuardrail[responseId]) {
+        return;
       }
+      this.#interruptedByGuardrail[responseId] = true;
+      const error = new OutputGuardrailTripwireTriggered(
+        `Output guardrail triggered: ${JSON.stringify(firstTripwireTriggered.output.outputInfo)}`,
+        firstTripwireTriggered,
+      );
+      this.emit('guardrail_tripped', this.#context, this.#currentAgent, error);
+      this.interrupt();
+
+      const feedbackText = getRealtimeGuardrailFeedbackMessage(
+        firstTripwireTriggered,
+      );
+      this.sendMessage(feedbackText);
+      return;
     }
   }
 
@@ -498,7 +503,7 @@ export class RealtimeSession<
       this.emit('agent_end', this.#context, this.#currentAgent, textOutput);
       this.#currentAgent.emit('agent_end', this.#context, textOutput);
 
-      this.#runOutputGuardrails(textOutput);
+      this.#runOutputGuardrails(textOutput, event.response.id);
     });
 
     this.#transport.on('audio_done', () => {
@@ -511,6 +516,7 @@ export class RealtimeSession<
       try {
         const delta = event.delta;
         const itemId = event.itemId;
+        const responseId = event.responseId;
         if (lastItemId !== itemId) {
           lastItemId = itemId;
           lastRunIndex = 0;
@@ -531,7 +537,7 @@ export class RealtimeSession<
           // We don't cancel existing runs because we want the first one to fail to fail
           // The transport layer should upon failure handle the interruption and stop the model
           // from generating further
-          this.#runOutputGuardrails(newText);
+          this.#runOutputGuardrails(newText, responseId);
         }
       } catch (err) {
         this.emit('error', {
@@ -672,6 +678,7 @@ export class RealtimeSession<
    * Disconnect from the session.
    */
   close() {
+    this.#interruptedByGuardrail = {};
     this.#transport.close();
   }
 
diff --git a/packages/agents-realtime/src/transportLayerEvents.ts b/packages/agents-realtime/src/transportLayerEvents.ts
@@ -34,6 +34,7 @@ export type TransportLayerTranscriptDelta = {
   type: 'transcript_delta';
   itemId: string;
   delta: string;
+  responseId: string;
 };
 
 export type TransportLayerResponseCompleted =
diff --git a/packages/agents-realtime/test/realtimeSession.test.ts b/packages/agents-realtime/test/realtimeSession.test.ts
@@ -171,8 +171,16 @@ describe('RealtimeSession', () => {
       outputGuardrailSettings: { debounceTextLength: 1 },
     });
     await s.connect({ apiKey: 'test' });
-    t.emit('audio_transcript_delta', { delta: 'a', itemId: '1' } as any);
-    t.emit('audio_transcript_delta', { delta: 'a', itemId: '2' } as any);
+    t.emit('audio_transcript_delta', {
+      delta: 'a',
+      itemId: '1',
+      responseId: 'z',
+    } as any);
+    t.emit('audio_transcript_delta', {
+      delta: 'a',
+      itemId: '2',
+      responseId: 'z',
+    } as any);
     await vi.waitFor(() => expect(runMock).toHaveBeenCalledTimes(2));
     vi.restoreAllMocks();
   });

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +'@openai/agents-core': patch
 +---
++
 +Export CURRENT_SCHEMA_VERSION constant and use it when serializing run state.
Original file line number	Diff line number	Diff line change
`@@ -256,6 +256,7 @@ export abstract class OpenAIRealtimeBase`
`256`	`256`	`type: 'transcript_delta',`
`257`	`257`	`delta: parsed.delta,`
`258`	`258`	`itemId: parsed.item_id,`
	`259`	`+ responseId: parsed.response_id,`
`259`	`260`	`});`
`260`	`261`	`}`
`261`	`262`	`// no support for partial transcripts yet.`