Merge branch 'main' of https://github.com/jambonz/jambonz-feature-server into feat/aws_s2s

* fix bug with race condition in dial which could spike cpu due to unhandled exception

* update to 0.2.12 speech-utils with azure ssml fix

* minor

data/example-voicemail-greetings.json

@@ -163,5 +163,16 @@
     "wird sich bei Ihnen melden",
     "ich melde mich bei dir",
     "wir können nicht"
+  ],
+  "it-IT": [
+    "segreteria telefonica",
+    "risponde la segreteria telefonica",
+    "lascia un messaggio",
+    "puoi lasciare un messaggio dopo il segnale",
+    "dopo il segnale acustico",
+    "il numero chiamato non è raggiungibile",
+    "non è raggiungibile",
+    "lascia pure un messaggio",
+    "puoi lasciare un messaggio"
   ]
 }

lib/http-routes/schemas/create-call.js

@@ -116,8 +116,8 @@ const customSanitizeFunction = (value) => {
       /* trims characters at the beginning and at the end of a string */
       value = value.trim();
 
-      /* Verify strings including 'http' via new URL */
-      if (value.includes('http')) {
+      // Only attempt to parse if the whole string is a URL
+      if (/^https?:\/\/\S+$/.test(value)) {
         value = new URL(value).toString();
       }
     }

lib/middleware.js

@@ -332,7 +332,7 @@ module.exports = function(srf, logger) {
       }
 
       // Resolve application.speech_synthesis_voice if it's custom voice
-      if (app2.speech_synthesis_vendor === 'google' && app2.speech_synthesis_voice.startsWith('custom_')) {
+      if (app2.speech_synthesis_vendor === 'google' && app2.speech_synthesis_voice?.startsWith('custom_')) {
         const arr = /custom_(.*)/.exec(app2.speech_synthesis_voice);
         if (arr) {
           const google_custom_voice_sid = arr[1];

lib/session/call-session.js

@@ -220,6 +220,18 @@ class CallSession extends Emitter {
     this._synthesizer = synth;
   }
 
+  /**
+   * Say stream enabled
+   */
+
+  get autoStreamTts() {
+    return this._autoStreamTts || false;
+  }
+
+  set autoStreamTts(i) {
+    this._autoStreamTts = i;
+  }
+
   /**
    * ASR TTS fallback
    */
@@ -1799,6 +1811,10 @@ Duration=${duration} `
       .catch((err) => this.logger.debug({err}, 'CallSession:_notifyTaskStatus - Error sending'));
   }
 
+  async _internalTtsStreamingBufferTokens(tokens) {
+    return await this.ttsStreamingBuffer?.bufferTokens(tokens) || {status: 'failed', reason: 'no tts streaming buffer'};
+  }
+
   _lccTtsFlush(opts) {
     this.ttsStreamingBuffer?.flush(opts);
   }

lib/tasks/config.js

@@ -17,7 +17,8 @@ class TaskConfig extends Task {
       'actionHookDelayAction',
       'boostAudioSignal',
       'vad',
-      'ttsStream'
+      'ttsStream',
+      'autoStreamTts'
     ].forEach((k) => this[k] = this.data[k] || {});
 
     if ('notifyEvents' in this.data) {
@@ -117,6 +118,7 @@ class TaskConfig extends Task {
     if (this.hasTtsStream) {
       phrase.push(`${this.ttsStream.enable ? 'enable' : 'disable'} ttsStream`);
     }
+    if ('autoStreamTts' in this.data) phrase.push(`enable Say.stream value ${this.data.autoStreamTts ? 'on' : 'off'}`);
     return `${this.name}{${phrase.join(',')}}`;
   }
 
@@ -296,6 +298,11 @@ class TaskConfig extends Task {
       });
     }
 
+    if ('autoStreamTts' in this.data) {
+      this.logger.info(`Config: autoStreamTts set to ${this.data.autoStreamTts}`);
+      cs.autoStreamTts = this.data.autoStreamTts;
+    }
+
     if (this.hasFillerNoise) {
       const {enable, ...opts} = this.fillerNoise;
       this.logger.info({fillerNoise: this.fillerNoise}, 'Config: fillerNoise');
@@ -330,7 +337,9 @@ class TaskConfig extends Task {
       };
       this.logger.info({opts: this.gatherOpts}, 'Config: enabling ttsStream');
       cs.enableBackgroundTtsStream(this.sayOpts);
-    } else if (!this.ttsStream.enable) {
+    }
+    // only disable ttsStream if it specifically set to false
+    else if (this.ttsStream.enable === false) {
       this.logger.info('Config: disabling ttsStream');
       cs.disableTtsStream();
     }

lib/tasks/dial.js

@@ -271,7 +271,12 @@ class TaskDial extends Task {
     }
     this._removeDtmfDetection(cs.dlg);
     this._removeDtmfDetection(this.dlg);
-    await this._killOutdials();
+    try {
+      await this._killOutdials();
+    }
+    catch (err) {
+      this.logger.info({err}, 'Dial:kill - error killing outdials');
+    }
     if (this.sd) {
       const byeReasonHeader = this.killReason === KillReason.MediaTimeout ? 'Media Timeout' : undefined;
       this.sd.kill(byeReasonHeader);
@@ -281,13 +286,22 @@ class TaskDial extends Task {
     }
     if (this.callSid) sessionTracker.remove(this.callSid);
     if (this.listenTask) {
-      await this.listenTask.kill(cs);
-      this.listenTask.span.end();
+      try {
+        await this.listenTask.kill(cs);
+        this.listenTask?.span?.end();
+      }
+      catch (err) {
+        this.logger.error({err}, 'Dial:kill - error killing listen task');
+      }
       this.listenTask = null;
     }
     if (this.transcribeTask) {
-      await this.transcribeTask.kill(cs);
-      this.transcribeTask.span.end();
+      try {
+        await this.transcribeTask.kill(cs);
+        this.transcribeTask?.span?.end();
+      } catch (err) {
+        this.logger.error({err}, 'Dial:kill - error killing transcribe task');
+      }
       this.transcribeTask = null;
     }
     this.notifyTaskDone();

lib/tasks/llm/index.js

@@ -6,6 +6,7 @@ const TaskLlmUltravox_S2S = require('./llms/ultravox_s2s');
 const TaskLlmElevenlabs_S2S = require('./llms/elevenlabs_s2s');
 const TaskLlmGoogle_S2S = require('./llms/google_s2s');
 const LlmMcpService = require('../../utils/llm-mcp');
+const TaskLlmAws_S2S = require('./llms/aws_s2s');
 
 class TaskLlm extends Task {
   constructor(logger, opts) {
@@ -85,6 +86,10 @@ class TaskLlm extends Task {
         llm = new TaskLlmGoogle_S2S(this.logger, this.data, this);
         break;
 
+      case 'aws':
+        llm = new TaskLlmAws_S2S(this.logger, this.data, this);
+        break;
+
       default:
         throw new Error(`Unsupported vendor ${this.vendor} for LLM`);
     }

lib/tasks/llm/llms/aws_s2s.js

@@ -0,0 +1,437 @@
+const Task = require('../../task');
+const crypto = require('crypto');
+const TaskName = 'Llm_Aws_s2s';
+const {LlmEvents_Aws} = require('../../../utils/constants');
+const ClientEvent = 'client.event';
+const SessionDelete = 'session.delete';
+
+const aws_server_events = [
+  'error',
+  'session.created',
+  'session.updated',
+];
+
+const expandWildcards = (events) => {
+  const expandedEvents = [];
+
+  events.forEach((evt) => {
+    if (evt.endsWith('.*')) {
+      const prefix = evt.slice(0, -2); // Remove the wildcard ".*"
+      const matchingEvents = aws_server_events.filter((e) => e.startsWith(prefix));
+      expandedEvents.push(...matchingEvents);
+    } else {
+      expandedEvents.push(evt);
+    }
+  });
+
+  return expandedEvents;
+};
+
+class TaskLlmAws_S2S extends Task {
+  constructor(logger, opts, parentTask) {
+    super(logger, opts, parentTask);
+    this.parent = parentTask;
+
+    this.vendor = this.parent.vendor;
+    this.model = this.parent.model;
+    this.auth = this.parent.auth;
+    this.connectionOptions = this.parent.connectOptions;
+
+    const {access_key_id, secret_access_key, aws_region} = this.auth || {};
+    if (!access_key_id) throw new Error('auth.access_key_id is required for Aws S2S');
+    if (!secret_access_key) throw new Error('auth.secret_access_key is required for Aws S2S');
+    if (!aws_region) throw new Error('auth.aws_region is required for Aws S2S');
+
+    this.access_key_id = access_key_id;
+    this.secret_access_key = secret_access_key;
+    this.aws_region = aws_region;
+    this.actionHook = this.data.actionHook;
+    this.eventHook = this.data.eventHook;
+    this.toolHook = this.data.toolHook;
+    const {inferenceConfiguration, toolConfiguration, voiceId, systemPrompt} = this.data.llmOptions;
+
+    this.inferenceConfiguration = inferenceConfiguration;
+    this.toolConfiguration = toolConfiguration;
+    this.voiceId = voiceId;
+    this.systemPrompt = systemPrompt;
+    this.promptName = crypto.randomUUID();
+
+    this.results = {
+      completionReason: 'normal conversation end'
+    };
+
+    /**
+     * only one of these will have items,
+     * if includeEvents, then these are the events to include
+     * if excludeEvents, then these are the events to exclude
+     */
+    this.includeEvents = [];
+    this.excludeEvents = [];
+
+    /* default to all events if user did not specify */
+    this._populateEvents(this.data.events || aws_server_events);
+
+    this.addCustomEventListener = parentTask.addCustomEventListener.bind(parentTask);
+    this.removeCustomEventListeners = parentTask.removeCustomEventListeners.bind(parentTask);
+  }
+
+  get name() { return TaskName; }
+
+  get host() {
+    const {host} = this.connectionOptions || {};
+    return host || (this.vendor === 'aws' ? 'api.aws.com' : void 0);
+  }
+
+  get path() {
+    const {path} = this.connectionOptions || {};
+    if (path) return path;
+
+    switch (this.vendor) {
+      case 'aws':
+        return `v1/realtime?model=${this.model}`;
+      case 'microsoft':
+        return `aws/realtime?api-version=2024-10-01-preview&deployment=${this.model}`;
+    }
+  }
+
+  async _api(ep, args) {
+    const res = await ep.api('uuid_aws_s2s', `^^|${args.join('|')}`);
+    if (!res.body?.startsWith('+OK')) {
+      throw new Error({args}, `Error calling uuid_aws_s2s: ${res.body}`);
+    }
+  }
+
+  async exec(cs, {ep}) {
+    await super.exec(cs);
+
+    await this._startListening(cs, ep);
+
+    await this.awaitTaskDone();
+
+    /* note: the parent llm verb started the span, which is why this is necessary */
+    await this.parent.performAction(this.results);
+
+    this._unregisterHandlers();
+  }
+
+  async kill(cs) {
+    super.kill(cs);
+
+    this._api(cs.ep, [cs.ep.uuid, SessionDelete])
+      .catch((err) => this.logger.info({err}, 'TaskLlmAws_S2S:kill - error deleting session'));
+
+    this.notifyTaskDone();
+  }
+
+  /**
+   * Send function call output to the Aws server in the form of conversation.item.create
+   * per https://platform.aws.com/docs/guides/realtime/function-calls
+   */
+  async processToolOutput(ep, tool_call_id, data) {
+    try {
+      this.logger.debug({tool_call_id, data}, 'TaskLlmAws_S2S:processToolOutput');
+
+      if (!data.type || data.type !== 'conversation.item.create') {
+        this.logger.info({data},
+          'TaskLlmAws_S2S:processToolOutput - invalid tool output, must be conversation.item.create');
+      }
+      else {
+        await this._api(ep, [ep.uuid, ClientEvent, JSON.stringify(data)]);
+
+        // spec also recommends to send immediate response.create
+        await this._api(ep, [ep.uuid, ClientEvent, JSON.stringify({type: 'response.create'})]);
+      }
+    } catch (err) {
+      this.logger.info({err}, 'TaskLlmAws_S2S:processToolOutput');
+    }
+  }
+
+  /**
+   * Send a session.update to the Aws server
+   * Note: creating and deleting conversation items also supported as well as interrupting the assistant
+   */
+  async processLlmUpdate(ep, data, _callSid) {
+    try {
+      this.logger.debug({data, _callSid}, 'TaskLlmAws_S2S:processLlmUpdate');
+
+      if (!data.type || ![
+        'session.update',
+        'conversation.item.create',
+        'conversation.item.delete',
+        'response.cancel'
+      ].includes(data.type)) {
+        this.logger.info({data}, 'TaskLlmAws_S2S:processLlmUpdate - invalid mid-call request');
+      }
+      else {
+        await this._api(ep, [ep.uuid, ClientEvent, JSON.stringify(data)]);
+      }
+    } catch (err) {
+      this.logger.info({err}, 'TaskLlmAws_S2S:processLlmUpdate');
+    }
+  }
+
+  async _startListening(cs, ep) {
+    this._registerHandlers(ep);
+
+    try {
+      const args = [ep.uuid, 'session.create', this.aws_region, this.model, this.access_key_id, this.secret_access_key];
+      await this._api(ep, args);
+    } catch (err) {
+      this.logger.error({err}, 'TaskLlmAws_S2S:_startListening');
+      this.notifyTaskDone();
+    }
+  }
+
+  async _sendClientEvent(ep, obj) {
+    let ok = true;
+    this.logger.debug({obj}, 'TaskLlmAws_S2S:_sendClientEvent');
+    try {
+      const args = [ep.uuid, ClientEvent, JSON.stringify(obj)];
+      await this._api(ep, args);
+    } catch (err) {
+      ok = false;
+      this.logger.error({err}, 'TaskLlmAws_S2S:_sendClientEvent - Error');
+    }
+    return ok;
+  }
+
+  async _sendInitialMessage(ep) {
+    const settionStart = {
+      event: {
+        sessionStart: {
+          inferenceConfiguration: this.inferenceConfiguration || {
+            maxTokens: 1024,
+            temperature: 0.7,
+            topP: 0.9,
+          },
+        }
+      }
+    };
+
+    if (!await this._sendClientEvent(ep, settionStart)) {
+      return this.notifyTaskDone();
+    }
+
+    const promptStart = {
+      event: {
+        promptStart: {
+          promptName: this.promptName,
+          textOutputConfiguration: {
+            mediaType: 'text/plain',
+          },
+          audioOutputConfiguration: {
+            voiceId: this.voiceId || 'tiffany',
+          },
+          toolUseOutputConfiguration: {
+            mediaType: 'application/json',
+          },
+          ...(this.toolConfiguration && {
+            toolConfiguration: this.toolConfiguration
+          })
+        }
+      }
+    };
+    if (!await this._sendClientEvent(ep, promptStart)) {
+      return this.notifyTaskDone();
+    }
+
+    if (this.systemPrompt) {
+      const contentName = crypto.randomUUID();
+      const systemPromptContentStart = {
+        event: {
+          contentStart: {
+            promptName: this.promptName,
+            contentName,
+            type: 'TEXT',
+            interactive: true,
+            role: 'SYSTEM',
+            textInputConfiguration: {
+              mediaType: 'text/plain',
+            }
+          }
+        }
+      };
+      if (!await this._sendClientEvent(ep, systemPromptContentStart)) {
+        return this.notifyTaskDone();
+      }
+
+      const systemPromptContent = {
+        event: {
+          textInput: {
+            promptName: this.promptName,
+            contentName,
+            content: this.systemPrompt
+          }
+        }
+      };
+
+      if (!await this._sendClientEvent(ep, systemPromptContent)) {
+        return this.notifyTaskDone();
+      }
+
+      const systemPromptContentEnd = {
+        event: {
+          contentEnd: {
+            promptName: this.promptName,
+            contentName,
+          }
+        }
+      };
+      if (!await this._sendClientEvent(ep, systemPromptContentEnd)) {
+        return this.notifyTaskDone();
+      }
+    }
+
+    const audioContentStart = {
+      event: {
+        contentStart: {
+          promptName: this.promptName,
+          contentName: crypto.randomUUID(),
+          type: 'AUDIO',
+          interactive: true,
+          role: 'USER',
+        }
+      }
+    };
+    if (!await this._sendClientEvent(ep, audioContentStart)) {
+      return this.notifyTaskDone();
+    }
+
+  }
+
+  _registerHandlers(ep) {
+    this.addCustomEventListener(ep, LlmEvents_Aws.Connect, this._onConnect.bind(this, ep));
+    this.addCustomEventListener(ep, LlmEvents_Aws.ConnectFailure, this._onConnectFailure.bind(this, ep));
+    this.addCustomEventListener(ep, LlmEvents_Aws.Disconnect, this._onDisconnect.bind(this, ep));
+    this.addCustomEventListener(ep, LlmEvents_Aws.ServerEvent, this._onServerEvent.bind(this, ep));
+  }
+
+  _unregisterHandlers() {
+    this.removeCustomEventListeners();
+  }
+
+  _onError(ep, evt) {
+    this.logger.info({evt}, 'TaskLlmAws_S2S:_onError');
+    this.notifyTaskDone();
+  }
+
+  _onConnect(ep) {
+    this.logger.debug('TaskLlmAws_S2S:_onConnect');
+    this._sendInitialMessage(ep);
+  }
+  _onConnectFailure(_ep, evt) {
+    this.logger.info(evt, 'TaskLlmAws_S2S:_onConnectFailure');
+    this.results = {completionReason: 'connection failure'};
+    this.notifyTaskDone();
+  }
+  _onDisconnect(_ep, evt) {
+    this.logger.info(evt, 'TaskLlmAws_S2S:_onConnectFailure');
+    this.results = {completionReason: 'disconnect from remote end'};
+    this.notifyTaskDone();
+  }
+  async _onServerEvent(ep, evt) {
+    let endConversation = false;
+    const type = evt.type;
+    this.logger.info({evt}, 'TaskLlmAws_S2S:_onServerEvent');
+
+    /* check for failures, such as rate limit exceeded, that should terminate the conversation */
+    if (type === 'response.done' && evt.response.status === 'failed') {
+      endConversation = true;
+      this.results = {
+        completionReason: 'server failure',
+        error: evt.response.status_details?.error
+      };
+    }
+
+    /* server errors of some sort */
+    else if (type === 'error') {
+      endConversation = true;
+      this.results = {
+        completionReason: 'server error',
+        error: evt.error
+      };
+    }
+
+    /* tool calls */
+    else if (type === 'response.output_item.done' && evt.item?.type === 'function_call') {
+      this.logger.debug({evt}, 'TaskLlmAws_S2S:_onServerEvent - function_call');
+      const {name, call_id} = evt.item;
+      const args = JSON.parse(evt.item.arguments);
+
+      const mcpTools = this.parent.isMcpEnabled ? await this.parent.mcpService.getAvailableMcpTools() : [];
+      if (mcpTools.some((tool) => tool.name === name)) {
+        this.logger.debug({call_id, name, args}, 'TaskLlmAws_S2S:_onServerEvent - calling mcp tool');
+        try {
+          const res = await this.parent.mcpService.callMcpTool(name, args);
+          this.logger.debug({res}, 'TaskLlmAws_S2S:_onServerEvent - function_call - mcp result');
+          this.processToolOutput(ep, call_id, {
+            type: 'conversation.item.create',
+            item: {
+              type: 'function_call_output',
+              call_id,
+              output: res.content[0]?.text || 'There is no output from the function call',
+            }
+          });
+          return;
+        } catch (err) {
+          this.logger.info({err, evt}, 'TaskLlmAws_S2S - error calling function');
+          this.results = {
+            completionReason: 'client error calling mcp function',
+            error: err
+          };
+          endConversation = true;
+        }
+      }
+      else if (!this.toolHook) {
+        this.logger.warn({evt}, 'TaskLlmAws_S2S:_onServerEvent - no toolHook defined!');
+      }
+      else {
+        try {
+          await this.parent.sendToolHook(call_id, {name, args});
+        } catch (err) {
+          this.logger.info({err, evt}, 'TaskLlmAws - error calling function');
+          this.results = {
+            completionReason: 'client error calling function',
+            error: err
+          };
+          endConversation = true;
+        }
+      }
+    }
+
+    /* check whether we should notify on this event */
+    if (this.includeEvents.length > 0 ? this.includeEvents.includes(type) : !this.excludeEvents.includes(type)) {
+      this.parent.sendEventHook(evt)
+        .catch((err) => this.logger.info({err}, 'TaskLlmAws_S2S:_onServerEvent - error sending event hook'));
+    }
+
+    if (endConversation) {
+      this.logger.info({results: this.results}, 'TaskLlmAws_S2S:_onServerEvent - ending conversation due to error');
+      this.notifyTaskDone();
+    }
+  }
+
+  _populateEvents(events) {
+    if (events.includes('all')) {
+      /* work by excluding specific events */
+      const exclude = events
+        .filter((evt) => evt.startsWith('-'))
+        .map((evt) => evt.slice(1));
+      if (exclude.length === 0) this.includeEvents = aws_server_events;
+      else this.excludeEvents = expandWildcards(exclude);
+    }
+    else {
+      /* work by including specific events */
+      const include = events
+        .filter((evt) => !evt.startsWith('-'));
+      this.includeEvents = expandWildcards(include);
+    }
+
+    this.logger.debug({
+      includeEvents: this.includeEvents,
+      excludeEvents: this.excludeEvents
+    }, 'TaskLlmAws_S2S:_populateEvents');
+  }
+}
+
+module.exports = TaskLlmAws_S2S;

lib/tasks/play.js

@@ -48,7 +48,7 @@ class TaskPlay extends Task {
          */
         ep.once('playback-start', (evt) => {
           this.logger.debug({evt}, 'Play got playback-start');
-          this.cs.stickyEventEmitter.once('uuid_break', (t) => {
+          this.cs.stickyEventEmitter?.once('uuid_break', (t) => {
             if (t?.taskId === this.taskId) {
               this.logger.debug(`Play got kill-playback, executing uuid_break, taskId: ${t?.taskId}`);
               this.ep.api('uuid_break', this.ep.uuid).catch((err) => this.logger.info(err, 'Error killing audio'));

lib/tasks/say.js

@@ -3,24 +3,32 @@ const TtsTask = require('./tts-task');
 const {TaskName, TaskPreconditions} = require('../utils/constants');
 const pollySSMLSplit = require('polly-ssml-split');
 const { SpeechCredentialError } = require('../utils/error');
+const { sleepFor } = require('../utils/helpers');
 
-const breakLengthyTextIfNeeded = (logger,  text) => {
-  const chunkSize = 1000;
+const breakLengthyTextIfNeeded = (logger, text) => {
+  // As The text can be used for tts streaming, we need to break lengthy text into smaller chunks
+// HIGH_WATER_BUFFER_SIZE defined in tts-streaming-buffer.js
+  const chunkSize = 900;
   const isSSML = text.startsWith('<speak>');
-  if (text.length <= chunkSize || !isSSML) return [text];
   const options = {
-    // MIN length
     softLimit: 100,
-    // MAX length, exclude 15 characters <speak></speak>
     hardLimit: chunkSize - 15,
-    // Set of extra split characters (Optional property)
     extraSplitChars: ',;!?',
   };
   pollySSMLSplit.configure(options);
   try {
-    return pollySSMLSplit.split(text);
+    if (text.length <= chunkSize) return [text];
+    if (isSSML) {
+      return pollySSMLSplit.split(text);
+    } else {
+      // Wrap with <speak> and split
+      const wrapped = `<speak>${text}</speak>`;
+      const splitArr = pollySSMLSplit.split(wrapped);
+      // Remove <speak> and </speak> from each chunk
+      return splitArr.map((str) => str.replace(/^<speak>/, '').replace(/<\/speak>$/, ''));
+    }
   } catch (err) {
-    logger.info({err}, 'Error spliting SSML long text');
+    logger.info({err}, 'Error splitting SSML long text');
     return [text];
   }
 };
@@ -39,6 +47,9 @@ class TaskSay extends TtsTask {
     assert.ok((typeof this.data.text === 'string' || Array.isArray(this.data.text)) || this.data.stream === true,
       'Say: either text or stream:true is required');
 
+    this.text = this.data.text ? (Array.isArray(this.data.text) ? this.data.text : [this.data.text])
+      .map((t) => breakLengthyTextIfNeeded(this.logger, t))
+      .flat() : [];
 
     if (this.data.stream === true) {
       this._isStreamingTts = true;
@@ -46,10 +57,6 @@ class TaskSay extends TtsTask {
     }
     else {
       this._isStreamingTts = false;
-      this.text = (Array.isArray(this.data.text) ? this.data.text : [this.data.text])
-        .map((t) => breakLengthyTextIfNeeded(this.logger, t))
-        .flat();
-
       this.loop = this.data.loop || 1;
       this.isHandledByPrimaryProvider = true;
     }
@@ -85,6 +92,10 @@ class TaskSay extends TtsTask {
     }
 
     try {
+      this._isStreamingTts = this._isStreamingTts || cs.autoStreamTts;
+      if (this.isStreamingTts) {
+        this.closeOnStreamEmpty = this.closeOnStreamEmpty || this.text.length !== 0;
+      }
       if (this.isStreamingTts) await this.handlingStreaming(cs, obj);
       else await this.handling(cs, obj);
       this.emit('playDone');
@@ -116,6 +127,54 @@ class TaskSay extends TtsTask {
 
       cs.requestor?.request('tts:streaming-event', '/streaming-event', {event_type: 'stream_open'})
         .catch((err) => this.logger.info({err}, 'TaskSay:handlingStreaming - Error sending'));
+
+      if (this.text.length !== 0) {
+        this.logger.info('TaskSay:handlingStreaming - sending text to TTS stream');
+        for (const t of this.text) {
+          const result = await cs._internalTtsStreamingBufferTokens(t);
+          if (result?.status === 'failed') {
+            if (result.reason === 'full') {
+              // Retry logic for full buffer
+              const maxRetries = 5;
+              let backoffMs = 1000;
+              for (let retryCount = 0; retryCount < maxRetries && !this.killed; retryCount++) {
+                this.logger.info(
+                  `TaskSay:handlingStreaming - retry ${retryCount + 1}/${maxRetries} after ${backoffMs}ms`);
+                await sleepFor(backoffMs);
+
+                const retryResult = await cs._internalTtsStreamingBufferTokens(t);
+
+                // Exit retry loop on success
+                if (retryResult?.status !== 'failed') {
+                  break;
+                }
+
+                // Handle failure for reason other than full buffer
+                if (retryResult.reason !== 'full') {
+                  this.logger.info(
+                    {result: retryResult}, 'TaskSay:handlingStreaming - TTS stream failed to buffer tokens');
+                  throw new Error(`TTS stream failed to buffer tokens: ${retryResult.reason}`);
+                }
+
+                // Last retry attempt failed
+                if (retryCount === maxRetries - 1) {
+                  this.logger.info('TaskSay:handlingStreaming - Maximum retries exceeded for full buffer');
+                  throw new Error('TTS stream buffer full - maximum retries exceeded');
+                }
+
+                // Increase backoff for next retry
+                backoffMs = Math.min(backoffMs * 1.5, 10000);
+              }
+            } else {
+              // Immediate failure for non-full buffer issues
+              this.logger.info({result}, 'TaskSay:handlingStreaming - TTS stream failed to buffer tokens');
+              throw new Error(`TTS stream failed to buffer tokens: ${result.reason}`);
+            }
+          } else {
+            await cs._lccTtsFlush();
+          }
+        }
+      }
     } catch (err) {
       this.logger.info({err}, 'TaskSay:handlingStreaming - Error setting channel vars');
       cs.requestor?.request('tts:streaming-event', '/streaming-event', {event_type: 'stream_closed'})

lib/utils/constants.json

@@ -194,6 +194,13 @@
     "Disconnect": "openai_s2s::disconnect",
     "ServerEvent": "openai_s2s::server_event"
   },
+  "LlmEvents_Aws": {
+    "Error": "error",
+    "Connect": "aws_s2s::connect",
+    "ConnectFailure": "aws_s2s::connect_failed",
+    "Disconnect": "aws_s2s::disconnect",
+    "ServerEvent": "aws_s2s::server_event"
+  },
   "LlmEvents_Google": {
     "Error": "error",
     "Connect": "google_s2s::connect",

lib/utils/ws-requestor.js

@@ -132,8 +132,8 @@ class WsRequestor extends BaseRequestor {
 
         while (retryCount <= this.maxReconnects) {
           try {
-            this.logger.error({retryCount, maxReconnects: this.maxReconnects},
-              'WsRequestor:request - attempting connection');
+            this.logger.debug({retryCount, maxReconnects: this.maxReconnects},
+              'WsRequestor:request - attempting connection retry');
 
             // Ensure clean state before each connection attempt
             if (this.ws) {
@@ -141,38 +141,29 @@ class WsRequestor extends BaseRequestor {
               this.ws = null;
             }
 
-            this.logger.error({retryCount}, 'WsRequestor:request - calling _connect()');
             const startAt = process.hrtime();
             await this._connect();
             const rtt = this._roundTrip(startAt);
             this.stats.histogram('app.hook.connect_time', rtt, ['hook_type:app']);
-            this.logger.error({retryCount}, 'WsRequestor:request - connection successful, exiting retry loop');
             lastError = null;
             break;
           } catch (error) {
             lastError = error;
             retryCount++;
-            this.logger.error({error: error.message, retryCount, maxReconnects: this.maxReconnects},
-              'WsRequestor:request - connection attempt failed');
 
             if (retryCount <= this.maxReconnects &&
                 this.retryPolicyValues?.length &&
                 this._shouldRetry(error, this.retryPolicyValues)) {
 
-              this.logger.error(
-                {url, error, retryCount, maxRetries: this.maxReconnects},
-                `WsRequestor:request - connection failed, retrying (${retryCount}/${this.maxReconnects})`
-              );
-
               const delay = this.backoffMs;
               this.backoffMs = this.backoffMs < 2000 ? this.backoffMs * 2 : (this.backoffMs + 2000);
-              this.logger.error({delay}, 'WsRequestor:request - waiting before retry');
+              this.logger.debug({delay}, 'WsRequestor:request - waiting before retry');
               await new Promise((resolve) => setTimeout(resolve, delay));
-              this.logger.error('WsRequestor:request - retry delay complete, attempting retry');
               continue;
             }
-            this.logger.error({lastError: lastError.message, retryCount, maxReconnects: this.maxReconnects},
-              'WsRequestor:request - throwing last error');
+
+            this.logger.error({error: error.message, retryCount, maxReconnects: this.maxReconnects},
+              'WsRequestor:request - all connection attempts failed');
             throw lastError;
           }
         }
@@ -370,7 +361,7 @@ class WsRequestor extends BaseRequestor {
 
       this
         .once('ready', (ws) => {
-          this.logger.error({retryCount: 'unknown'}, 'WsRequestor:_connect - ready event fired, resolving Promise');
+          this.logger.debug('WsRequestor:_connect - ready event fired, resolving Promise');
           this.removeAllListeners('not-ready');
           if (this.connections > 1) this.request('session:reconnect', this.url);
           resolve();

package-lock.json

@@ -15,10 +15,10 @@
         "@jambonz/http-health-check": "^0.0.1",
         "@jambonz/mw-registrar": "^0.2.7",
         "@jambonz/realtimedb-helpers": "^0.8.13",
-        "@jambonz/speech-utils": "^0.2.11",
+        "@jambonz/speech-utils": "^0.2.12",
         "@jambonz/stats-collector": "^0.1.10",
         "@jambonz/time-series": "^0.2.13",
-        "@jambonz/verb-specifications": "^0.0.104",
+        "@jambonz/verb-specifications": "^0.0.105",
         "@modelcontextprotocol/sdk": "^1.9.0",
         "@opentelemetry/api": "^1.8.0",
         "@opentelemetry/exporter-jaeger": "^1.23.0",
@@ -1466,9 +1466,9 @@
       }
     },
     "node_modules/@jambonz/speech-utils": {
-      "version": "0.2.11",
-      "resolved": "https://registry.npmjs.org/@jambonz/speech-utils/-/speech-utils-0.2.11.tgz",
-      "integrity": "sha512-5V+OJUUnK1CpKKrB0PrAbGVDcwRGQYH/ZPHFMBayW67XaNRpiL3b9jDpMvq35yzx8MGY18JpzCPgVKHTxERlqQ==",
+      "version": "0.2.12",
+      "resolved": "https://registry.npmjs.org/@jambonz/speech-utils/-/speech-utils-0.2.12.tgz",
+      "integrity": "sha512-1xik/ZRUtPE2SOztxweGI+RTXUbiUXRShJ8G/l7VJJBkSWbfKKerYIRfHicAPumHicaUrbqSzZ6hr0eghv80KA==",
       "dependencies": {
         "23": "^0.0.0",
         "@aws-sdk/client-polly": "^3.496.0",
@@ -1504,9 +1504,9 @@
       }
     },
     "node_modules/@jambonz/verb-specifications": {
-      "version": "0.0.104",
-      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.104.tgz",
-      "integrity": "sha512-G1LjK6ISujdg0zALudtUvdaPXmvA4FU6x3s8S9MwUbWbFo2WERMUcNOgQAutDZwOMrLH9DnbPL8ZIdnTCKnlkA==",
+      "version": "0.0.105",
+      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.105.tgz",
+      "integrity": "sha512-MD6RMJyXMoHpR7Wl3xmYmU54P0eF/9LNywRNNsdkAmSf0EogFqSJft4xD/yGeRWlO5O6eAYZEJdaMQeLSxitcg==",
       "license": "MIT",
       "dependencies": {
         "debug": "^4.3.4",

package.json

@@ -31,10 +31,10 @@
     "@jambonz/http-health-check": "^0.0.1",
     "@jambonz/mw-registrar": "^0.2.7",
     "@jambonz/realtimedb-helpers": "^0.8.13",
-    "@jambonz/speech-utils": "^0.2.11",
+    "@jambonz/speech-utils": "^0.2.12",
     "@jambonz/stats-collector": "^0.1.10",
     "@jambonz/time-series": "^0.2.13",
-    "@jambonz/verb-specifications": "^0.0.104",
+    "@jambonz/verb-specifications": "^0.0.105",
     "@modelcontextprotocol/sdk": "^1.9.0",
     "@opentelemetry/api": "^1.8.0",
     "@opentelemetry/exporter-jaeger": "^1.23.0",