Support reusable rt session & phonic say

Author: toubatbrian

agents/src/llm/realtime.ts

@@ -49,6 +49,9 @@ export interface RealtimeCapabilities {
   autoToolReplyGeneration: boolean;
   audioOutput: boolean;
   manualFunctionCalls: boolean;
+  midSessionContextUpdate?: boolean;
+  midSessionInstructionsUpdate?: boolean;
+  midSessionToolsUpdate?: boolean;
 }
 
 export interface InputTranscriptionCompleted {
@@ -155,6 +158,13 @@ export abstract class RealtimeSession extends EventEmitter {
     return;
   }
 
+  say(
+    _text: string | ReadableStream<string>,
+    _options?: { allowInterruptions?: boolean },
+  ): Promise<GenerationCreatedEvent> {
+    throw new Error(`${this.constructor.name} does not implement say(). use a TTS model instead`);
+  }
+
   private async _mainTaskImpl(signal: AbortSignal): Promise<void> {
     const reader = this.deferredInputStream.stream.getReader();
     while (true) {
@@ -167,6 +177,14 @@ export abstract class RealtimeSession extends EventEmitter {
   }
 
   setInputAudioStream(audioStream: ReadableStream<AudioFrame>): void {
-    this.deferredInputStream.setSource(audioStream);
+    if (this.deferredInputStream.isSourceSet) {
+      // Reused sessions must detach the previous audio source before rebinding.
+      void this.deferredInputStream.detachSource();
+    }
+    try {
+      this.deferredInputStream.setSource(audioStream);
+    } catch (error) {
+      throw error;
+    }
   }
 }

agents/src/stream/deferred_stream.ts

@@ -41,6 +41,7 @@ export class DeferredReadableStream<T> {
   private transform: IdentityTransform<T>;
   private writer: WritableStreamDefaultWriter<T>;
   private sourceReader?: ReadableStreamDefaultReader<T>;
+  private detachRequested = false;
 
   constructor() {
     this.transform = new IdentityTransform<T>();
@@ -83,6 +84,11 @@ export class DeferredReadableStream<T> {
 
       sourceError = e;
     } finally {
+      if (this.detachRequested) {
+        this.detachRequested = false;
+        return;
+      }
+
       // any other error from source will be propagated to the consumer
       if (sourceError) {
         try {
@@ -124,6 +130,7 @@ export class DeferredReadableStream<T> {
     }
 
     const sourceReader = this.sourceReader!;
+    this.detachRequested = true;
     // Clear source first so future setSource() calls can reattach cleanly.
     this.sourceReader = undefined;