fixed gather should ignore transcription if task is killed/resolved. (#1465)

* fixed gather should ignore transcription if task is killed/resolved.

* wip

lib/config.js

@@ -119,7 +119,7 @@ const ENCRYPTION_SECRET = process.env.ENCRYPTION_SECRET;
 const HTTP_POOL = process.env.HTTP_POOL && parseInt(process.env.HTTP_POOL);
 const HTTP_POOLSIZE = parseInt(process.env.HTTP_POOLSIZE, 10) || 10;
 const HTTP_PIPELINING = parseInt(process.env.HTTP_PIPELINING, 10) || 1;
-const HTTP_TIMEOUT = 10000;
+const HTTP_TIMEOUT = parseInt(process.env.JAMBONES_HTTP_TIMEOUT, 10) || 10000;
 const HTTP_PROXY_IP = process.env.JAMBONES_HTTP_PROXY_IP;
 const HTTP_PROXY_PORT = process.env.JAMBONES_HTTP_PROXY_PORT;
 const HTTP_PROXY_PROTOCOL = process.env.JAMBONES_HTTP_PROXY_PROTOCOL || 'http';
@@ -139,6 +139,10 @@ const JAMBONES_USE_FREESWITCH_TIMER_FD = process.env.JAMBONES_USE_FREESWITCH_TIM
 const JAMBONES_DIAL_SBC_FOR_REGISTERED_USER = process.env.JAMBONES_DIAL_SBC_FOR_REGISTERED_USER || false;
 const JAMBONES_MEDIA_TIMEOUT_MS = process.env.JAMBONES_MEDIA_TIMEOUT_MS || 0;
 const JAMBONES_MEDIA_HOLD_TIMEOUT_MS = process.env.JAMBONES_MEDIA_HOLD_TIMEOUT_MS || 0;
+
+/* say / tts */
+const JAMBONES_SAY_CHUNK_SIZE = parseInt(process.env.JAMBONES_SAY_CHUNK_SIZE, 10) || 900;
+
 // jambonz
 const JAMBONES_TRANSCRIBE_EP_DESTROY_DELAY_MS =
   process.env.JAMBONES_TRANSCRIBE_EP_DESTROY_DELAY_MS;
@@ -231,5 +235,6 @@ module.exports = {
   JAMBONES_DIAL_SBC_FOR_REGISTERED_USER,
   JAMBONES_MEDIA_TIMEOUT_MS,
   JAMBONES_MEDIA_HOLD_TIMEOUT_MS,
+  JAMBONES_SAY_CHUNK_SIZE,
   JAMBONES_TRANSCRIBE_EP_DESTROY_DELAY_MS,
 };

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

@@ -291,7 +291,7 @@ router.post('/',
             }, {
               ...(account.enable_debug_log && {level: 'debug'})
             });
-            app.requestor.logger = app.notifier.logger = sipLogger;
+            app.requestor.logger = app.notifier.logger = restDial.logger = sipLogger;
             const callInfo = new CallInfo({
               direction: CallDirection.Outbound,
               req: inviteReq,

lib/session/call-session.js

@@ -927,7 +927,7 @@ class CallSession extends Emitter {
         this.logger.debug('CallSession:enableBackgroundTtsStream - ttsStream enabled');
       } else {
         this.logger.debug(
-          'CallSession:enableBackgroundTtsStream - ignoring request as call does not have required conditions');
+          'CallSession:enableBackgroundTtsStream - ignoring request; conditions not met (probably not using ws api)');
       }
     }  catch (err) {
       this.logger.info({err, say}, 'CallSession:enableBackgroundTtsStream - Error creating background tts stream task');
@@ -941,9 +941,11 @@ class CallSession extends Emitter {
     }
   }
   clearTtsStream() {
-    this.requestor?.request('tts:streaming-event', '/streaming-event', {event_type: 'user_interruption'})
-      .catch((err) => this.logger.info({err}, 'CallSession:clearTtsStream - Error sending user_interruption'));
-    this.ttsStreamingBuffer?.clear();
+    if (this.isTtsStreamEnabled) {
+      this.requestor?.request('tts:streaming-event', '/streaming-event', {event_type: 'user_interruption'})
+        .catch((err) => this.logger.info({err}, 'CallSession:clearTtsStream - Error sending user_interruption'));
+      this.ttsStreamingBuffer?.clear();
+    }
   }
 
   startTtsStream() {
@@ -951,7 +953,7 @@ class CallSession extends Emitter {
   }
 
   stopTtsStream() {
-    if (this.appIsUsingWebsockets) {
+    if (this.isTtsStreamEnabled) {
       this.requestor?.request('tts:streaming-event', '/streaming-event', {event_type: 'stream_closed'})
         .catch((err) => this.logger.info({err}, 'CallSession:clearTtsStream - Error sending user_interruption'));
       this.ttsStreamingBuffer?.stop();
@@ -1248,9 +1250,10 @@ class CallSession extends Emitter {
       }
       else {
         writeAlerts({
-          alert_type: AlertType.STT_NOT_PROVISIONED,
+          alert_type: type === 'tts' ? AlertType.TTS_NOT_PROVISIONED : AlertType.STT_NOT_PROVISIONED,
           account_sid: this.accountSid,
           vendor,
+          label,
           target_sid: this.callSid
         }).catch((err) => this.logger.error({err}, 'Error writing tts alert'));
       }

lib/tasks/dial.js

@@ -158,7 +158,7 @@ class TaskDial extends Task {
 
   get canReleaseMedia() {
     const keepAnchor = this.data.anchorMedia ||
-      this.weAreTranscoding ||
+      this.isTranscoding ||
       this.cs.isBackGroundListen ||
       this.cs.onHoldMusic ||
       ANCHOR_MEDIA_ALWAYS ||
@@ -576,7 +576,7 @@ class TaskDial extends Task {
       proxy: `sip:${sbcAddress}`,
       callingNumber: this.callerId || fromUri.user,
       ...(this.callerName && {callingName: this.callerName}),
-      opusFirst: isOpusFirst(this.cs.ep.remote.sdp),
+      opusFirst: isOpusFirst(this.cs.ep.local.sdp),
       isVideoCall: this.cs.ep.remote.sdp.includes('m=video')
     };
 
@@ -773,6 +773,15 @@ class TaskDial extends Task {
   }
 
   async _connectSingleDial(cs, sd) {
+    // start connect with dialed leg, this is the soonest we can identify transcoding
+    if (this.epOther && sd.ep) {
+      const codecA = getLeadingCodec(this.epOther.local.sdp);
+      const codecB = getLeadingCodec(sd.ep.remote.sdp);
+      this.isTranscoding = (codecA !== codecB);
+      if (this.isTranscoding) {
+        this.logger.info(`Dial:_connectSingleDial - transcoding from ${codecA} (A leg) to ${codecB} (B leg)`);
+      }
+    }
     if (!this.bridged && !this.canReleaseMedia) {
       this.logger.debug('Dial:_connectSingleDial bridging endpoints');
       if (this.epOther) {
@@ -930,13 +939,6 @@ class TaskDial extends Task {
         this.logger.info({err}, 'Dial:_selectSingleDial - Error boosting audio signal');
       }
     }
-    /* basic determination to see if call is being transcoded */
-    const codecA = getLeadingCodec(this.epOther.local.sdp);
-    const codecB = getLeadingCodec(this.ep.remote.sdp);
-    this.weAreTranscoding = (codecA !== codecB);
-    if (this.weAreTranscoding) {
-      this.logger.info(`Dial:_selectSingleDial - transcoding from ${codecA} (A leg) to ${codecB} (B leg)`);
-    }
     /* if we can release the media back to the SBC, do so now */
     if (this.canReleaseMedia || this.shouldExitMediaPathEntirely) {
       setTimeout(this._releaseMedia.bind(this, cs, sd, this.shouldExitMediaPathEntirely), 200);

lib/tasks/gather.js

@@ -881,7 +881,7 @@ class TaskGather extends SttTask {
         this._fillerNoiseOn = false;  // in a race, if we just started audio it may sneak through here
         this.ep.api('uuid_break', this.ep.uuid)
           .catch((err) => this.logger.info(err, 'Error killing audio'));
-        cs.clearTtsStream();
+        if (cs.isTtsStreamEnabled) cs.clearTtsStream();
       }
       return;
     }
@@ -1318,6 +1318,8 @@ class TaskGather extends SttTask {
     }
 
     this.resolved = true;
+    // gather is resolved, prevent any further transcription events while resolve in progress
+    this.removeCustomEventListeners();
     // If bargin is false and ws application return ack to verb:hook
     // the gather should not play any audio
     this._killAudio(this.cs);

lib/tasks/llm/llms/ultravox_s2s.js

@@ -146,8 +146,9 @@ class TaskLlmUltravox_S2S extends Task {
     return data;
   }
 
-  _unregisterHandlers() {
+  _unregisterHandlers(ep) {
     this.removeCustomEventListeners();
+    ep.removeAllListeners('dtmf');
   }
 
   _registerHandlers(ep) {
@@ -155,6 +156,7 @@ class TaskLlmUltravox_S2S extends Task {
     this.addCustomEventListener(ep, LlmEvents_Ultravox.ConnectFailure, this._onConnectFailure.bind(this, ep));
     this.addCustomEventListener(ep, LlmEvents_Ultravox.Disconnect, this._onDisconnect.bind(this, ep));
     this.addCustomEventListener(ep, LlmEvents_Ultravox.ServerEvent, this._onServerEvent.bind(this, ep));
+    ep.on('dtmf', this._onDtmf.bind(this, ep));
   }
 
   async _startListening(cs, ep) {
@@ -189,7 +191,7 @@ class TaskLlmUltravox_S2S extends Task {
     /* note: the parent llm verb started the span, which is why this is necessary */
     await this.parent.performAction(this.results);
 
-    this._unregisterHandlers();
+    this._unregisterHandlers(ep);
   }
 
   async kill(cs) {
@@ -346,6 +348,18 @@ class TaskLlmUltravox_S2S extends Task {
       excludeEvents: this.excludeEvents
     }, 'TaskLlmUltravox_S2S:_populateEvents');
   }
+
+  _onDtmf(ep, evt) {
+    this.logger.info({evt}, 'TaskLlmUltravox_S2S:_onDtmf - DTMF received');
+    const {dtmf} = evt;
+    const data = {
+      type: 'user_text_message',
+      text: `DTMF received: ${dtmf}`,
+      urgency: 'immediate'
+    };
+    this._api(ep, [ep.uuid, ClientEvent, JSON.stringify(data)])
+      .catch((err) => this.logger.info({err, evt}, 'TaskLlmUltravox_S2S:_onDtmf - Error sending DTMF as text message'));
+  }
 }
 
 module.exports = TaskLlmUltravox_S2S;

lib/tasks/say.js

@@ -1,6 +1,7 @@
 const assert = require('assert');
 const TtsTask = require('./tts-task');
 const {TaskName, TaskPreconditions} = require('../utils/constants');
+const {JAMBONES_SAY_CHUNK_SIZE} = require('../config');
 const pollySSMLSplit = require('polly-ssml-split');
 const { SpeechCredentialError, NonFatalTaskError } = require('../utils/error');
 const { sleepFor } = require('../utils/helpers');
@@ -31,7 +32,7 @@ const isMatchingEvent = (logger, filename, playbackId, evt) => {
 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 chunkSize = JAMBONES_SAY_CHUNK_SIZE;
   const isSSML = text.startsWith('<speak>');
   const options = {
     softLimit: 100,

lib/tasks/stt-task.js

@@ -286,6 +286,7 @@ class SttTask extends Task {
         account_sid: cs.accountSid,
         alert_type: AlertType.STT_NOT_PROVISIONED,
         vendor,
+        label,
         target_sid: cs.callSid
       }).catch((err) => this.logger.info({err}, 'Error generating alert for no stt'));
       // the ASR might have fallback configuration, should not done task here.
@@ -486,6 +487,7 @@ class SttTask extends Task {
       message: 'STT failure reported by vendor',
       detail: evt.error,
       vendor: this.vendor,
+      label: this.label,
       target_sid: cs.callSid
     }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
   }
@@ -499,6 +501,7 @@ class SttTask extends Task {
       alert_type: AlertType.STT_FAILURE,
       message: `Failed connecting to ${this.vendor} speech recognizer: ${reason}`,
       vendor: this.vendor,
+      label: this.label,
       target_sid: cs.callSid
     }).catch((err) => this.logger.info({err}, `Error generating alert for ${this.vendor} connection failure`));
   }

lib/tasks/tts-task.js

@@ -89,8 +89,9 @@ class TtsTask extends Task {
     // api_key, model_id, api_uri, custom_tts_streaming_url, and auth_token are encoded in the credentials
     // allow them to be overriden via config, using options
     // give preference to options passed in via config
-    const local_options = {...JSON.parse(options), ...this.options};
-    const local_voice_settings = {...JSON.parse(options).voice_settings, ...this.options.voice_settings};
+    const parsed_options = options ? JSON.parse(options) : {};
+    const local_options = {...parsed_options, ...this.options};
+    const local_voice_settings = {...(parsed_options.voice_settings || {}), ...(this.options.voice_settings || {})};
     const local_api_key =  local_options.api_key ?? api_key;
     const local_model_id = local_options.model_id ?? model_id;
     const local_api_uri = local_options.api_uri ?? api_uri;
@@ -273,6 +274,7 @@ class TtsTask extends Task {
           account_sid,
           alert_type: AlertType.TTS_NOT_PROVISIONED,
           vendor,
+          label,
           target_sid: cs.callSid
         }).catch((err) => this.logger.info({err}, 'Error generating alert for no tts'));
         throw new SpeechCredentialError('no provisioned speech credentials for TTS');
@@ -359,6 +361,7 @@ class TtsTask extends Task {
             account_sid: cs.accountSid,
             alert_type: AlertType.TTS_FAILURE,
             vendor,
+            label,
             detail: err.message,
             target_sid: cs.callSid
           }).catch((err) => this.logger.info({err}, 'Error generating alert for tts failure'));

lib/utils/http-requestor.js

@@ -191,7 +191,7 @@ class HttpRequestor extends BaseRequestor {
         method,
         headers: hdrs,
         ...('POST' === method && {body: JSON.stringify(payload)}),
-        timeout: HTTP_TIMEOUT,
+        headersTimeout: HTTP_TIMEOUT,
         followRedirects: false
       };
 

lib/utils/stt-latency-calculator.js

@@ -127,7 +127,6 @@ class SttLatencyCalculator extends Emitter {
 
   calculateLatency() {
     if (!this.isRunning) {
-      this.logger.debug('Latency calculator is not running, cannot calculate latency, returning default values');
       return null;
     }
 

lib/utils/tts-streaming-buffer.js

@@ -163,7 +163,6 @@ class TtsStreamingBuffer extends Emitter {
   }
 
   clear() {
-    this.logger.debug('TtsStreamingBuffer:clear');
     if (this._connectionStatus !== TtsStreamingConnectionStatus.Connected) return;
     clearTimeout(this.timer);
     this._api(this.ep, [this.ep.uuid, 'clear']).catch((err) =>
@@ -437,7 +436,15 @@ class TtsStreamingBuffer extends Emitter {
 
 const findSentenceBoundary = (text, limit) => {
   // Look for punctuation or double newline that signals sentence end.
-  const sentenceEndRegex = /[.!?](?=\s|$)|\n\n/g;
+  // Includes:
+  //   - ASCII: . ! ?
+  //   - Arabic: ؟ (question mark), ۔ (full stop)
+  //   - Japanese: 。 (full stop), ！, ？ (full-width exclamation/question)
+  //
+  // For languages that use spaces between sentences, we still require
+  // whitespace or end-of-string after the mark. For Japanese (no spaces),
+  // we treat the punctuation itself as a boundary regardless of following char.
+  const sentenceEndRegex = /[.!?؟۔](?=\s|$)|[。！？]|\n\n/g;
   let lastSentenceBoundary = -1;
   let match;
   while ((match = sentenceEndRegex.exec(text)) && match.index < limit) {

package.json

@@ -33,7 +33,7 @@
     "@jambonz/realtimedb-helpers": "^0.8.15",
     "@jambonz/speech-utils": "^0.2.26",
     "@jambonz/stats-collector": "^0.1.10",
-    "@jambonz/time-series": "^0.2.14",
+    "@jambonz/time-series": "^0.2.15",
     "@jambonz/verb-specifications": "^0.0.122",
     "@modelcontextprotocol/sdk": "^1.9.0",
     "@opentelemetry/api": "^1.8.0",