Feat/advanced conferencing features (#730)

* update drachtio-fsmrf and fixes to setCoachMode

* wip

* wip

* wip

* wip

* wip

* update gh actions

.github/workflows/build.yml

@@ -6,8 +6,8 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
-      - uses: actions/setup-node@v3
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
         with:
           node-version: 20
       - run: npm ci

lib/session/call-session.js

@@ -193,6 +193,24 @@ class CallSession extends Emitter {
     this._synthesizer = synth;
   }
 
+  /**
+   * ASR TTS fallback
+   */
+  get hasFallbackAsr() {
+    return this._hasFallbackAsr || false;
+  }
+
+  set hasFallbackAsr(i) {
+    this._hasFallbackAsr = i;
+  }
+
+  get hasFallbackTts() {
+    return this._hasFallbackTts || false;
+  }
+
+  set hasFallbackTts(i) {
+    this._hasFallbackTts = i;
+  }
   /**
    * default vendor to use for speech synthesis if not provided in the app
    */
@@ -702,7 +720,7 @@ class CallSession extends Emitter {
       task = await this.backgroundTaskManager.newTask('bargeIn', gather);
       task.sticky = autoEnable;
       // listen to the bargein-done from background manager
-      this.backgroundTaskManager.once('bargeIn-done', () => {
+      this.backgroundTaskManager.on('bargeIn-done', () => {
         if (this.requestor instanceof WsRequestor) {
           try {
             this.kill(true);
@@ -855,6 +873,19 @@ class CallSession extends Emitter {
             model_id: credential.model_id,
             options: credential.options
           };
+        } else if ('playht' === vendor) {
+          return {
+            api_key: credential.api_key,
+            user_id: credential.user_id,
+            voice_engine: credential.voice_engine,
+            options: credential.options
+          };
+        } else if ('rimelabs' === vendor) {
+          return {
+            api_key: credential.api_key,
+            model_id: credential.model_id,
+            options: credential.options
+          };
         } else if ('assemblyai' === vendor) {
           return {
             speech_credential_sid: credential.speech_credential_sid,
@@ -1585,7 +1616,23 @@ Duration=${duration} `
   }
 
   _preCacheAudio(newTasks) {
-    for (const task of newTasks) {
+    /**
+     * only precache audio for the a queued say if we have one or more non-Config verbs
+     * ahead of it in the queue.  This is because the Config verb returns immediately
+     * and would not give us enough time to generate the audio.  The point of precaching
+     * is to take advantage of getting the audio in advance of being needed, so we need
+     * to be confident we have some time before the say verb is executed, and the Config
+     * does not give us that confidence since it returns immediately.
+     */
+    const haveQueuedNonConfig = this.tasks.findIndex((t) => t.name !== TaskName.Config) !== -1;
+    let tasks = haveQueuedNonConfig ? newTasks : [];
+    if (!haveQueuedNonConfig) {
+      const idxFirstNotConfig = newTasks.findIndex((t) => t.name !== TaskName.Config);
+      if (-1 === idxFirstNotConfig) return;
+      tasks = newTasks.slice(idxFirstNotConfig + 1);
+    }
+
+    for (const task of tasks) {
       if (task.name === TaskName.Config && task.hasSynthesizer) {
         /* if they change synthesizer settings don't try to precache */
         break;
@@ -1951,6 +1998,7 @@ Duration=${duration} `
   /**
    * called when the caller has hung up.  Provided for subclasses to override
    * in order to apply logic at this point if needed.
+   * return true if success fallback, return false if not
    */
   _callerHungup() {
     assert(false, 'subclass responsibility to override this method');
@@ -2392,6 +2440,7 @@ Duration=${duration} `
 
   _startActionHookNoResponseTimer(options) {
     this._clearActionHookNoResponseTimer();
+    this._actionHookDelayResolved = false;
     if (options.noResponseTimeoutMs) {
       this.logger.debug(`CallSession:_startActionHookNoResponseTimer ${options.noResponseTimeoutMs}`);
       this._actionHookNoResponseTimer = setTimeout(() => {
@@ -2405,7 +2454,9 @@ Duration=${duration} `
         if (t.length) {
           t[0].on('playDone', (err) => {
             if (err) this.logger.error({err}, `Call-Session:exec Error delay action, play ${verb}`);
-            this._startActionHookNoResponseTimer(options);
+            if (!this._actionHookDelayResolved) {
+              this._startActionHookNoResponseTimer(options);
+            }
           });
         }
         this.tasks.push(...t);
@@ -2423,7 +2474,16 @@ Duration=${duration} `
 
   _clearActionHookNoResponseTimer() {
     if (this._actionHookNoResponseTimer) {
+      // Action Hook delay is solved.
+      this._actionHookDelayResolved = true;
       clearTimeout(this._actionHookNoResponseTimer);
+      // if delay action is enabled
+      // and bot has responded with list of new verbs
+      // Only kill current running play task.
+      //https://github.com/jambonz/jambonz-feature-server/issues/710
+      if (this.currentTask?.name === TaskName.Play) {
+        this.currentTask.kill(this);
+      }
     }
     this._actionHookNoResponseTimer = null;
   }

lib/tasks/conference.js

@@ -60,6 +60,8 @@ class Conference extends Task {
 
     this.emitter = new Emitter();
     this.results = {};
+    this.coaching = [];
+    this.speakOnlyTo = this.data.speakOnlyTo;
 
     // transferred from another server in order to bridge to a local caller?
     if (this.data._ && this.data._.connectTime) {
@@ -348,7 +350,7 @@ class Conference extends Task {
       Object.assign(opts, {flags: {
         ...(this.endConferenceOnExit && {endconf: true}),
         ...(this.startConferenceOnEnter && {moderator: true}),
-        ...((this.joinMuted || this.data.speakOnlyTo) && {joinMuted: true}),
+        ...((this.joinMuted || this.speakOnlyTo) && {joinMuted: true}),
       }});
 
       /**
@@ -361,7 +363,7 @@ class Conference extends Task {
     try {
       const {memberId, confUuid} = await this.ep.join(this.confName, opts);
       this.logger.debug({memberId, confUuid}, `Conference:_joinConference: successfully joined ${this.confName}`);
-      this.memberId = memberId;
+      this.memberId = parseInt(memberId, 10);
       this.confUuid = confUuid;
 
       // set a tag for this member, if provided
@@ -395,8 +397,8 @@ class Conference extends Task {
           .catch((err) => {});
       }
 
-      if (this.data.speakOnlyTo) {
-        this.setCoachMode(this.data.speakOnlyTo);
+      if (this.speakOnlyTo) {
+        this.setCoachMode(this.speakOnlyTo);
       }
     } catch (err) {
       this.logger.error(err, `Failed to join conference ${this.confName}`);
@@ -586,7 +588,7 @@ class Conference extends Task {
       const response = await this.ep.api('conference', [this.confName, 'get', 'count']);
       if (response.body && confNoMatch(response.body)) this.participantCount = 0;
       else if (response.body && /^\d+$/.test(response.body)) this.participantCount = parseInt(response.body) - 1;
-      this.logger.debug({response}, `Conference:_doFinalMemberCheck conference count ${this.participantCount}`);
+      this.logger.debug(`Conference:_doFinalMemberCheck conference count ${this.participantCount}`);
     } catch (err) {
       this.logger.info({err}, 'Conference:_doFinalMemberCheck error retrieving count (we were probably kicked');
     }
@@ -699,7 +701,12 @@ class Conference extends Task {
 
   // conference event handlers
   _onAddMember(logger, cs, evt) {
-    logger.debug({evt}, `Conference:_onAddMember - member added to conference ${this.confName}`);
+    const memberId = parseInt(evt.getHeader('Member-ID')) ;
+    if (this.speakOnlyTo) {
+      logger.debug(`Conference:_onAddMember - member ${memberId} added to ${this.confName}, updating coaching mode`);
+      this.setCoachMode(this.speakOnlyTo).catch(() => {});
+    }
+    else logger.debug(`Conference:_onAddMember - member ${memberId} added to conference ${this.confName}`);
   }
   _onDelMember(logger, cs, evt) {
     const memberId = parseInt(evt.getHeader('Member-ID')) ;
@@ -734,13 +741,46 @@ class Conference extends Task {
     }
   }
 
+  _onTag(logger, cs, evt) {
+    const memberId = parseInt(evt.getHeader('Member-ID')) ;
+    const tag = evt.getHeader('Tag') || '';
+    if (memberId !== this.memberId && this.speakOnlyTo) {
+      logger.info(`Conference:_onTag - member ${memberId} set tag to '${tag }'; updating coach mode accordingly`);
+      this.setCoachMode(this.speakOnlyTo).catch(() => {});
+    }
+  }
+
+  /**
+   * Set the conference to "coaching" mode, where the audio of the participant is only heard
+   * by a subset of the participants in the conference.
+   * We do this by first getting all of the members who do *not* have this tag, and then
+   * we configure this members audio to not be sent to them.
+   * @param {string} speakOnlyTo - tag of the members who should receive our audio
+   *
+   * N.B.: this feature requires jambonz patches to freeswitch mod_conference
+   */
   async setCoachMode(speakOnlyTo) {
+    this.speakOnlyTo = speakOnlyTo;
+    if (!this.memberId) {
+      this.logger.info('Conference:_setCoachMode: no member id yet');
+      return;
+    }
     try {
-      const response = await this.ep.api('conference', [this.confName, 'gettag', speakOnlyTo, 'nomatch']);
-      this.logger.info(`Conference:_setCoachMode: my audio will only be sent to particpants ${response}`);
-      await this.ep.api('conference', [this.confName, 'relate', this.memberId, response, 'nospeak']);
-      this.speakOnlyTo = speakOnlyTo;
-      this.coaching = response;
+      const members = (await this.ep.getNonMatchingConfParticipants(this.confName, speakOnlyTo))
+        .filter((m) => m !== this.memberId);
+      if (members.length === 0) {
+        this.logger.info({members}, 'Conference:_setCoachMode: all participants have the tag, so all will hear me');
+        if (this.coaching.length) {
+          await this.ep.api('conference', [this.confName, 'relate', this.memberId, this.coaching.join(','), 'clear']);
+          this.coaching = [];
+        }
+      }
+      else {
+        const memberList = members.join(',');
+        this.logger.info(`Conference:_setCoachMode: my audio will NOT be sent to ${memberList}`);
+        await this.ep.api('conference', [this.confName, 'relate', this.memberId, memberList, 'nospeak']);
+        this.coaching = members;
+      }
     } catch (err) {
       this.logger.error({err, speakOnlyTo}, '_setCoachMode: Error');
     }
@@ -748,14 +788,14 @@ class Conference extends Task {
 
   async clearCoachMode() {
     try {
-      if (!this.coaching) {
+      if (this.coaching.length === 0) {
         this.logger.info('Conference:_clearCoachMode: no coaching mode to clear');
         return;
       }
       this.logger.info(`Conference:_clearCoachMode: now sending my audio to all, including ${this.coaching}`);
       await this.ep.api('conference', [this.confName, 'relate', this.memberId, this.coaching, 'clear']);
       this.speakOnlyTo = null;
-      this.coaching = null;
+      this.coaching = [];
     } catch (err) {
       this.logger.error({err}, '_clearCoachMode: Error');
     }

lib/tasks/config.js

@@ -259,8 +259,7 @@ class TaskConfig extends Task {
         cs.stopBackgroundTask('transcribe');
       }
     }
-
-    if (this.actionHookDelayAction) {
+    if (Object.keys(this.actionHookDelayAction).length !== 0) {
       cs.actionHookDelayEnabled = this.actionHookDelayAction.enabled || false;
       cs.actionHookNoResponseTimeout = this.actionHookDelayAction.noResponseTimeout || 0;
       cs.actionHookNoResponseGiveUpTimeout = this.actionHookDelayAction.noResponseGiveUpTimeout || 0;

lib/tasks/gather.js

@@ -191,12 +191,7 @@ class TaskGather extends SttTask {
           this._startTranscribing(ep);
           return updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid);
         } catch (e) {
-          if (this.fallbackVendor && this.isHandledByPrimaryProvider) {
-            await this._fallback();
-            startListening(cs, ep);
-          } else {
-            this.logger.error({error: e}, 'error in initSpeech');
-          }
+          await this._startFallback(cs, ep, {error: e});
         }
       }
     };
@@ -204,12 +199,7 @@ class TaskGather extends SttTask {
     try {
       if (this.sayTask) {
         const {span, ctx} = this.startChildSpan(`nested:${this.sayTask.summary}`);
-        this.sayTask.span = span;
-        this.sayTask.ctx = ctx;
-        this.sayTask.exec(cs, {ep});  // kicked off, _not_ waiting for it to complete
-        this.sayTask.on('playDone', (err) => {
-          span.end();
-          if (err) this.logger.error({err}, 'Gather:exec Error playing tts');
+        const process = () => {
           this.logger.debug('Gather: nested say task completed');
           if (!this.killed) {
             startListening(cs, ep);
@@ -220,16 +210,22 @@ class TaskGather extends SttTask {
               });
             }
           }
+        };
+        this.sayTask.span = span;
+        this.sayTask.ctx = ctx;
+        this.sayTask.exec(cs, {ep})  // kicked off, _not_ waiting for it to complete
+          .catch((err) => {
+            process();
+          });
+        this.sayTask.on('playDone', (err) => {
+          span.end();
+          if (err) this.logger.error({err}, 'Gather:exec Error playing tts');
+          process();
         });
       }
       else if (this.playTask) {
         const {span, ctx} = this.startChildSpan(`nested:${this.playTask.summary}`);
-        this.playTask.span = span;
-        this.playTask.ctx = ctx;
-        this.playTask.exec(cs, {ep});  // kicked off, _not_ waiting for it to complete
-        this.playTask.on('playDone', (err) => {
-          span.end();
-          if (err) this.logger.error({err}, 'Gather:exec Error playing url');
+        const process = () => {
           this.logger.debug('Gather: nested play task completed');
           if (!this.killed) {
             startListening(cs, ep);
@@ -240,6 +236,17 @@ class TaskGather extends SttTask {
               });
             }
           }
+        };
+        this.playTask.span = span;
+        this.playTask.ctx = ctx;
+        this.playTask.exec(cs, {ep})  // kicked off, _not_ waiting for it to complete
+          .catch((err) => {
+            process();
+          });
+        this.playTask.on('playDone', (err) => {
+          span.end();
+          if (err) this.logger.error({err}, 'Gather:exec Error playing url');
+          process();
         });
       }
       else {
@@ -837,7 +844,7 @@ class TaskGather extends SttTask {
         if (this.bargein && (words + bufferedWords) < this.minBargeinWordCount) {
           this.logger.debug({evt, words, bufferedWords},
             'TaskGather:_onTranscription - final transcript but < min barge words');
-          this._bufferedTranscripts.push(evt);
+          if (!emptyTranscript) this._bufferedTranscripts.push(evt);
           if (!['soniox', 'aws', 'microsoft', 'deepgram'].includes(this.vendor)) this._startTranscribing(ep);
           return;
         }
@@ -932,9 +939,9 @@ class TaskGather extends SttTask {
   _onTranscriptionComplete(cs, ep) {
     this.logger.debug('TaskGather:_onTranscriptionComplete');
   }
-  async _onJambonzError(cs, ep, evt) {
-    this.logger.info({evt}, 'TaskGather:_onJambonzError');
-    if (this.isHandledByPrimaryProvider && this.fallbackVendor) {
+
+  async _startFallback(cs, ep, evt) {
+    if (this.canFallback) {
       ep.stopTranscription({
         vendor: this.vendor,
         bugname: this.bugname
@@ -942,17 +949,35 @@ class TaskGather extends SttTask {
         .catch((err) => this.logger.error({err}, `Error stopping transcription for primary vendor ${this.vendor}`));
       const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, cs.srf);
       try {
-        await this._fallback();
-        await this._initSpeech(cs, ep);
+        this.logger.debug('gather:_startFallback');
+        this.notifyError({ msg: 'ASR error',
+          details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'in progress'});
+        await this._initFallback();
+        this._speechHandlersSet = false;
+        await this._setSpeechHandlers(cs, ep);
         this._startTranscribing(ep);
         updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid);
-        return;
+        return true;
       } catch (error) {
         this.logger.info({error}, `There is error while falling back to ${this.fallbackVendor}`);
+        this.notifyError({ msg: 'ASR error',
+          details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'not available'});
       }
+    } else {
+      this.logger.debug('gather:_startFallback no condition for falling back');
+      this.notifyError({ msg: 'ASR error',
+        details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'not available'});
     }
-    const {writeAlerts, AlertType} = cs.srf.locals;
+    return false;
+  }
 
+  async _onJambonzError(cs, ep, evt) {
+    if (this.vendor === 'google' && evt.error_code === 0) {
+      this.logger.info({evt}, 'TaskTranscribe:_onJambonzError - ignoring google error code 0');
+      return;
+    }
+    this.logger.info({evt}, 'TaskGather:_onJambonzError');
+    const {writeAlerts, AlertType} = cs.srf.locals;
     if (this.vendor === 'nuance') {
       const {code, error} = evt;
       if (code === 404 && error === 'No speech') return this._resolve('timeout');
@@ -965,17 +990,23 @@ class TaskGather extends SttTask {
       message: `Custom speech vendor ${this.vendor} error: ${evt.error}`,
       vendor: this.vendor,
     }).catch((err) => this.logger.info({err}, 'Error generating alert for jambonz custom connection failure'));
-    this.notifyError({msg: 'ASR error', details:`Custom speech vendor ${this.vendor} error: ${evt.error}`});
+    if (!(await this._startFallback(cs, ep, evt))) {
+      this.notifyTaskDone();
+    }
   }
 
-  _onVendorConnectFailure(cs, _ep, evt) {
+  async _onVendorConnectFailure(cs, _ep, evt) {
     super._onVendorConnectFailure(cs, _ep, evt);
-    this.notifyTaskDone();
+    if (!(await this._startFallback(cs, _ep, evt))) {
+      this.notifyTaskDone();
+    }
   }
 
-  _onVendorError(cs, _ep, evt) {
+  async _onVendorError(cs, _ep, evt) {
     super._onVendorError(cs, _ep, evt);
-    this._resolve('stt-error', evt);
+    if (!(await this._startFallback(cs, _ep, evt))) {
+      this._resolve('stt-error', evt);
+    }
   }
 
   _onVadDetected(cs, ep) {

lib/tasks/listen.js

@@ -8,6 +8,10 @@ const DTMF_SPAN_NAME = 'dtmf';
 class TaskListen extends Task {
   constructor(logger, opts, parentTask) {
     super(logger, opts);
+    /**
+     * @deprecated
+     * use bidirectionalAudio.enabled
+     */
     this.disableBidirectionalAudio = opts.disableBidirectionalAudio;
     this.preconditions = TaskPreconditions.Endpoint;
 
@@ -25,6 +29,15 @@ class TaskListen extends Task {
     this.results = {};
     this.playAudioQueue = [];
     this.isPlayingAudioFromQueue = false;
+    this.bidirectionalAudio = {
+      enabled: this.disableBidirectionalAudio === true ? false : true,
+      ...(this.data['bidirectionalAudio']),
+    };
+
+    // From drachtio-version 3.0.40, forkAudioStart will send empty bugname, metadata together with
+    // bidirectionalAudio params that cause old version of freeswitch missunderstand between bugname and
+    // bidirectionalAudio params
+    this._bugname = 'audio_fork';
 
     if (this.transcribe) this.transcribeTask = makeTask(logger, {'transcribe': opts.transcribe}, this);
   }
@@ -133,7 +146,8 @@ class TaskListen extends Task {
       mixType: this.mixType,
       sampling: this.sampleRate,
       ...(this._bugname && {bugname: this._bugname}),
-      metadata
+      metadata,
+      bidirectionalAudio: this.bidirectionalAudio || {}
     });
     this.recordStartTime = moment();
     if (this.maxLength) {
@@ -153,7 +167,7 @@ class TaskListen extends Task {
     }
 
     /* support bi-directional audio */
-    if (!this.disableBidirectionalAudio) {
+    if (this.bidirectionalAudio.enabled) {
       ep.addCustomEventListener(ListenEvents.PlayAudio, this._onPlayAudio.bind(this, ep));
     }
     ep.addCustomEventListener(ListenEvents.KillAudio, this._onKillAudio.bind(this, ep));

lib/tasks/say.js

@@ -52,6 +52,151 @@ class TaskSay extends TtsTask {
     return `${this.name}{${this.text[0]}}`;
   }
 
+  _validateURL(urlString) {
+    try {
+      new URL(urlString);
+      return true;
+    } catch (e) {
+      return false;
+    }
+  }
+
+  async _synthesizeWithSpecificVendor(cs, ep, {vendor, language, voice, label, preCache = false}) {
+    const {srf, accountSid:account_sid} = cs;
+    const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, srf);
+    const {writeAlerts, AlertType, stats} = srf.locals;
+    const {synthAudio} = srf.locals.dbHelpers;
+    const engine = this.synthesizer.engine || 'standard';
+    const salt = cs.callSid;
+
+    let credentials = cs.getSpeechCredentials(vendor, 'tts', label);
+    /* parse Nuance voices into name and model */
+    let model;
+    if (vendor === 'nuance' && voice) {
+      const arr = /([A-Za-z-]*)\s+-\s+(enhanced|standard)/.exec(voice);
+      if (arr) {
+        voice = arr[1];
+        model = arr[2];
+      }
+    } else if (vendor === 'deepgram') {
+      model = voice;
+    }
+
+    /* allow for microsoft custom region voice and api_key to be specified as an override */
+    if (vendor === 'microsoft' && this.options.deploymentId) {
+      credentials = credentials || {};
+      credentials.use_custom_tts = true;
+      credentials.custom_tts_endpoint = this.options.deploymentId;
+      credentials.api_key = this.options.apiKey || credentials.apiKey;
+      credentials.region = this.options.region || credentials.region;
+      voice = this.options.voice || voice;
+    } else if (vendor === 'elevenlabs') {
+      credentials = credentials || {};
+      credentials.model_id = this.options.model_id || credentials.model_id;
+      credentials.voice_settings = this.options.voice_settings || {};
+      credentials.optimize_streaming_latency = this.options.optimize_streaming_latency
+      || credentials.optimize_streaming_latency;
+      voice = this.options.voice_id || voice;
+    }
+
+    ep.set({
+      tts_engine: vendor,
+      tts_voice: voice,
+      cache_speech_handles: 1,
+    }).catch((err) => this.logger.info({err}, 'Error setting tts_engine on endpoint'));
+
+    if (!preCache) this.logger.info({vendor, language, voice, model}, 'TaskSay:exec');
+    try {
+      if (!credentials) {
+        writeAlerts({
+          account_sid,
+          alert_type: AlertType.TTS_NOT_PROVISIONED,
+          vendor
+        }).catch((err) => this.logger.info({err}, 'Error generating alert for no tts'));
+        throw new Error('no provisioned speech credentials for TTS');
+      }
+      // synthesize all of the text elements
+      let lastUpdated = false;
+
+      /* produce an audio segment from the provided text */
+      const generateAudio = async(text) => {
+        if (this.killed) return;
+        if (text.startsWith('silence_stream://')) return text;
+
+        /* otel: trace time for tts */
+        if (!preCache)  {
+          const {span} = this.startChildSpan('tts-generation', {
+            'tts.vendor': vendor,
+            'tts.language': language,
+            'tts.voice': voice
+          });
+          this.otelSpan = span;
+        }
+        try {
+          const {filePath, servedFromCache, rtt} = await synthAudio(stats, {
+            account_sid,
+            text,
+            vendor,
+            language,
+            voice,
+            engine,
+            model,
+            salt,
+            credentials,
+            options: this.options,
+            disableTtsCache : this.disableTtsCache,
+            preCache
+          });
+          if (!filePath.startsWith('say:')) {
+            this.logger.debug(`file ${filePath}, served from cache ${servedFromCache}`);
+            if (filePath) cs.trackTmpFile(filePath);
+            if (this.otelSpan) {
+              this.otelSpan.setAttributes({'tts.cached': servedFromCache});
+              this.otelSpan.end();
+              this.otelSpan = null;
+            }
+            if (!servedFromCache && !lastUpdated) {
+              lastUpdated = true;
+              updateSpeechCredentialLastUsed(credentials.speech_credential_sid).catch(() => {/* logged error */});
+            }
+            if (!servedFromCache && rtt && !preCache) {
+              this.notifyStatus({
+                event: 'synthesized-audio',
+                vendor,
+                language,
+                characters: text.length,
+                elapsedTime: rtt
+              });
+            }
+          }
+          else {
+            this.logger.debug('a streaming tts api will be used');
+            const modifiedPath = filePath.replace('say:{', `say:{session-uuid=${ep.uuid},`);
+            return modifiedPath;
+          }
+          return filePath;
+        } catch (err) {
+          this.logger.info({err}, 'Error synthesizing tts');
+          if (this.otelSpan) this.otelSpan.end();
+          writeAlerts({
+            account_sid: cs.accountSid,
+            alert_type: AlertType.TTS_FAILURE,
+            vendor,
+            detail: err.message
+          }).catch((err) => this.logger.info({err}, 'Error generating alert for tts failure'));
+          throw err;
+        }
+      };
+
+      const arr = this.text.map((t) => (this._validateURL(t) ? t : generateAudio(t)));
+      return (await Promise.all(arr)).filter((fp) => fp && fp.length);
+    } catch (err) {
+      this.logger.info(err, 'TaskSay:exec error');
+      throw err;
+    }
+
+  }
+
   async exec(cs, {ep}) {
     const {srf, accountSid:account_sid} = cs;
     const {writeAlerts, AlertType} = srf.locals;
@@ -61,16 +206,16 @@ class TaskSay extends TtsTask {
     await super.exec(cs);
     this.ep = ep;
 
-    const vendor = this.synthesizer.vendor && this.synthesizer.vendor !== 'default' ?
+    let vendor = this.synthesizer.vendor && this.synthesizer.vendor !== 'default' ?
       this.synthesizer.vendor :
       cs.speechSynthesisVendor;
-    const language = this.synthesizer.language && this.synthesizer.language !== 'default' ?
+    let language = this.synthesizer.language && this.synthesizer.language !== 'default' ?
       this.synthesizer.language :
       cs.speechSynthesisLanguage ;
-    const voice =  this.synthesizer.voice && this.synthesizer.voice !== 'default' ?
+    let voice =  this.synthesizer.voice && this.synthesizer.voice !== 'default' ?
       this.synthesizer.voice :
       cs.speechSynthesisVoice;
-    const label = this.synthesizer.label && this.synthesizer.label !== 'default' ?
+    let label = this.synthesizer.label && this.synthesizer.label !== 'default' ?
       this.synthesizer.label :
       cs.speechSynthesisLabel;
 
@@ -87,12 +232,22 @@ class TaskSay extends TtsTask {
       this.synthesizer.fallbackLabel :
       cs.fallbackSpeechSynthesisLabel;
 
+    if (cs.hasFallbackTts) {
+      vendor = fallbackVendor;
+      language = fallbackLanguage;
+      voice = fallbackVoice;
+      label = fallbackLabel;
+    }
+
     let filepath;
     try {
       filepath = await this._synthesizeWithSpecificVendor(cs, ep, {vendor, language, voice, label});
     } catch (error) {
-      if (fallbackVendor && this.isHandledByPrimaryProvider) {
+      if (fallbackVendor && this.isHandledByPrimaryProvider && !cs.hasFallbackTts) {
+        this.notifyError(
+          { msg: 'TTS error', details:`TTS vendor ${vendor} error: ${error}`, failover: 'in progress'});
         this.isHandledByPrimaryProvider = false;
+        cs.hasFallbackTts = true;
         this.logger.info(`Synthesize error, fallback to ${fallbackVendor}`);
         filepath = await this._synthesizeWithSpecificVendor(cs, ep,
           {
@@ -102,6 +257,8 @@ class TaskSay extends TtsTask {
             label: fallbackLabel
           });
       } else {
+        this.notifyError(
+          { msg: 'TTS error', details:`TTS vendor ${vendor} error: ${error}`, failover: 'not available'});
         throw error;
       }
     }
@@ -187,6 +344,9 @@ class TaskSay extends TtsTask {
       if (key.startsWith('variable_tts_')) {
         let newKey = key.substring('variable_tts_'.length)
           .replace('whisper_', 'whisper.')
+          .replace('deepgram_', 'deepgram.')
+          .replace('playht_', 'playht.')
+          .replace('rimelabs_', 'rimelabs.')
           .replace('elevenlabs_', 'elevenlabs.');
         if (spanMapping[newKey]) newKey = spanMapping[newKey];
         attrs[newKey] = value;
@@ -198,6 +358,9 @@ class TaskSay extends TtsTask {
 }
 
 const spanMapping = {
+  // IMPORTANT!!! JAMBONZ WEBAPP WILL SHOW TEXT PERFECTLY IF THE SPAN NAME IS SMALLER OR EQUAL 25 CHARACTERS.
+  // EX: whisper.ratelim_reqs has length 20 <= 25 which is perfect
+  // Elevenlabs
   'elevenlabs.reported_latency_ms': 'elevenlabs.latency_ms',
   'elevenlabs.request_id': 'elevenlabs.req_id',
   'elevenlabs.history_item_id': 'elevenlabs.item_id',
@@ -205,11 +368,33 @@ const spanMapping = {
   'elevenlabs.name_lookup_time_ms': 'name_lookup_ms',
   'elevenlabs.connect_time_ms': 'connect_ms',
   'elevenlabs.final_response_time_ms': 'final_response_ms',
+  // Whisper
   'whisper.reported_latency_ms': 'whisper.latency_ms',
   'whisper.request_id': 'whisper.req_id',
+  'whisper.reported_organization': 'whisper.organization',
+  'whisper.reported_ratelimit_requests': 'whisper.ratelimit',
+  'whisper.reported_ratelimit_remaining_requests': 'whisper.ratelimit_remain',
+  'whisper.reported_ratelimit_reset_requests': 'whisper.ratelimit_reset',
   'whisper.name_lookup_time_ms': 'name_lookup_ms',
   'whisper.connect_time_ms': 'connect_ms',
   'whisper.final_response_time_ms': 'final_response_ms',
+  // Deepgram
+  'deepgram.request_id': 'deepgram.req_id',
+  'deepgram.reported_model_name': 'deepgram.model_name',
+  'deepgram.reported_model_uuid': 'deepgram.model_uuid',
+  'deepgram.reported_char_count': 'deepgram.char_count',
+  'deepgram.name_lookup_time_ms': 'name_lookup_ms',
+  'deepgram.connect_time_ms': 'connect_ms',
+  'deepgram.final_response_time_ms': 'final_response_ms',
+  // Playht
+  'playht.request_id': 'playht.req_id',
+  'playht.name_lookup_time_ms': 'name_lookup_ms',
+  'playht.connect_time_ms': 'connect_ms',
+  'playht.final_response_time_ms': 'final_response_ms',
+  // Rimelabs
+  'rimelabs.name_lookup_time_ms': 'name_lookup_ms',
+  'rimelabs.connect_time_ms': 'connect_ms',
+  'rimelabs.final_response_time_ms': 'final_response_ms',
 };
 
 module.exports = TaskSay;

lib/tasks/stt-task.js

@@ -98,6 +98,13 @@ class SttTask extends Task {
       this.fallbackLabel = cs.fallbackSpeechRecognizerLabel;
       if (this.data.recognizer) this.data.recognizer.fallbackLabel = this.fallbackLabel;
     }
+    // If call is already fallback to 2nd ASR vendor
+    // use that.
+    if (cs.hasFallbackAsr) {
+      this.vendor = this.fallbackVendor;
+      this.language = this.fallbackLanguage;
+      this.label = this.fallbackLabel;
+    }
     if (!this.data.recognizer.vendor) {
       this.data.recognizer.vendor = this.vendor;
     }
@@ -115,9 +122,11 @@ class SttTask extends Task {
       try {
         this.sttCredentials = await this._initSpeechCredentials(this.cs, this.vendor, this.label);
       } catch (error) {
-        if (this.fallbackVendor && this.isHandledByPrimaryProvider) {
-          await this._fallback();
+        if (this.canFallback) {
+          await this._initFallback();
+          this.notifyError({ msg: 'ASR error', details:`Invalid vendor ${this.vendor}`, failover: 'in progress'});
         } else {
+          this.notifyError({ msg: 'ASR error', details:`Invalid vendor ${this.vendor}`, failover: 'not available'});
           throw error;
         }
       }
@@ -186,9 +195,14 @@ class SttTask extends Task {
     return credentials;
   }
 
-  async _fallback() {
+  get canFallback() {
+    return this.fallbackVendor && this.isHandledByPrimaryProvider && !this.cs.hasFallbackAsr;
+  }
+
+  async _initFallback() {
     assert(this.fallbackVendor, 'fallback failed without fallbackVendor configuration');
     this.isHandledByPrimaryProvider = false;
+    this.cs.hasFallbackAsr = true;
     this.logger.info(`Failed to use primary STT provider, fallback to ${this.fallbackVendor}`);
     this.vendor = this.fallbackVendor;
     this.language = this.fallbackLanguage;
@@ -197,6 +211,8 @@ class SttTask extends Task {
     this.data.recognizer.language = this.language;
     this.data.recognizer.label = this.label;
     this.sttCredentials = await this._initSpeechCredentials(this.cs, this.vendor, this.label);
+    // cleanup previous listener from previous vendor
+    this.removeCustomEventListeners();
   }
 
   async compileHintsForCobalt(ep, hostport, model, token, hints) {
@@ -259,7 +275,6 @@ class SttTask extends Task {
       detail: evt.error,
       vendor: this.vendor,
     }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
-    this.notifyError({msg: 'ASR error', details:`Failed connecting to speech vendor ${this.vendor}: ${evt.error}`});
   }
 
   _onVendorConnectFailure(cs, _ep, evt) {
@@ -272,7 +287,6 @@ class SttTask extends Task {
       message: `Failed connecting to ${this.vendor} speech recognizer: ${reason}`,
       vendor: this.vendor,
     }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
-    this.notifyError({msg: 'ASR error', details:`Failed connecting to speech vendor ${this.vendor}: ${reason}`});
   }
 }
 

lib/tasks/transcribe.js

@@ -104,12 +104,15 @@ class TaskTranscribe extends SttTask {
 
       updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid)
         .catch(() => {/*already logged error */});
-
-      await this.awaitTaskDone();
     } catch (err) {
-      this.logger.info(err, 'TaskTranscribe:exec - error');
-      this.parentTask && this.parentTask.emit('error', err);
+      if (!(await this._startFallback(cs, ep, {error: err}))) {
+        this.logger.info(err, 'TaskTranscribe:exec - error');
+        this.parentTask && this.parentTask.emit('error', err);
+        this.removeCustomEventListeners();
+        return;
+      }
     }
+    await this.awaitTaskDone();
     this.removeCustomEventListeners();
   }
 
@@ -123,7 +126,7 @@ class TaskTranscribe extends SttTask {
       })
         .catch((err) => this.logger.info(err, 'Error TaskTranscribe:kill'));
     }
-    if (this.transcribing2 && this.ep2.connected) {
+    if (this.transcribing2 && this.ep2?.connected) {
       stopTranscription = true;
       this.ep2.stopTranscription({vendor: this.vendor, bugname: this.bugname})
         .catch((err) => this.logger.info(err, 'Error TaskTranscribe:kill'));
@@ -534,10 +537,8 @@ class TaskTranscribe extends SttTask {
     }
   }
 
-  async _onJambonzError(cs, _ep, evt) {
-    this.logger.info({evt}, 'TaskTranscribe:_onJambonzError');
-    if (this.paused) return;
-    if (this.isHandledByPrimaryProvider && this.fallbackVendor) {
+  async _startFallback(cs, _ep, evt) {
+    if (this.canFallback) {
       _ep.stopTranscription({
         vendor: this.vendor,
         bugname: this.bugname
@@ -545,38 +546,57 @@ class TaskTranscribe extends SttTask {
         .catch((err) => this.logger.error({err}, `Error stopping transcription for primary vendor ${this.vendor}`));
       const {updateSpeechCredentialLastUsed} = require('../utils/db-utils')(this.logger, cs.srf);
       try {
-        await this._fallback();
+        this.notifyError({ msg: 'ASR error',
+          details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'in progress'});
+        await this._initFallback();
         let channel = 1;
         if (this.ep !== _ep) {
           channel = 2;
         }
+        this[`_speechHandlersSet_${channel}`] = false;
         this._startTranscribing(cs, _ep, channel);
         updateSpeechCredentialLastUsed(this.sttCredentials.speech_credential_sid);
-        return;
+        return true;
       } catch (error) {
+        this.notifyError({ msg: 'ASR error',
+          details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'not available'});
         this.logger.info({error}, `There is error while falling back to ${this.fallbackVendor}`);
       }
     } else {
-      const {writeAlerts, AlertType} = cs.srf.locals;
+      this.logger.debug('transcribe:_startFallback no condition for falling back');
+      this.notifyError({ msg: 'ASR error',
+        details:`STT Vendor ${this.vendor} error: ${evt.error || evt.reason}`, failover: 'not available'});
+    }
+    return false;
+  }
 
-      if (this.vendor === 'nuance') {
-        const {code, error} = evt;
-        //TODO: fix below, currently _resolve does not send timeout events
-        if (code === 404 && error === 'No speech') return this._resolve('timeout');
-        if (code === 413 && error === 'Too much speech') return this._resolve('timeout');
-      }
-      this.logger.info({evt}, 'TaskTranscribe:_onJambonzError');
-      writeAlerts({
-        account_sid: cs.accountSid,
-        alert_type: AlertType.STT_FAILURE,
-        message: `Custom speech vendor ${this.vendor} error: ${evt.error}`,
-        vendor: this.vendor,
-      }).catch((err) => this.logger.info({err}, 'Error generating alert for jambonz custom connection failure'));
-      this.notifyError({msg: 'ASR error', details:`Custom speech vendor ${this.vendor} error: ${evt.error}`});
+  async _onJambonzError(cs, _ep, evt) {
+    if (this.vendor === 'google' && evt.error_code === 0) {
+      this.logger.info({evt}, 'TaskTranscribe:_onJambonzError - ignoring google error code 0');
+      return;
+    }
+    this.logger.info({evt}, 'TaskTranscribe:_onJambonzError');
+    if (this.paused) return;
+    const {writeAlerts, AlertType} = cs.srf.locals;
+
+    if (this.vendor === 'nuance') {
+      const {code, error} = evt;
+      if (code === 404 && error === 'No speech') return this._resolve('timeout');
+      if (code === 413 && error === 'Too much speech') return this._resolve('timeout');
+    }
+    this.logger.info({evt}, 'TaskTranscribe:_onJambonzError');
+    writeAlerts({
+      account_sid: cs.accountSid,
+      alert_type: AlertType.STT_FAILURE,
+      message: `Custom speech vendor ${this.vendor} error: ${evt.error}`,
+      vendor: this.vendor,
+    }).catch((err) => this.logger.info({err}, 'Error generating alert for jambonz custom connection failure'));
+    if (!(await this._startFallback(cs, _ep, evt))) {
+      this.notifyTaskDone();
     }
   }
 
-  _onVendorConnectFailure(cs, _ep, channel, evt) {
+  async _onVendorConnectFailure(cs, _ep, channel, evt) {
     super._onVendorConnectFailure(cs, _ep, evt);
     if (this.childSpan[channel - 1] && this.childSpan[channel - 1].span) {
       this.childSpan[channel - 1].span.setAttributes({
@@ -585,7 +605,9 @@ class TaskTranscribe extends SttTask {
       });
       this.childSpan[channel - 1].span.end();
     }
-    this.notifyTaskDone();
+    if (!(await this._startFallback(cs, _ep, evt))) {
+      this.notifyTaskDone();
+    }
   }
 
   _startAsrTimer(channel) {

lib/utils/background-task-manager.js

@@ -26,7 +26,7 @@ class BackgroundTaskManager extends Emitter {
     return this.tasks.size;
   }
 
-  async newTask(type, opts) {
+  async newTask(type, opts, sticky = false) {
     this.logger.info({opts}, `initiating Background task ${type}`);
     if (this.tasks.has(type)) {
       this.logger.info(`Background task ${type} is running, skipped`);
@@ -52,6 +52,7 @@ class BackgroundTaskManager extends Emitter {
     if (task) {
       this.tasks.set(type, task);
     }
+    if (task && sticky) task.sticky = true;
     return task;
   }
 
@@ -116,7 +117,7 @@ class BackgroundTaskManager extends Emitter {
           this._taskCompleted('bargeIn', task);
           if (task.sticky && !this.cs.callGone && !this.cs._stopping) {
             this.logger.info('BackgroundTaskManager:_initBargeIn: restarting background bargeIn');
-            this.newTask('bargeIn', opts);
+            this.newTask('bargeIn', opts, true);
           }
           return;
         })

lib/utils/db-utils.js

@@ -94,6 +94,17 @@ const speechMapper = (cred) => {
       obj.api_key = o.api_key;
       obj.model_id = o.model_id;
       obj.options = o.options;
+    } else if ('playht' === obj.vendor) {
+      const o = JSON.parse(decrypt(credential));
+      obj.api_key = o.api_key;
+      obj.user_id = o.user_id;
+      obj.voice_engine = o.voice_engine;
+      obj.options = o.options;
+    } else if ('rimelabs' === obj.vendor) {
+      const o = JSON.parse(decrypt(credential));
+      obj.api_key = o.api_key;
+      obj.model_id = o.model_id;
+      obj.options = o.options;
     } else if ('assemblyai' === obj.vendor) {
       const o = JSON.parse(decrypt(credential));
       obj.api_key = o.api_key;

lib/utils/transcription-utils.js

@@ -690,7 +690,9 @@ module.exports = (logger) => {
         ...(deepgramOptions.keywords) &&
           {DEEPGRAM_SPEECH_KEYWORDS: deepgramOptions.keywords.join(',')},
         ...('endpointing' in deepgramOptions) &&
-          {DEEPGRAM_SPEECH_ENDPOINTING: deepgramOptions.endpointing === false ? 'false' : deepgramOptions.endpointing},
+          {DEEPGRAM_SPEECH_ENDPOINTING: deepgramOptions.endpointing === false ? 'false' : deepgramOptions.endpointing,
+            // default DEEPGRAM_SPEECH_UTTERANCE_END_MS is 1000, will be override by user settings later if there is.
+            DEEPGRAM_SPEECH_UTTERANCE_END_MS: 1000},
         ...(deepgramOptions.utteranceEndMs) &&
           {DEEPGRAM_SPEECH_UTTERANCE_END_MS: deepgramOptions.utteranceEndMs},
         ...(deepgramOptions.vadTurnoff) &&

package.json

@@ -1,6 +1,6 @@
 {
   "name": "jambonz-feature-server",
-  "version": "0.8.6",
+  "version": "0.9.0",
   "main": "app.js",
   "engines": {
     "node": ">= 18.x"
@@ -25,57 +25,57 @@
     "jslint:fix": "eslint app.js tracer.js lib --fix"
   },
   "dependencies": {
-    "@aws-sdk/client-auto-scaling": "^3.360.0",
-    "@aws-sdk/client-sns": "^3.360.0",
+    "@aws-sdk/client-auto-scaling": "^3.549.0",
+    "@aws-sdk/client-sns": "^3.549.0",
     "@jambonz/db-helpers": "^0.9.3",
     "@jambonz/http-health-check": "^0.0.1",
-    "@jambonz/mw-registrar": "^0.2.4",
-    "@jambonz/realtimedb-helpers": "^0.8.7",
-    "@jambonz/speech-utils": "^0.0.47",
+    "@jambonz/mw-registrar": "^0.2.7",
+    "@jambonz/realtimedb-helpers": "^0.8.8",
+    "@jambonz/speech-utils": "^0.0.51",
     "@jambonz/stats-collector": "^0.1.9",
     "@jambonz/time-series": "^0.2.8",
     "@jambonz/verb-specifications": "^0.0.69",
-    "@opentelemetry/api": "^1.4.0",
-    "@opentelemetry/exporter-jaeger": "^1.9.0",
-    "@opentelemetry/exporter-trace-otlp-http": "^0.35.0",
-    "@opentelemetry/exporter-zipkin": "^1.9.0",
-    "@opentelemetry/instrumentation": "^0.35.0",
-    "@opentelemetry/resources": "^1.9.0",
-    "@opentelemetry/sdk-trace-base": "^1.9.0",
-    "@opentelemetry/sdk-trace-node": "^1.9.0",
-    "@opentelemetry/semantic-conventions": "^1.9.0",
+    "@opentelemetry/api": "^1.8.0",
+    "@opentelemetry/exporter-jaeger": "^1.23.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.50.0",
+    "@opentelemetry/exporter-zipkin": "^1.23.0",
+    "@opentelemetry/instrumentation": "^0.50.0",
+    "@opentelemetry/resources": "^1.23.0",
+    "@opentelemetry/sdk-trace-base": "^1.23.0",
+    "@opentelemetry/sdk-trace-node": "^1.23.0",
+    "@opentelemetry/semantic-conventions": "^1.23.0",
     "bent": "^7.3.12",
     "debug": "^4.3.4",
     "deepcopy": "^2.1.0",
-    "drachtio-fsmrf": "^3.0.39",
+    "drachtio-fsmrf": "^3.0.41",
     "drachtio-srf": "^4.5.31",
-    "express": "^4.18.2",
+    "express": "^4.19.2",
     "express-validator": "^7.0.1",
-    "ip": "^1.1.9",
-    "moment": "^2.29.4",
-    "parse-url": "^8.1.0",
-    "pino": "^8.8.0",
+    "ip": "^2.0.1",
+    "moment": "^2.30.1",
+    "parse-url": "^9.2.0",
+    "pino": "^8.20.0",
     "polly-ssml-split": "^0.1.0",
     "proxyquire": "^2.1.3",
-    "sdp-transform": "^2.14.1",
+    "sdp-transform": "^2.14.2",
     "short-uuid": "^4.2.2",
-    "sinon": "^15.0.1",
+    "sinon": "^17.0.1",
     "to-snake-case": "^1.0.0",
-    "undici": "^5.28.3",
+    "undici": "^6.11.1",
     "uuid-random": "^1.3.2",
     "verify-aws-sns-signature": "^0.1.0",
-    "ws": "^8.9.0",
+    "ws": "^8.16.0",
     "xml2js": "^0.6.2"
   },
   "devDependencies": {
     "clear-module": "^4.1.2",
-    "eslint": "^7.32.0",
-    "eslint-plugin-promise": "^4.3.1",
+    "eslint": "7.32.0",
+    "eslint-plugin-promise": "^6.1.1",
     "nyc": "^15.1.0",
-    "tape": "^5.6.1"
+    "tape": "^5.7.5"
   },
   "optionalDependencies": {
-    "bufferutil": "^4.0.6",
-    "utf-8-validate": "^5.0.8"
+    "bufferutil": "^4.0.8",
+    "utf-8-validate": "^6.0.3"
   }
 }

test/docker-compose-testbed.yaml

@@ -57,7 +57,7 @@ services:
         condition: service_healthy
 
   freeswitch:
-    image: drachtio/drachtio-freeswitch-mrf:0.6.2
+    image: drachtio/drachtio-freeswitch-mrf:0.7.3
     restart: always
     command: freeswitch --rtp-range-start 20000 --rtp-range-end 20100
     environment: