add connection closed message on ws close (#148)

Author: glharper

javascript/standalone/src/client.ts

@@ -9,6 +9,7 @@ import {
   ItemInputAudioTranscriptionFailedMessage,
   MessageRole,
   RealtimeError,
+  ConnectionClosedMessage,
   Response,
   ResponseAudioDeltaMessage,
   ResponseAudioDoneMessage,
@@ -85,6 +86,12 @@ export class LowLevelRTClient {
         }
       },
       serialize: (message: UserMessageType) => JSON.stringify(message),
+      createClosedMessage: (_event): ServerMessageType => ({
+        type: "connection.closed",
+        event_id: generateId("evt", 32),
+        code: _event.code,
+        reason: _event.reason
+      } as ConnectionClosedMessage),
     };
 
     return new WebSocketClient<UserMessageType, ServerMessageType>(

javascript/standalone/src/index.ts

@@ -6,6 +6,7 @@ export {
   AssistantMessageItem,
   AudioFormat,
   ClientMessageBase,
+  ConnectionClosedMessage,
   ErrorMessage,
   FunctionCallItem,
   FunctionCallOutputItem,

javascript/standalone/src/model-utils.ts

@@ -12,6 +12,7 @@ export const isServerMessageType = (
   isRealtimeEvent(message) &&
   [
     "error",
+  "connection.closed",
     "session.created",
     "session.updated",
     "input_audio_buffer.committed",

javascript/standalone/src/models.ts

@@ -210,6 +210,12 @@ export interface ErrorMessage extends ServerMessageBase {
   error: RealtimeError;
 }
 
+export interface ConnectionClosedMessage extends ServerMessageBase {
+  type: "connection.closed";
+  code?: number;
+  reason?: string;
+}
+
 export interface Session {
   id: string;
   model: string;
@@ -547,6 +553,7 @@ export type UserMessageType =
 
 export type ServerMessageType =
   | ErrorMessage
+  | ConnectionClosedMessage
   | SessionCreatedMessage
   | SessionUpdatedMessage
   | InputAudioBufferCommittedMessage

javascript/standalone/src/util/websocket-client.ts

@@ -40,6 +40,8 @@ export type SerializeMessage<T> = (
 export interface MessageProtocolHandler<U, D> {
   validate: ValidateProtocolMessage<D>;
   serialize: SerializeMessage<U>;
+  // Optional: create a synthetic protocol message when the websocket closes
+  createClosedMessage?: (event: CloseEvent) => D;
 }
 
 export class WebSocketClient<U, D> implements AsyncIterable<D> {
@@ -50,6 +52,7 @@ export class WebSocketClient<U, D> implements AsyncIterable<D> {
   private messageQueue: D[] = [];
   private validate: ValidateProtocolMessage<D>;
   private serialize: SerializeMessage<U>;
+  private createClosedMessage?: (event: CloseEvent) => D;
 
   private receiverQueue: [ResolveFn<D>, RejectFn<Error>][] = [];
   private done: boolean = false;
@@ -60,6 +63,7 @@ export class WebSocketClient<U, D> implements AsyncIterable<D> {
   ) {
     this.validate = handler.validate;
     this.serialize = handler.serialize;
+  this.createClosedMessage = handler.createClosedMessage;
     this.connectedPromise = new Promise(async (resolve, reject) => {
       this.socket = await getWebsocket(settings);
       this.socket.onopen = () => {
@@ -88,7 +92,17 @@ export class WebSocketClient<U, D> implements AsyncIterable<D> {
   private getClosedHandler(
     closeResolve: (_: void) => void,
   ): (_: CloseEvent) => void {
-    return (_: CloseEvent) => {
+    return (event: CloseEvent) => {
+      // If provided, enqueue a synthetic closed message for consumers
+      if (this.createClosedMessage) {
+        const closedMsg = this.createClosedMessage(event);
+        if (this.receiverQueue.length > 0) {
+          const [resolve, _] = this.receiverQueue.shift()!;
+          resolve({ value: closedMsg, done: false });
+        } else {
+          this.messageQueue.push(closedMsg);
+        }
+      }
       this.done = true;
       while (this.receiverQueue.length > 0) {
         const [resolve, reject] = this.receiverQueue.shift()!;
@@ -126,11 +140,11 @@ export class WebSocketClient<U, D> implements AsyncIterable<D> {
       next: (): Promise<IteratorResult<D>> => {
         if (this.error) {
           return Promise.reject(this.error);
-        } else if (this.done) {
-          return Promise.resolve({ value: undefined, done: true });
         } else if (this.messageQueue.length > 0) {
           const message = this.messageQueue.shift()!;
           return Promise.resolve({ value: message, done: false });
+        } else if (this.done) {
+          return Promise.resolve({ value: undefined, done: true });
         } else {
           return new Promise((resolve, reject) => {
             this.receiverQueue.push([resolve, reject]);

javascript/standalone/test/model-utils.spec.ts

@@ -246,6 +246,10 @@ describe("isServerMessageType", () => {
           },
         ],
       },
+      {
+        type: "connection.closed",
+        event_id: "event32",
+      },
     ];
 
     validMessages.forEach((message) => {