Fix concurrent agent joins and voice identities

.github/workflows/ci.yml

@@ -12,3 +12,4 @@ jobs:
       - run: bun install --frozen-lockfile
       - run: bun x tsc --noEmit
       - run: bun test
+      - run: node --test test/agent-phone.test.mjs

README.md

@@ -46,8 +46,9 @@ to the older phone path below.
 
 Agents can also join the way people do: they place a call to the number and key the code with DTMF.
 The fallback for ours runs on the fleet box: a worker claims invites from
-`POST /api/v1/agent-queue/next` and runs `agent-phone join`. That process speaks with
-ElevenLabs voices through Telnyx and hears the room through Telnyx transcription. Anyone
+`POST /api/v1/agent-queue/next` and runs `agent-phone join`. The tracked worker source is
+`ops/agent-phone.mjs`; its tests run with `node --test test/agent-phone.test.mjs`. It speaks
+through Telnyx and hears the room through Telnyx transcription. Anyone
 else's agent can do the same from any telephony API, after naming itself with
 `POST /api/v1/rooms/:code/expect`.
 

src/db.ts

@@ -36,7 +36,7 @@ export type AgentTask = { id: string; agent: string; code: string; roomName: str
 export type Pass = { id: string; sub: string; amountCents: number; status: string; checkoutUrl: string; days: number; paidAt: number | null; createdAt: number }
 
 /** A call the room placed (a phone rung in): what it was estimated to cost a minute and, once Telnyx says, what it cost. */
-export type DialledLeg = { leg: string; code: string; to: string; provider: string; rate: number; placedAt: number; endedAt: number | null; cost: number | null }
+export type DialledLeg = { leg: string; code: string; to: string; provider: string; name: string; rate: number; placedAt: number; endedAt: number | null; cost: number | null }
 
 export class Sealer {
   private readonly key: Buffer
@@ -114,7 +114,7 @@ export class Store {
       );
       CREATE INDEX IF NOT EXISTS leg_quality_code ON leg_quality(code, id DESC);
       CREATE TABLE IF NOT EXISTS dialled_legs (
-        leg TEXT PRIMARY KEY, code TEXT NOT NULL, to_number TEXT NOT NULL, provider TEXT NOT NULL,
+        leg TEXT PRIMARY KEY, code TEXT NOT NULL, to_number TEXT NOT NULL, provider TEXT NOT NULL, name TEXT NOT NULL DEFAULT '',
         rate REAL NOT NULL, placed_at INTEGER NOT NULL, ended_at INTEGER, cost REAL
       );
       CREATE INDEX IF NOT EXISTS dialled_legs_placed ON dialled_legs(placed_at);
@@ -188,7 +188,7 @@ export class Store {
     // Group texts: every number a room has texted, and whether it said STOP.
     this.db.exec('CREATE TABLE IF NOT EXISTS sms_numbers (number TEXT PRIMARY KEY, opted_out INTEGER NOT NULL DEFAULT 0, first_at INTEGER NOT NULL, updated_at INTEGER NOT NULL)')
     // Added later: an agent that waits in the room with early callers until the host arrives.
-    for (const sql of ['ALTER TABLE rooms ADD COLUMN greeter TEXT', 'ALTER TABLE rooms ADD COLUMN host TEXT', "ALTER TABLE agent_invites ADD COLUMN role TEXT NOT NULL DEFAULT ''"]) {
+    for (const sql of ['ALTER TABLE rooms ADD COLUMN greeter TEXT', 'ALTER TABLE rooms ADD COLUMN host TEXT', "ALTER TABLE agent_invites ADD COLUMN role TEXT NOT NULL DEFAULT ''", "ALTER TABLE dialled_legs ADD COLUMN name TEXT NOT NULL DEFAULT ''"]) {
       try {
         this.db.exec(sql)
       } catch {}
@@ -443,8 +443,8 @@ export class Store {
 
   // ---- what calls out cost ----------------------------------------------
 
-  addDialled(d: Omit<DialledLeg, 'endedAt' | 'cost'>): void {
-    this.db.query('INSERT OR REPLACE INTO dialled_legs (leg, code, to_number, provider, rate, placed_at) VALUES (?, ?, ?, ?, ?, ?)').run(d.leg, d.code, d.to, d.provider, d.rate, d.placedAt)
+  addDialled(d: Omit<DialledLeg, 'endedAt' | 'cost' | 'name'> & { name?: string }): void {
+    this.db.query('INSERT OR REPLACE INTO dialled_legs (leg, code, to_number, provider, name, rate, placed_at) VALUES (?, ?, ?, ?, ?, ?, ?)').run(d.leg, d.code, d.to, d.provider, d.name ?? '', d.rate, d.placedAt)
   }
 
   dialled(leg: string): DialledLeg | null {
@@ -641,4 +641,4 @@ export class Store {
   }
 }
 
-const dialledRow = (r: any): DialledLeg => ({ leg: r.leg, code: r.code, to: r.to_number, provider: r.provider, rate: r.rate, placedAt: r.placed_at, endedAt: r.ended_at, cost: r.cost })
+const dialledRow = (r: any): DialledLeg => ({ leg: r.leg, code: r.code, to: r.to_number, provider: r.provider, name: r.name ?? '', rate: r.rate, placedAt: r.placed_at, endedAt: r.ended_at, cost: r.cost })

src/service.ts

@@ -202,6 +202,8 @@ export class GroupCallService {
   readonly affiliates: Affiliates
   private readonly listeners = new Map<string, Set<(e: LiveEvent) => void>>()
   private readonly store: Store
+  /** Prevent simultaneous room-code webhooks from creating parallel conference bridges. */
+  private telnyxEventQueue: Promise<void> = Promise.resolve()
   readonly config: Config
   readonly broadcastRelay: LiveBroadcastRelay | null
   private readonly agentNumbers: Record<string, string>
@@ -487,7 +489,7 @@ export class GroupCallService {
     this.budgetFor(SIP_RATE, 'sip')
     const placed = await this.line.dial(room.code, uri.trim(), { from: this.config.dialIn, connectionId, name: clean, direct: true, headers })
     if (!placed) throw new UserError('The call could not be placed. Try again in a minute.', 'sip', 502)
-    this.store.addDialled({ leg: placed.leg, code: room.code, to: uri.trim(), provider: 'sip', rate: SIP_RATE, placedAt: Date.now() })
+    this.store.addDialled({ leg: placed.leg, code: room.code, to: uri.trim(), provider: 'sip', name: clean, rate: SIP_RATE, placedAt: Date.now() })
     const text = askedBy ? `is being called in by ${askedBy}` : 'is being called in'
     this.store.addEvent({ code: room.code, kind: 'invite', who: clean, text })
     this.emit({ type: 'invite', code: room.code, who: clean, text, callers: this.line.callersIn(room.code), at: Date.now() })
@@ -630,8 +632,11 @@ export class GroupCallService {
     }
     this.recordQuality(event)
     this.costEvent(event)
-    // Answered asynchronously: Telnyx wants a quick 200, and the commands go back on their own.
-    void this.line.handle(event).catch((e) => console.error('group call:', e))
+    // Return Telnyx's quick 200, then process events in arrival order. In
+    // particular, only one join may observe an empty room and create its bridge.
+    this.telnyxEventQueue = this.telnyxEventQueue
+      .then(() => this.line!.handle(event))
+      .catch((e) => console.error('group call:', e))
     return true
   }
 
@@ -643,6 +648,20 @@ export class GroupCallService {
     if (!this.line) return { rooms: 0, legs: 0 }
     const saved = this.store.liveSnapshot()
     if (saved.length === 0) return { rooms: 0, legs: 0 }
+    // SIP legs have no caller number to identify them after a restart. Keep the
+    // name beside the outbound leg and restore it before rebuilding the room.
+    for (const room of saved) {
+      for (const p of room.participants) {
+        if (p.name) continue
+        const dialled = this.store.dialled(p.leg)
+        if (dialled?.provider !== 'sip') continue
+        const configured = Object.entries(this.config.agentSip ?? {}).find(([, uri]) => uri.trim().toLowerCase() === dialled.to.trim().toLowerCase())?.[0]
+        const name = dialled.name || configured || ''
+        if (!name) continue
+        p.name = name
+        this.store.saveLeg(room.code, room.conferenceId, room.startedAt, p)
+      }
+    }
     const result = await this.line.restore(saved)
     const keep = new Map<string, Set<string>>()
     for (const r of saved) if (this.line.callersIn(r.code) > 0) keep.set(r.code, new Set(this.line.participants(r.code).map((p) => p.leg)))

test/agent-phone.test.mjs

@@ -0,0 +1,82 @@
+import assert from 'node:assert/strict';
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { after, test } from 'node:test';
+
+const testHome = mkdtempSync(join(tmpdir(), 'groupcall-agent-phone-'));
+process.env.AGENT_PHONE_HOME = testHome;
+const source = readFileSync(new URL('../ops/agent-phone.mjs', import.meta.url), 'utf8');
+const worker = await import(`data:text/javascript;base64,${Buffer.from(source).toString('base64')}`);
+after(() => {
+  delete process.env.AGENT_PHONE_HOME;
+  rmSync(testHome, { recursive: true, force: true });
+});
+
+const minute = 60_000;
+const now = Date.now();
+
+test('configured Polly voice bypasses exhausted ElevenLabs and omits ElevenLabs credentials', () => {
+  assert.equal(worker.voiceFor('riotcoder', 'AWS.Polly.Matthew-Neural'), worker.FALLBACK_VOICE);
+  assert.deepEqual(worker.speechRequest('Hello.', worker.FALLBACK_VOICE), {
+    payload: 'Hello.',
+    voice: 'AWS.Polly.Matthew-Neural',
+  });
+});
+
+test('ElevenLabs requests keep their API key reference', () => {
+  assert.deepEqual(worker.speechRequest('Hello.', 'ElevenLabs.Default.voice-id'), {
+    payload: 'Hello.',
+    voice: 'ElevenLabs.Default.voice-id',
+    voice_settings: { api_key_ref: 'qrypt_elevenlabs' },
+  });
+});
+
+test('a new call identifies its actual agent even when room history has a different agent', () => {
+  const history = [{ who: 'cha0spunk', text: 'old line', at: new Date(now - 15 * minute).toISOString(), history: true }];
+  assert.equal(worker.shouldIntroduce('riotcoder', history, now), true);
+  assert.equal(worker.agentIntro('riotcoder'), "Hi, I’m RiotCoder, Anthony’s AI agent. I’m here and listening.");
+  assert.notEqual(worker.agentIntro('riotcoder'), worker.agentIntro('cha0spunk'));
+});
+
+test('Nora gets her own spoken identity and answer routing', () => {
+  assert.equal(worker.agentIntro('nora'), "Hi, I’m Nora, Anthony’s AI agent. I’m here and listening.");
+  assert.equal(worker.addressed('nora', 'Nora, can you hear me'), true);
+  assert.equal(worker.addressed('riotcoder', 'Nora, can you hear me'), false);
+});
+
+test('a recent line from the same agent suppresses a duplicate intro on a fast redial', () => {
+  const recent = [{ who: 'riotcoder', text: 'Hi, I’m RiotCoder.', at: new Date(now - 30_000).toISOString() }];
+  assert.equal(worker.shouldIntroduce('riotcoder', recent, now), false);
+});
+
+test('room questions work when transcription omitted punctuation', () => {
+  assert.equal(worker.roomQuestion('Anthony: Who’s here'), true);
+  assert.equal(worker.roomQuestion('Anthony: Can you guys hear me'), true);
+  assert.equal(worker.roomQuestion('Anthony: I told everyone to listen'), false);
+});
+
+test('the name the listener hears addresses only that agent', () => {
+  assert.equal(worker.addressed('riotcoder', 'RiotCoder, are you there?'), true);
+  assert.equal(worker.addressed('cha0spunk', 'RiotCoder, are you there?'), false);
+  assert.equal(worker.addressed('cha0spunk', 'Chaos Punk, can you hear me?'), true);
+  assert.equal(worker.addressed('riotcoder', 'Chaos Punk, can you hear me?'), false);
+});
+
+test('all room agents can recognize their spoken names and avoid answering for another agent', () => {
+  for (const [agent, heard] of [
+    ['riotcoder', 'Riot Coder'],
+    ['cha0spunk', 'Chaos Punk'],
+    ['rootkey', 'Root Key'],
+    ['coinpay', 'Coin Pay'],
+    ['knox', 'Knocks'],
+    ['nora', 'Nora'],
+  ]) {
+    assert.equal(worker.addressed(agent, `${heard}, answer this please`), true, `${agent} should recognize ${heard}`);
+    for (const [other] of [
+      ['riotcoder'], ['cha0spunk'], ['rootkey'], ['coinpay'], ['knox'], ['nora'],
+    ]) {
+      if (other !== agent) assert.equal(worker.addressed(other, `${heard}, answer this please`), false, `${other} should not answer for ${agent}`);
+    }
+  }
+});

test/app.test.ts

@@ -19,6 +19,7 @@ let smshubMode: 'ok' | 'down' | 'badkey' = 'ok'
 /** A fetch that records what Telnyx was asked; a conference creation answers with an id. */
 function telnyxRecorder() {
   const calls: { path: string; body: any }[] = []
+  const participants = new Set<string>()
   const fetch = (async (url: string | URL | Request, init?: RequestInit) => {
     if (String(url).startsWith('https://files.test/')) return new Response(new Uint8Array([0x49, 0x44, 0x33, 1, 2, 3]), { status: 200 })
     if (String(url).startsWith('https://smshub.test/')) {
@@ -32,38 +33,43 @@ function telnyxRecorder() {
     const path = String(url).replace('https://api.telnyx.com/v2', '')
     calls.push({ path, body: JSON.parse(String(init?.body ?? '{}')) })
     if (path === '/messages' && smsReply === '40010') return new Response(JSON.stringify({ errors: [{ code: '40010', title: 'Not 10DLC registered' }] }), { status: 400 })
+    if (path === '/conferences/conf-1' && methodIsGet(init)) return new Response(JSON.stringify({ data: { id: 'conf-1', status: 'in_progress' } }))
+    if (path.startsWith('/conferences/conf-1/participants') && methodIsGet(init)) return new Response(JSON.stringify({ data: [...participants].map((id) => ({ call_control_id: id, status: 'joined' })) }))
     const json = path === '/conferences' ? { data: { id: 'conf-1' } } : path === '/calls' ? { data: { call_control_id: 'out-1' } } : {}
     return new Response(JSON.stringify(json), { status: 200 })
   }) as unknown as typeof globalThis.fetch
-  return { calls, fetch }
+  return { calls, fetch, participants }
 }
 
+const methodIsGet = (init?: RequestInit) => (init?.method ?? 'GET') === 'GET'
+
 const DIAL_IN = '+15106169327'
 const AGENT_CALLER = '+19492847328'
 
 function setup(extra: Record<string, unknown> = {}) {
   const keys = telnyxKeys()
   const telnyx = telnyxRecorder()
   const store = new Store(':memory:', new Sealer('k'.repeat(32)))
+  const config = {
+    key: 'k'.repeat(32),
+    clientId: 'cp_test',
+    coinpayBase: 'https://coinpayportal.com',
+    coinpayWebhookSecret: 'whsec',
+    telnyxApiKey: 'KEY_test',
+    telnyxPublicKey: keys.publicKey,
+    dialIn: DIAL_IN,
+    telnyxConnectionId: 'conn-gc',
+    agentOperators: ['sub-op'],
+    agentWorkerKey: 'worker-secret',
+    agentCallerIds: { riotcoder: AGENT_CALLER, cha0spunk: '+15107667447' },
+    fetch: telnyx.fetch,
+    ...extra,
+  }
   const { app, svc } = createApp({
     store,
     origin: 'https://group.call.test',
     secureCookies: false,
-    config: {
-      key: 'k'.repeat(32),
-      clientId: 'cp_test',
-      coinpayBase: 'https://coinpayportal.com',
-      coinpayWebhookSecret: 'whsec',
-      telnyxApiKey: 'KEY_test',
-      telnyxPublicKey: keys.publicKey,
-      dialIn: DIAL_IN,
-      telnyxConnectionId: 'conn-gc',
-      agentOperators: ['sub-op'],
-      agentWorkerKey: 'worker-secret',
-      agentCallerIds: { riotcoder: AGENT_CALLER, cha0spunk: '+15107667447' },
-      fetch: telnyx.fetch,
-      ...extra,
-    },
+    config,
   })
   const tokens = { accessToken: 'a', refreshToken: null, expiresAt: 0 }
   const op = store.upsertUser('sub-op', 'Anthony', 'op@example.com', tokens)
@@ -83,7 +89,7 @@ function setup(extra: Record<string, unknown> = {}) {
     await new Promise((resolve) => setTimeout(resolve, 5))
     return r
   }
-  return { app, svc, store, telnyx, call, event, op, guest }
+  return { app, svc, store, telnyx, call, event, op, guest, config }
 }
 
 describe('rooms', () => {
@@ -219,6 +225,44 @@ describe('agents', () => {
     expect(((await done.json()) as any).status).toBe('done')
   })
 
+  test('Nora has a dedicated caller ID, so a worker arrival is shown by name', async () => {
+    const noraNumber = '+19492847328'
+    const { call, event, svc } = setup({
+      agentCallerIds: {
+        riotcoder: '+14159185508', cha0spunk: '+15107667447', coinpay: '+15623079115',
+        knox: '+18313290328', rootkey: '+14106398993', nora: noraNumber,
+      },
+    })
+    const room = (await (await call('gc_op', 'POST', '/api/v1/rooms', { name: 'Agents' })).json()) as any
+    expect((await call('gc_op', 'POST', `/api/v1/rooms/${room.code}/agents`, { agent: 'nora' })).status).toBe(202)
+    const next = svc.nextInvite()!
+    expect(next).toMatchObject({ agent: 'nora', from: noraNumber, status: 'dialling' })
+    await event('call.initiated', { call_control_id: 'leg-nora', direction: 'incoming', from: noraNumber })
+    await event('call.gather.ended', { call_control_id: 'leg-nora', digits: room.code })
+    expect(svc.live(room.code)).toEqual([{ name: 'nora', kind: 'agent', since: expect.any(Number) }])
+  })
+
+  test('six simultaneous agent arrivals all join the same conference', async () => {
+    const agentCallerIds = {
+      riotcoder: '+14159185508', cha0spunk: '+15107667447', coinpay: '+15623079115',
+      knox: '+18313290328', rootkey: '+14106398993', nora: '+19492847328',
+    }
+    const { call, event, svc, telnyx } = setup({ agentCallerIds })
+    const room = (await (await call('gc_op', 'POST', '/api/v1/rooms', { name: 'Team call' })).json()) as any
+    for (const agent of Object.keys(agentCallerIds)) expect((await call('gc_op', 'POST', `/api/v1/rooms/${room.code}/agents`, { agent })).status).toBe(202)
+    for (const [agent, from] of Object.entries(agentCallerIds)) expect(svc.nextInvite()).toMatchObject({ agent, from })
+
+    await Promise.all(Object.entries(agentCallerIds).map(async ([agent, from]) => {
+      const leg = `leg-${agent}`
+      await event('call.initiated', { call_control_id: leg, direction: 'incoming', from })
+      await event('call.gather.ended', { call_control_id: leg, digits: room.code })
+    }))
+
+    expect(telnyx.calls.filter((c) => c.path === '/conferences')).toHaveLength(1)
+    expect(telnyx.calls.filter((c) => c.path === '/conferences/conf-1/actions/join')).toHaveLength(5)
+    expect(svc.live(room.code).map((p) => p.name).sort()).toEqual(Object.keys(agentCallerIds).sort())
+  })
+
   test("your own agent can be named before it dials in from its own number", async () => {
     const { call, event, svc } = setup()
     const room = (await (await call('gc_guest', 'POST', '/api/v1/rooms', { name: 'Mine' })).json()) as any
@@ -785,6 +829,22 @@ describe('recording a room', () => {
     const guestRoom = (await (await call('gc_guest', 'POST', '/api/v1/rooms', { name: 'Mine' })).json()) as any
     expect((await call('gc_guest', 'POST', `/api/v1/rooms/${guestRoom.code}/call`, { sip: 'sip:+15106169327@sip.rtc.elevenlabs.io', name: 'x' })).status).toBe(403)
   })
+
+  test('a SIP agent keeps its name after a restart even if the old live-leg row lost it', async () => {
+    const noraSip = 'sip:+16505550147@sip.rtc.elevenlabs.io;transport=tcp'
+    const { call, event, telnyx, svc, store, config } = setup({ agentSip: { nora: noraSip } })
+    const room = (await (await call('gc_op', 'POST', '/api/v1/rooms', { name: 'Standup' })).json()) as any
+    expect((await call('gc_op', 'POST', `/api/v1/rooms/${room.code}/agents`, { agent: 'nora' })).status).toBe(202)
+    await event('call.answered', { call_control_id: 'out-1' })
+    expect(svc.live(room.code)).toMatchObject([{ name: 'nora', kind: 'agent' }])
+    expect(store.dialled('out-1')).toMatchObject({ provider: 'sip', name: 'nora', to: noraSip })
+
+    telnyx.participants.add('out-1')
+    store.db.query('UPDATE live_legs SET name = ? WHERE leg = ?').run('', 'out-1')
+    const restarted = createApp({ store, origin: 'https://group.call.test', config })
+    expect(await restarted.svc.restore()).toEqual({ rooms: 1, legs: 1 })
+    expect(restarted.svc.live(room.code)).toEqual([{ name: 'nora', kind: 'agent', since: expect.any(Number) }])
+  })
 })
 
 describe('the TUI', () => {