feat(elevenlabs): forward DTMF to the agent and support llm:update (#1592)

* feat(elevenlabs): forward DTMF to the agent and support llm:update

Adds opt-in llmOptions.dtmf {enable, interDigitTimeout, finishOnKey}.
Digits are buffered until finishOnKey, the inter-digit timeout or 50
digits, then sent as a user_message with source_medium "dtmf". Each
key press also sends user_activity so the agent turn timeout does not
fire while the caller is typing.

llm:update now forwards user_message, user_activity and
contextual_update client events instead of ignoring them.

Client events escape "|" so payloads containing it are no longer
split by the api argument delimiter (this also fixes tool results).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(elevenlabs): send DTMF per digit like ultravox, drop llmOptions.dtmf

Match the ultravox adapter: DTMF is always forwarded, with no llmOptions
setting. Each digit is sent as soon as it arrives, as a user_message with
source_medium "dtmf". Digit buffering, the inter-digit timeout and
finishOnKey are left to the agent's dtmf_input_settings.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(elevenlabs): buffer DTMF behind llmOptions.dtmf again

Per-digit sending does not work for ElevenLabs. Over the agent
WebSocket each user_message becomes its own turn: "1234#" produced 5
turns, and the agent answered "Are you still there?". The same digits
sent as a single message produce one dtmf turn, "1234".

- restores the opt-in llmOptions.dtmf {enable, interDigitTimeout, finishOnKey}
- digits are buffered and sent as one user_message, source_medium "dtmf"

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat: shared DtmfInput util with gather dtmf semantics

Any llm or agent vendor can turn caller DTMF into whole entries without
copying gather or touching it.

- lib/utils/dtmf-input.js uses gather options: finishOnKey, numDigits,
  minDigits, maxDigits, interDigitTimeout, timeout (15s default, 0 = off)
- no invented defaults: no implicit "#", no 2s timeout, no 50-digit cap
- minDigits defaults to 1; in gather an unset minDigits never arms the
  inter-digit timer
- stop() removes only its own dtmf listener
- elevenlabs uses it behind llmOptions.dtmf.enable

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(dtmf): review fixes for DtmfInput and elevenlabs

- inter-digit timer replaces the overall timeout while typing, as in gather
- warn when no option can ever complete an entry
- no DtmfInput when the session connects after kill (listener leak)
- llm:update rejects user_message/contextual_update without string text,
  since an ElevenLabs error event ends the conversation
- unit tests use mock timers instead of wall-clock waits

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

lib/tasks/llm/llms/elevenlabs_s2s.js

@@ -2,8 +2,10 @@ const Task = require('../../task');
 const TaskName = 'Llm_Elevenlabs_s2s';
 const {LlmEvents_Elevenlabs} = require('../../../utils/constants');
 const {request} = require('undici');
+const DtmfInput = require('../../../utils/dtmf-input');
 const ClientEvent = 'client.event';
 const SessionDelete = 'session.delete';
+const llmUpdateTypes = ['user_message', 'user_activity', 'contextual_update'];
 
 const elevenlabs_server_events = [
   'conversation_initiation_metadata',
@@ -47,11 +49,13 @@ class TaskLlmElevenlabs_S2S extends Task {
     const {
       conversation_initiation_client_data,
       input_sample_rate = 16000,
-      output_sample_rate = 16000
+      output_sample_rate = 16000,
+      dtmf
     } = this.data.llmOptions;
     this.conversation_initiation_client_data = conversation_initiation_client_data;
     this.input_sample_rate = input_sample_rate;
     this.output_sample_rate = output_sample_rate;
+    if (dtmf?.enable) this.dtmfOpts = dtmf;
     this.results = {
       completionReason: 'normal conversation end'
     };
@@ -105,7 +109,7 @@ class TaskLlmElevenlabs_S2S extends Task {
   async _api(ep, args) {
     const res = await ep.api('uuid_elevenlabs_s2s', `^^|${args.join('|')}`);
     if (!res.body?.startsWith('+OK')) {
-      throw new Error({args}, `Error calling uuid_elevenlabs_s2s: ${res.body}`);
+      throw new Error(`Error calling uuid_elevenlabs_s2s: ${res.body}`);
     }
   }
 
@@ -114,6 +118,7 @@ class TaskLlmElevenlabs_S2S extends Task {
     await this._startListening(cs, ep);
 
     await this.awaitTaskDone();
+    this.dtmfInput?.stop();
 
     /* note: the parent llm verb started the span, which is why this is necessary */
     await this.parent.performAction(this.results);
@@ -123,6 +128,7 @@ class TaskLlmElevenlabs_S2S extends Task {
 
   async kill(cs) {
     super.kill(cs);
+    this.dtmfInput?.stop();
 
     this._api(cs.ep, [cs.ep.uuid, SessionDelete])
       .catch((err) => this.logger.info({err}, 'TaskLlmElevenlabs_S2S:kill - error deleting session'));
@@ -144,7 +150,7 @@ class TaskLlmElevenlabs_S2S extends Task {
           'TaskLlmElevenlabs_S2S:processToolOutput - invalid tool output, must be client_tool_result');
       }
       else {
-        await this._api(ep, [ep.uuid, ClientEvent, JSON.stringify(data)]);
+        await this._sendClientEvent(ep, data);
       }
     } catch (err) {
       this.logger.info({err}, 'TaskLlmElevenlabs_S2S:processToolOutput');
@@ -152,11 +158,16 @@ class TaskLlmElevenlabs_S2S extends Task {
   }
 
   /**
-   * Send a session.update to the Elevenlabs server
-   * Note: creating and deleting conversation items also supported as well as interrupting the assistant
+   * Forward a user_message, user_activity or contextual_update client event to the agent
    */
   async processLlmUpdate(ep, data, _callSid) {
-    this.logger.debug({data, _callSid}, 'TaskLlmElevenlabs_S2S:processLlmUpdate, ignored');
+    this.logger.debug({data, _callSid}, 'TaskLlmElevenlabs_S2S:processLlmUpdate');
+    // an ElevenLabs error event ends the conversation, so reject a malformed text message here
+    if (!llmUpdateTypes.includes(data?.type) || (data.type !== 'user_activity' && typeof data.text !== 'string')) {
+      this.logger.info({data}, 'TaskLlmElevenlabs_S2S:processLlmUpdate - invalid mid-call request');
+      return;
+    }
+    await this._sendClientEvent(ep, data);
   }
 
   async _startListening(cs, ep) {
@@ -178,7 +189,8 @@ class TaskLlmElevenlabs_S2S extends Task {
     let ok = true;
     this.logger.debug({obj}, 'TaskLlmElevenlabs_S2S:_sendClientEvent');
     try {
-      const args = [ep.uuid, ClientEvent, JSON.stringify(obj)];
+      // '|' is the api argument delimiter
+      const args = [ep.uuid, ClientEvent, JSON.stringify(obj).replace(/\|/g, '\\u007c')];
       await this._api(ep, args);
     } catch (err) {
       ok = false;
@@ -216,6 +228,14 @@ class TaskLlmElevenlabs_S2S extends Task {
 
   _onConnect(ep) {
     this.logger.debug('TaskLlmElevenlabs_S2S:_onConnect');
+    if (this.dtmfOpts && !this.dtmfInput && !this.killed) {
+      this.dtmfInput = new DtmfInput(this.logger, this.dtmfOpts, {
+        // keeps the agent's turn timeout from firing while the caller is typing
+        onKey: () => this._sendClientEvent(ep, {type: 'user_activity'}),
+        onEntry: (text) => this._sendClientEvent(ep, {type: 'user_message', text, source_medium: 'dtmf'})
+      });
+      this.dtmfInput.start(ep);
+    }
     this._sendInitialMessage(ep);
   }
   _onConnectFailure(_ep, evt) {

lib/utils/dtmf-input.js

@@ -0,0 +1,91 @@
+/**
+ * Collects caller DTMF into whole entries for vendors that take keypad input as text.
+ * Options and termination rules follow the gather verb (seconds; timeout 15 by default, 0 = off).
+ */
+const DEFAULT_TIMEOUT = 15;
+
+const seconds = (v) => (typeof v === 'number' && v >= 0 ? v : undefined);
+
+class DtmfInput {
+  /**
+   * @param {object} logger - a pino logger
+   * @param {object} opts - gather-style dtmf options
+   * @param {object} handlers - onKey(digit) for every key press, onEntry(digits) for each completed entry
+   */
+  constructor(logger, opts = {}, {onKey, onEntry}) {
+    this.logger = logger;
+    this.finishOnKey = opts.finishOnKey;
+    this.numDigits = opts.numDigits;
+    this.maxDigits = opts.maxDigits;
+    // gather never arms its inter-digit timer when minDigits is unset; 1 makes interDigitTimeout work on its own
+    this.minDigits = opts.minDigits || 1;
+    this.interDigitTimeout = (seconds(opts.interDigitTimeout) || 0) * 1000;
+    const timeout = seconds(opts.timeout);
+    this.timeout = (timeout === undefined ? DEFAULT_TIMEOUT : timeout) * 1000;
+    this.onKey = onKey;
+    this.onEntry = onEntry;
+    this.buffer = '';
+    if (!this.finishOnKey && !this.numDigits && !this.maxDigits && !this.interDigitTimeout && !this.timeout) {
+      this.logger.warn({opts}, 'DtmfInput: no option ever completes an entry, digits will never be sent');
+    }
+  }
+
+  start(ep) {
+    if (this._onDtmf) return;
+    this.ep = ep;
+    this._onDtmf = (evt) => this.process(evt.dtmf);
+    ep.on('dtmf', this._onDtmf);
+  }
+
+  stop() {
+    if (this._onDtmf) this.ep.removeListener('dtmf', this._onDtmf);
+    this._onDtmf = null;
+    this._clearTimers();
+    if (this.buffer) this.logger.info({digits: this.buffer}, 'DtmfInput: discarding an incomplete entry');
+    this.buffer = '';
+  }
+
+  process(digit) {
+    this.onKey?.(digit);
+    clearTimeout(this.interDigitTimer);
+    if (!this.timer && this.timeout > 0) {
+      this.timer = setTimeout(this._onTimeout.bind(this), this.timeout);
+    }
+    if (digit === this.finishOnKey) return this._complete();
+
+    this.buffer += digit;
+    const len = this.buffer.length;
+    if (len === this.numDigits || len === this.maxDigits) return this._complete();
+    if (this.interDigitTimeout > 0 && len >= this.minDigits) {
+      // as in gather, the inter-digit timer replaces the overall timeout while the caller keeps typing
+      clearTimeout(this.timer);
+      this.timer = null;
+      this.interDigitTimer = setTimeout(this._complete.bind(this), this.interDigitTimeout);
+    }
+  }
+
+  _onTimeout() {
+    if (this.buffer.length < this.minDigits) {
+      this.logger.info({digits: this.buffer}, 'DtmfInput: timeout before minDigits, discarding entry');
+      this._clearTimers();
+      this.buffer = '';
+      return;
+    }
+    this._complete();
+  }
+
+  _complete() {
+    this._clearTimers();
+    const digits = this.buffer;
+    this.buffer = '';
+    if (digits) this.onEntry(digits);
+  }
+
+  _clearTimers() {
+    clearTimeout(this.interDigitTimer);
+    clearTimeout(this.timer);
+    this.interDigitTimer = this.timer = null;
+  }
+}
+
+module.exports = DtmfInput;

test/unit/dtmf-input.test.js

@@ -0,0 +1,122 @@
+const test = require('node:test');
+const assert = require('node:assert');
+const Emitter = require('events');
+
+const DtmfInput = require('../../lib/utils/dtmf-input');
+
+const noop = () => {};
+
+const make = (t, opts) => {
+  t.mock.timers.enable({apis: ['setTimeout']});
+  const keys = [];
+  const entries = [];
+  const warnings = [];
+  const logger = {error: noop, info: noop, debug: noop, warn: (...args) => warnings.push(args)};
+  const ep = new Emitter();
+  const input = new DtmfInput(logger, opts, {
+    onKey: (d) => keys.push(d),
+    onEntry: (digits) => entries.push(digits)
+  });
+  input.start(ep);
+  const press = (digits) => [...digits].forEach((dtmf) => ep.emit('dtmf', {dtmf}));
+  const tick = (secs) => t.mock.timers.tick(secs * 1000);
+  return {ep, input, keys, entries, warnings, press, tick};
+};
+
+test('finishOnKey completes the entry and is not included', (t) => {
+  const {keys, entries, press} = make(t, {finishOnKey: '#'});
+  press('1234#');
+  assert.deepStrictEqual(entries, ['1234']);
+  assert.deepStrictEqual(keys, ['1', '2', '3', '4', '#']);
+});
+
+test('finishOnKey on an empty buffer sends nothing', (t) => {
+  const {entries, press} = make(t, {finishOnKey: '#'});
+  press('#');
+  assert.deepStrictEqual(entries, []);
+});
+
+test('no finishOnKey by default: # is an ordinary digit', (t) => {
+  const {entries, press} = make(t, {numDigits: 3});
+  press('1#2');
+  assert.deepStrictEqual(entries, ['1#2']);
+});
+
+test('numDigits completes the entry', (t) => {
+  const {entries, press} = make(t, {numDigits: 2});
+  press('1234');
+  assert.deepStrictEqual(entries, ['12', '34']);
+});
+
+test('maxDigits completes the entry', (t) => {
+  const {entries, press} = make(t, {maxDigits: 3});
+  press('12345');
+  assert.deepStrictEqual(entries, ['123']);
+});
+
+test('each key restarts the inter-digit timer', (t) => {
+  const {entries, press, tick} = make(t, {interDigitTimeout: 2});
+  press('56');
+  tick(1.5);
+  press('7');
+  tick(1.5);
+  assert.deepStrictEqual(entries, []);
+  tick(0.5);
+  assert.deepStrictEqual(entries, ['567']);
+});
+
+test('the inter-digit timer replaces the overall timeout while typing', (t) => {
+  const {entries, press, tick} = make(t, {interDigitTimeout: 5, timeout: 15});
+  for (let i = 0; i < 20; i++) {
+    press(String(i % 10));
+    tick(1);
+  }
+  assert.deepStrictEqual(entries, []);
+  tick(4);
+  assert.deepStrictEqual(entries, ['01234567890123456789']);
+});
+
+test('interDigitTimeout waits for minDigits', (t) => {
+  const {entries, press, tick} = make(t, {interDigitTimeout: 1, minDigits: 3, timeout: 0});
+  press('12');
+  tick(5);
+  assert.deepStrictEqual(entries, []);
+  press('3');
+  tick(1);
+  assert.deepStrictEqual(entries, ['123']);
+});
+
+test('timeout defaults to 15s and completes the entry', (t) => {
+  const {entries, press, tick} = make(t, {});
+  press('42');
+  tick(14.9);
+  assert.deepStrictEqual(entries, []);
+  tick(0.1);
+  assert.deepStrictEqual(entries, ['42']);
+});
+
+test('timeout below minDigits drops the entry', (t) => {
+  const {entries, press, tick} = make(t, {timeout: 1, minDigits: 3});
+  press('42');
+  tick(1);
+  assert.deepStrictEqual(entries, []);
+});
+
+test('warns when no option can ever complete an entry', (t) => {
+  assert.strictEqual(make(t, {timeout: 0}).warnings.length, 1);
+});
+
+test('no warning when one option completes an entry', (t) => {
+  assert.strictEqual(make(t, {timeout: 0, finishOnKey: '#'}).warnings.length, 0);
+});
+
+test('stop removes only its own listener and discards the buffer', (t) => {
+  const {ep, input, entries, press, tick} = make(t, {interDigitTimeout: 1});
+  ep.on('dtmf', noop);
+  press('9');
+  input.stop();
+  tick(2);
+  press('8');
+  assert.deepStrictEqual(entries, []);
+  assert.deepStrictEqual(ep.listeners('dtmf'), [noop]);
+});