Update background-task-manager.js

support rimelabs tts streaming (#1047 )
Feat/1041 (#1045 )
2026-04-03 14:06:03 +00:00 · 2025-01-22 09:55:14 +00:00 · 2025-01-18 08:17:33 -05:00 · 2025-01-17 08:48:39 -05:00 · 2025-01-16 10:13:00 -05:00 · 2025-01-16 08:05:17 -05:00
16 changed files with 109 additions and 29 deletions
--- a/lib/middleware.js
+++ b/lib/middleware.js
@@ -1,5 +1,5 @@
 const uuidv4 = require('uuid-random');
-const {CallDirection, AllowedSipRecVerbs} = require('./utils/constants');
+const {CallDirection, AllowedSipRecVerbs, WS_CLOSE_CODES} = require('./utils/constants');
 const {parseSiprecPayload} = require('./utils/siprec-utils');
 const CallInfo = require('./session/call-info');
 const HttpRequestor = require('./utils/http-requestor');
@@ -460,7 +460,7 @@ module.exports = function(srf, logger) {
      }).catch((err) => this.logger.info({err}, 'Error generating alert for parsing application'));
      logger.info({err}, `Error retrieving or parsing application: ${err?.message}`);
      res.send(480, {headers: {'X-Reason': err?.message || 'unknown'}});
-      app.requestor.close();
+      app.requestor.close(WS_CLOSE_CODES.GoingAway);
    }
  }

--- a/lib/session/call-session.js
+++ b/lib/session/call-session.js
@@ -1468,7 +1468,7 @@ class CallSession extends Emitter {
    if (!listenTask) {
      return this.logger.info('CallSession:_lccListenStatus - invalid listen_status: Dial does not have a listen');
    }
-    listenTask.updateListen(opts.listen_status);
+    listenTask.updateListen(opts.listen_status || opts.stream_status);
  }

  /**
@@ -1610,7 +1610,7 @@ Duration=${duration} `
    // this whole thing requires us to be in a Dial verb
    const task = this.currentTask;
    if (!task || ![TaskName.Dial, TaskName.Listen].includes(task.name)) {
-      return this.logger.info('CallSession:_lccWhisper - invalid command since we are not in a dial or listen');
+      return this.logger.info('CallSession:_lccWhisper - invalid command since we are not in a dial or stream/listen');
    }

    // allow user to provide a url object, a url string, an array of tasks, or a single task
@@ -1799,7 +1799,7 @@ Duration=${duration} `
    if (opts.call_hook || opts.child_call_hook) {
      return await this._lccCallHook(opts);
    }
-    if (opts.listen_status) {
+    if (opts.listen_status || opts.stream_status) {
      await this._lccListenStatus(opts);
    }
    if (opts.transcribe_status) {
@@ -2109,6 +2109,7 @@ Duration=${duration} `
        break;

      case 'listen:status':
+      case 'stream:status':
        this._lccListenStatus(data);
        break;

--- a/lib/tasks/dial.js
+++ b/lib/tasks/dial.js
@@ -121,8 +121,9 @@ class TaskDial extends Task {
      }
    }

-    if (this.data.listen) {
-      this.listenTask = makeTask(logger, {'listen': this.data.listen}, this);
+    const listenData = this.data.listen || this.data.stream;
+    if (listenData) {
+      this.listenTask = makeTask(logger, {'listen': listenData }, this);
    }
    if (this.data.transcribe) {
      this.transcribeTask = makeTask(logger, {'transcribe' : this.data.transcribe}, this);
@@ -873,7 +874,7 @@ class TaskDial extends Task {
    if (cs.sipRequestWithinDialogHook) this._initSipIndialogRequestListener(cs, this.dlg);

    if (this.transcribeTask) this.transcribeTask.exec(cs, {ep: this.epOther, ep2:this.ep});
-    if (this.listenTask) this.listenTask.exec(cs, {ep: this.epOther});
+    if (this.listenTask) this.listenTask.exec(cs, {ep: this.listenTask.channel === 2 ? this.ep : this.epOther});
    if (this.startAmd) {
      try {
        this.startAmd(cs, this.ep, this, this.data.amd);
--- a/lib/tasks/gather.js
+++ b/lib/tasks/gather.js
@@ -24,6 +24,7 @@ const makeTask = require('./make_task');
 const assert = require('assert');
 const SttTask = require('./stt-task');
 const { SpeechCredentialError } = require('../utils/error');
+const SPEECHMATICS_DEFAULT_ASR_TIMEOUT = 1200;

 class TaskGather extends SttTask {
  constructor(logger, opts, parentTask) {
@@ -162,6 +163,16 @@ class TaskGather extends SttTask {
      this.logger.debug({hints: this.data.recognizer.hints, hintsBoost: this.data.recognizer.hintsBoost},
        'Gather:exec - applying global sttHints');
    }
+
+    // specials case for speechmatics: they dont do endpointing so we need to enable continuous ASR
+    if (this.vendor === 'speechmatics' && !this.isContinuousAsr) {
+      const maxDelay = this.recognizer?.speechmaticsOptions?.transcription_config?.max_delay;
+      if (maxDelay) this.asrTimeout = Math.min(SPEECHMATICS_DEFAULT_ASR_TIMEOUT, maxDelay * 1000);
+      else this.asrTimeout = SPEECHMATICS_DEFAULT_ASR_TIMEOUT;
+      this.isContinuousAsr = true;
+      this.logger.debug(`Gather:exec - auto-enabling continuous ASR for speechmatics w/ timeout ${this.asrTimeout}`);
+    }
+
    if (!this.isContinuousAsr && cs.isContinuousAsr) {
      this.isContinuousAsr = true;
      this.asrTimeout = cs.asrTimeout * 1000;
@@ -832,7 +843,7 @@ class TaskGather extends SttTask {
        const t = evt.alternatives[0].transcript;
        if (t) {
          /* remove trailing punctuation */
-          if (/[,;:\.!\?]$/.test(t)) {
+          if (this.vendor !== 'speechmatics' && /[,;:\.!\?]$/.test(t)) {
            this.logger.debug('TaskGather:_onTranscription - removing trailing punctuation');
            evt.alternatives[0].transcript = t.slice(0, -1);
          }
--- a/lib/tasks/listen.js
+++ b/lib/tasks/listen.js
@@ -17,7 +17,7 @@ class TaskListen extends Task {

    [
      'action', 'auth', 'method', 'url', 'finishOnKey', 'maxLength', 'metadata', 'mixType', 'passDtmf', 'playBeep',
-      'sampleRate', 'timeout', 'transcribe', 'wsAuth', 'disableBidirectionalAudio'
+      'sampleRate', 'timeout', 'transcribe', 'wsAuth', 'disableBidirectionalAudio', 'channel'
    ].forEach((k) => this[k] = this.data[k]);

    this.mixType = this.mixType || 'mono';
--- a/lib/tasks/make_task.js
+++ b/lib/tasks/make_task.js
@@ -84,6 +84,7 @@ function makeTask(logger, obj, parent) {
      const TaskTranscribe = require('./transcribe');
      return new TaskTranscribe(logger, data, parent);
    case TaskName.Listen:
+    case TaskName.Stream:
      const TaskListen = require('./listen');
      return new TaskListen(logger, data, parent);
    case TaskName.Redirect:
--- a/lib/tasks/tts-task.js
+++ b/lib/tasks/tts-task.js
@@ -94,6 +94,23 @@ class TtsTask extends Task {
          ...(style && {ELEVENLABS_TTS_STREAMING_VOICE_SETTINGS_STYLE: style})
        };
        break;
+      case 'rimelabs':
+        const {
+          pauseBetweenBrackets, phonemizeBetweenBrackets, inlineSpeedAlpha, speedAlpha, reduceLatency
+        } = this.options;
+        obj = {
+          RIMELABS_API_KEY: api_key,
+          RIMELABS_TTS_STREAMING_MODEL_ID: model_id,
+          RIMELABS_TTS_STREAMING_VOICE_ID: voice,
+          RIMELABS_TTS_STREAMING_LANGUAGE: language || 'en',
+          ...(pauseBetweenBrackets && {RIMELABS_TTS_STREAMING_PAUSE_BETWEEN_BRACKETS: pauseBetweenBrackets}),
+          ...(phonemizeBetweenBrackets &&
+            {RIMELABS_TTS_STREAMING_PHONEMIZE_BETWEEN_BRACKETS: phonemizeBetweenBrackets}),
+          ...(inlineSpeedAlpha && {RIMELABS_TTS_STREAMING_INLINE_SPEED_ALPHA: inlineSpeedAlpha}),
+          ...(speedAlpha && {RIMELABS_TTS_STREAMING_SPEED_ALPHA: speedAlpha}),
+          ...(reduceLatency && {RIMELABS_TTS_STREAMING_REDUCE_LATENCY: reduceLatency})
+        };
+        break;
      default:
        if (vendor.startsWith('custom:')) {
          const use_tls = custom_tts_streaming_url.startsWith('wss://');
--- a/lib/utils/background-task-manager.js
+++ b/lib/utils/background-task-manager.js
@@ -79,7 +79,7 @@ class BackgroundTaskManager extends Emitter {
  }

  // Initiate Listen
-  async _initListen(opts, bugname = 'jambonz-background-listen', ignoreCustomerData = false, type = 'listen') {
+  async _initListen(opts, bugname = 'jambonz-background-listen', ignoreCustomerData = true, type = 'listen') {
    let task;
    try {
      const t = normalizeJambones(this.logger, [opts]);
--- a/lib/utils/constants.json
+++ b/lib/utils/constants.json
@@ -28,6 +28,7 @@
    "SipRedirect": "sip:redirect",
    "Say": "say",
    "SayLegacy": "say:legacy",
+    "Stream": "stream",
    "Tag": "tag",
    "Transcribe": "transcribe"
  },
@@ -258,6 +259,11 @@
    "ConnectFailure": "elevenlabs_tts_streaming::connect_failed",
    "Connect": "elevenlabs_tts_streaming::connect"
  },
+  "RimelabsTtsStreamingEvents": {
+    "Empty": "rimelabs_tts_streaming::empty",
+    "ConnectFailure": "rimelabs_tts_streaming::connect_failed",
+    "Connect": "rimelabs_tts_streaming::connect"
+  },
  "CustomTtsStreamingEvents": {
    "Empty": "custom_tts_streaming::empty",
    "ConnectFailure": "custom_tts_streaming::connect_failed",
@@ -283,5 +289,9 @@
    "Offline": "OFFLINE",
    "GracefulShutdownInProgress":"SHUTDOWN_IN_PROGRESS"
  },
-  "FEATURE_SERVER" : "feature-server"
+  "FEATURE_SERVER" : "feature-server",
+  "WS_CLOSE_CODES": {
+    "NormalClosure": 1000,
+    "GoingAway": 1001
+  }
 }
--- a/lib/utils/error.js
+++ b/lib/utils/error.js
@@ -17,8 +17,17 @@ class PlayFileNotFoundError extends NonFatalTaskError {
  }
 }

+class HTTPResponseError extends Error {
+  constructor(statusCode) {
+    super('Unexpected HTTP Response');
+    delete this.stack;
+    this.statusCode = statusCode;
+  }
+}
+
 module.exports = {
  SpeechCredentialError,
  NonFatalTaskError,
-  PlayFileNotFoundError
+  PlayFileNotFoundError,
+  HTTPResponseError
 };
--- a/lib/utils/http-requestor.js
+++ b/lib/utils/http-requestor.js
@@ -16,6 +16,7 @@ const {
  NODE_ENV,
  HTTP_USER_AGENT_HEADER,
 } = require('../config');
+const {HTTPResponseError} = require('./error');

 const toBase64 = (str) => Buffer.from(str || '', 'utf8').toString('base64');

@@ -190,8 +191,7 @@ class HttpRequestor extends BaseRequestor {
        followRedirects: false
      });
      if (![200, 202, 204].includes(statusCode)) {
-        const err = new Error();
-        err.statusCode = statusCode;
+        const err = new HTTPResponseError(statusCode);
        throw err;
      }
      if (headers['content-type']?.includes('application/json')) {
--- a/lib/utils/transcription-utils.js
+++ b/lib/utils/transcription-utils.js
@@ -209,7 +209,7 @@ const consolidateTranscripts = (bufferedTranscripts, channel, language, vendor)
    const lastChar = acc.alternatives[0].transcript.slice(-1);
    const firstChar = newTranscript.charAt(0);

-    if (lastChar.match(/\d/) && firstChar.match(/\d/)) {
+    if (vendor === 'speechmatics' || (lastChar.match(/\d/) && firstChar.match(/\d/))) {
      acc.alternatives[0].transcript += newTranscript;
    } else {
      acc.alternatives[0].transcript += ` ${newTranscript}`;
@@ -957,6 +957,36 @@ module.exports = (logger) => {
            SPEECHMATICS_TRANSLATION_PARTIALS: speechmaticsOptions.translation_config.enable_partials ? 1 : 0
          }
        ),
+        ...(speechmaticsOptions.transcription_config?.domain &&
+          {SPEECHMATICS_DOMAIN: speechmaticsOptions.transcription_config.domain}),
+        ...{SPEECHMATICS_MAX_DELAY: speechmaticsOptions.transcription_config?.max_delay || 0.7},
+        ...{SPEECHMATICS_MAX_DELAY_MODE: speechmaticsOptions.transcription_config?.max_delay_mode || 'flexible'},
+        ...(speechmaticsOptions.transcription_config?.diarization &&
+            {SPEECHMATICS_DIARIZATION: speechmaticsOptions.transcription_config.diarization}),
+        ...(speechmaticsOptions.transcription_config?.speaker_diarization_config?.speaker_sensitivity &&
+          {SPEECHMATICS_DIARIZATION_SPEAKER_SENSITIVITY:
+            speechmaticsOptions.transcription_config.speaker_diarization_config.speaker_sensitivity}),
+        ...(speechmaticsOptions.transcription_config?.speaker_diarization_config?.max_speakers &&
+          {SPEECHMATICS_DIARIZATION_MAX_SPEAKERS:
+            speechmaticsOptions.transcription_config.speaker_diarization_config.max_speakers}),
+        ...(speechmaticsOptions.transcription_config?.output_locale &&
+          {SPEECHMATICS_OUTPUT_LOCALE: speechmaticsOptions.transcription_config.output_locale}),
+        ...(speechmaticsOptions.transcription_config?.punctuation_overrides?.permitted_marks &&
+          {SPEECHMATICS_PUNCTUATION_ALLOWED:
+            speechmaticsOptions.transcription_config.punctuation_overrides.permitted_marks.join(',')}),
+        ...(speechmaticsOptions.transcription_config?.punctuation_overrides?.sensitivity &&
+          {SPEECHMATICS_PUNCTUATION_SENSITIVITY:
+            speechmaticsOptions.transcription_config?.punctuation_overrides?.sensitivity}),
+        ...(speechmaticsOptions.transcription_config?.operating_point &&
+          {SPEECHMATICS_OPERATING_POINT: speechmaticsOptions.transcription_config.operating_point}),
+        ...(speechmaticsOptions.transcription_config?.enable_entities &&
+          {SPEECHMATICS_ENABLE_ENTTIES: speechmaticsOptions.transcription_config.enable_entities}),
+        ...(speechmaticsOptions.transcription_config?.audio_filtering_config?.volume_threshold &&
+          {SPEECHMATICS_VOLUME_THRESHOLD:
+            speechmaticsOptions.transcription_config.audio_filtering_config.volume_threshold}),
+        ...(speechmaticsOptions.transcription_config?.transcript_filtering_config?.remove_disfluencies &&
+          {SPEECHMATICS_REMOVE_DISFLUENCIES:
+            speechmaticsOptions.transcription_config.transcript_filtering_config.remove_disfluencies})
      };
    }
    else if (vendor.startsWith('custom:')) {
--- a/lib/utils/tts-streaming-buffer.js
+++ b/lib/utils/tts-streaming-buffer.js
@@ -278,6 +278,7 @@ class TtsStreamingBuffer extends Emitter {
      'deepgram',
      'cartesia',
      'elevenlabs',
+      'rimelabs',
      'custom'
    ].forEach((vendor) => {
      const eventClassName = `${vendor.charAt(0).toUpperCase() + vendor.slice(1)}TtsStreamingEvents`;
--- a/lib/utils/ws-requestor.js
+++ b/lib/utils/ws-requestor.js
@@ -1,7 +1,7 @@
 const assert = require('assert');
 const BaseRequestor = require('./base-requestor');
 const short = require('short-uuid');
-const {HookMsgTypes} = require('./constants.json');
+const {HookMsgTypes, WS_CLOSE_CODES} = require('./constants.json');
 const Websocket = require('ws');
 const snakeCaseKeys = require('./snakecase-keys');
 const {
@@ -261,13 +261,13 @@ class WsRequestor extends BaseRequestor {
    }
  }

-  close() {
+  close(code = WS_CLOSE_CODES.NormalClosure) {
    this.closedGracefully = true;
-    this.logger.debug('WsRequestor:close closing socket');
+    this.logger.debug(`WsRequestor:close closing socket with code ${code}`);
    this._stopPingTimer();
    try {
      if (this.ws) {
-        this.ws.close(1000);
+        this.ws.close(code);
        this.ws.removeAllListeners();
        this.ws = null;
      }
--- a/package-lock.json
+++ b/package-lock.json
@@ -18,7 +18,7 @@
        "@jambonz/speech-utils": "^0.2.1",
        "@jambonz/stats-collector": "^0.1.10",
        "@jambonz/time-series": "^0.2.13",
-        "@jambonz/verb-specifications": "^0.0.92",
+        "@jambonz/verb-specifications": "^0.0.94",
        "@opentelemetry/api": "^1.8.0",
        "@opentelemetry/exporter-jaeger": "^1.23.0",
        "@opentelemetry/exporter-trace-otlp-http": "^0.50.0",
@@ -1671,10 +1671,9 @@
      }
    },
    "node_modules/@jambonz/verb-specifications": {
-      "version": "0.0.92",
-      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.92.tgz",
-      "integrity": "sha512-zb1y5Hq+FqGYleYYKZafEIHyhhlH3VHapTJh3N0s+2xdy8I2Gf17zJpUc45mhV/4ficlT3SSwETsBbMt20Hwog==",
-      "license": "MIT",
+      "version": "0.0.94",
+      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.94.tgz",
+      "integrity": "sha512-gFqZvbzM+us9T2CLkMaFSjlnclTGCMzuP9BD9fDPJNU36/CaIX3TO+3/EDCcIVhg4+b/smr6cqTHjaLQ5fZn4g==",
      "dependencies": {
        "debug": "^4.3.4",
        "pino": "^8.8.0"
@@ -10939,9 +10938,9 @@
      }
    },
    "@jambonz/verb-specifications": {
-      "version": "0.0.92",
-      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.92.tgz",
-      "integrity": "sha512-zb1y5Hq+FqGYleYYKZafEIHyhhlH3VHapTJh3N0s+2xdy8I2Gf17zJpUc45mhV/4ficlT3SSwETsBbMt20Hwog==",
+      "version": "0.0.94",
+      "resolved": "https://registry.npmjs.org/@jambonz/verb-specifications/-/verb-specifications-0.0.94.tgz",
+      "integrity": "sha512-gFqZvbzM+us9T2CLkMaFSjlnclTGCMzuP9BD9fDPJNU36/CaIX3TO+3/EDCcIVhg4+b/smr6cqTHjaLQ5fZn4g==",
      "requires": {
        "debug": "^4.3.4",
        "pino": "^8.8.0"
--- a/package.json
+++ b/package.json
@@ -33,7 +33,7 @@
    "@jambonz/realtimedb-helpers": "^0.8.8",
    "@jambonz/speech-utils": "^0.2.1",
    "@jambonz/stats-collector": "^0.1.10",
-    "@jambonz/verb-specifications": "^0.0.92",
+    "@jambonz/verb-specifications": "^0.0.94",
    "@jambonz/time-series": "^0.2.13",
    "@opentelemetry/api": "^1.8.0",
    "@opentelemetry/exporter-jaeger": "^1.23.0",
Author	SHA1	Message	Date
Sam Machin	f8b1c429e0	Update background-task-manager.js	2025-01-22 09:55:14 +00:00
Hoan Luu Huu	ba282d775d	support rimelabs tts streaming (#1047 )	2025-01-18 08:17:33 -05:00
Dave Horton	a194ba833e	Feat/1041 (#1045 ) * initial changes for stream synonym to listen * listen on B endpoint if nested listen in dial has channel === 2	2025-01-17 08:48:39 -05:00
rammohan-y	77f3d9d7ec	feat/1034: sending socket close code when there is no response from the websocket app (#1035 )	2025-01-16 10:13:00 -05:00
Sam Machin	4dbc7df93d	new error for HTTP responses without stack trace (#1044 ) * new error for HTTP responses without stack trace * lint	2025-01-16 08:05:17 -05:00
Dave Horton	f71f0ac69a	Fix/speechmatics (#1042 ) * add speechmatics options * wip * speechmatics does not do endpointing for us so we need to flip on continuousAsr * speechmatics: continousAsr should be at least equal to max_delay, if set	2025-01-15 19:12:15 -05:00