diff --git a/ops/agent-phone.mjs b/ops/agent-phone.mjs index 2507fb1..7352702 100755 --- a/ops/agent-phone.mjs +++ b/ops/agent-phone.mjs @@ -26,7 +26,7 @@ */ import { execFile, spawn } from 'node:child_process'; import { pathToFileURL } from 'node:url'; -import { appendFileSync, mkdirSync, readFileSync, statSync, utimesSync, writeFileSync } from 'node:fs'; +import { appendFileSync, mkdirSync, readFileSync, renameSync, rmSync, statSync, utimesSync, writeFileSync } from 'node:fs'; const HOME = process.env.AGENT_PHONE_HOME || process.env.HOME || '/home/anthony'; const read = (p) => readFileSync(p, 'utf8'); @@ -64,9 +64,10 @@ export function shouldIntroduce(agent, transcript, now = Date.now(), recentMs = export function roomQuestion(text) { const body = String(text).replace(/(^|\s)[^:\s]*( ending \d+)?:\s*/g, ' ').trim(); const isGreeting = /^(hello|hi|hey)[\s,.!]*$/i.test(body); - const soundsLikeQuestion = /[?]\s*$/.test(body) || /^(who|what|can|could|are|is|anyone|anybody)\b/i.test(body) || isGreeting; - if (!soundsLikeQuestion) return false; - return /\b(any ?one|any ?body|every ?one|everybody|you guys|you all|y'?all|agents?|bots?|can you hear|are you there|is anyone|hello|still there|who(?:['’]s| is| are)? (?:here|there|you|on the line)|what(?:['’]s| is) happening)\b/i.test(body); + return /[?]\s*$/.test(body) + || /^(who|what|when|where|why|which|how|can|could|are|is|anyone|anybody|do|does|did|will|would|should|have|has)\b/i.test(body) + || /^(?:you(?: all| guys)?|everybody|everyone)\s+(?:hear|listen|understand|there|on the line)\b/i.test(body) + || isGreeting; } const QUIET_MS = 1000; // speak once they have paused this long (1.5 s on a group call: more voices, longer gaps) const TURN_CWD = `${HOME}/.local/share/agent-phone/turn-cwd`; @@ -124,8 +125,9 @@ export const addressed = (agent, text) => { * speaks. A line meant for nobody in particular gets one agent answering, not two. */ const floorFile = (pin) => `${CALL_LOGS}/floor-${pin}.json`; +const floorLockDir = (pin) => `${floorFile(pin)}.lock`; /** Who holds the floor in this room right now (an agent mid-sentence), or ''. */ -function floorHolder(pin) { +export function floorHolder(pin) { try { const f = JSON.parse(readFileSync(floorFile(pin), 'utf8')); return f.until > Date.now() ? String(f.agent) : ''; @@ -137,11 +139,44 @@ const floorTakenByOther = (pin, agent) => { const h = floorHolder(pin); return h !== '' && h !== agent; }; -/** Hold the floor for about as long as the line takes to say (ElevenLabs reads ~14 chars a second). */ -function takeFloor(pin, agent, text = '') { + +/** Run a short floor-file update under an atomic directory lock shared by call workers. */ +function withFloorLock(pin, update) { + const lock = floorLockDir(pin); + let locked = false; try { - writeFileSync(floorFile(pin), JSON.stringify({ agent, at: Date.now(), until: Date.now() + 2500 + (String(text).length / 14) * 1000 })); - } catch {} + mkdirSync(lock); + locked = true; + } catch (e) { + if (e?.code !== 'EEXIST') return false; + try { + if (Date.now() - statSync(lock).mtimeMs > 10_000) { + rmSync(lock, { recursive: true, force: true }); + mkdirSync(lock); + locked = true; + } + } catch {} + } + if (!locked) return false; + try { + return update(); + } finally { + rmSync(lock, { recursive: true, force: true }); + } +} + +/** Hold the floor for about as long as the line takes to say (roughly 14 chars a second). */ +export function takeFloor(pin, agent, text = '') { + try { + return withFloorLock(pin, () => { + if (floorTakenByOther(pin, agent)) return false; + const file = floorFile(pin); + const temp = `${file}.${process.pid}.${Date.now()}.tmp`; + writeFileSync(temp, JSON.stringify({ agent, at: Date.now(), until: Date.now() + 2500 + (String(text).length / 14) * 1000 })); + renameSync(temp, file); + return true; + }); + } catch { return false; } } /** The other agent speaking now or within `afterMs` of finishing: its words, not a person's. */ function otherAgentTalking(pin, agent, afterMs = 1500) { @@ -173,12 +208,17 @@ function floorHolderRecently(pin, agent, withinMs = 6000) { return false; } } -function releaseFloor(pin, agent) { +export function releaseFloor(pin, agent) { try { - if (floorHolder(pin) === agent) { + withFloorLock(pin, () => { + if (floorHolder(pin) !== agent) return false; const f = JSON.parse(readFileSync(floorFile(pin), 'utf8')); - writeFileSync(floorFile(pin), JSON.stringify({ ...f, until: 0 })); - } + const file = floorFile(pin); + const temp = `${file}.${process.pid}.${Date.now()}.tmp`; + writeFileSync(temp, JSON.stringify({ ...f, until: 0 })); + renameSync(temp, file); + return true; + }); } catch {} } @@ -639,7 +679,7 @@ async function main() { if (floorTakenByOther(a.pin, agent)) return; // still theirs after 8 s: let it go // The other agent took the floor and answered while we waited: unless this was ours by name, leave it. if (introduced && !mine && kind !== 'hangup' && Date.now() - t0 > 2500 && floorHolderRecently(a.pin, agent)) return; - takeFloor(a.pin, agent, text); + if (!takeFloor(a.pin, agent, text)) return; } if (kind === 'task') { // Work only for someone this agent works for: the room's own transcript says who spoke. diff --git a/src/service.ts b/src/service.ts index f6db18a..cb927fc 100644 --- a/src/service.ts +++ b/src/service.ts @@ -1170,6 +1170,17 @@ export class GroupCallService { return } const room = this.store.room(code) + if (room && shown.kind === 'person') { + // An invitee leaves after three minutes if nobody has arrived. Bring that + // team back when the host joins, rather than leaving stale names behind. + for (const previous of this.store.invites(code, 50)) { + if (previous.status !== 'done' || !/\bnobody came\b/i.test(previous.note) || Date.now() - previous.createdAt > 15 * 60_000) continue + if (AGENTS[previous.agent]) { + this.store.finishInvite(previous.id, 'done', 'requeued after host arrival') + this.queueAgent(room, previous.agent, previous.sub) + } + } + } const greeter = room?.greeter if (!room || !greeter || !this.config.agentSip?.[greeter] || !this.hostHere(code)) return // It hears the host arrive and hands over itself (its prompt says so); if it has not left in 45 s, it is shown out. diff --git a/test/agent-phone.test.mjs b/test/agent-phone.test.mjs index 83097d1..5735e1a 100644 --- a/test/agent-phone.test.mjs +++ b/test/agent-phone.test.mjs @@ -58,9 +58,24 @@ test('a recent line from the same agent suppresses a duplicate intro on a fast r 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: What was that?'), true); + assert.equal(worker.roomQuestion('Anthony: What was that'), true); + assert.equal(worker.roomQuestion('Anthony: How does this work?'), true); assert.equal(worker.roomQuestion('Anthony: I told everyone to listen'), false); }); +test('only one agent can claim the room floor until it releases it', () => { + const pin = `floor-test-${process.pid}`; + assert.equal(worker.takeFloor(pin, 'riotcoder', 'I will answer.'), true); + assert.equal(worker.floorHolder(pin), 'riotcoder'); + assert.equal(worker.takeFloor(pin, 'nora', 'I will also answer.'), false); + worker.releaseFloor(pin, 'riotcoder'); + assert.equal(worker.takeFloor(pin, 'nora', 'I will answer now.'), true); + assert.equal(worker.floorHolder(pin), 'nora'); + worker.releaseFloor(pin, 'nora'); + assert.equal(worker.floorHolder(pin), ''); +}); + 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); diff --git a/test/app.test.ts b/test/app.test.ts index 96ce65b..81b07f0 100644 --- a/test/app.test.ts +++ b/test/app.test.ts @@ -472,6 +472,20 @@ test('an invite stuck "dialling" from an agent that never arrived does not block expect(store.invite(first.id)?.status).toBe('failed') }) +test('agents who timed out before the host arrived are dialled back into the room', async () => { + const { call, event, store, telnyx } = setup({ agentSip: { riotcoder: 'sip:+15106169327@sip.rtc.elevenlabs.io' } }) + const room = (await (await call('gc_op', 'POST', '/api/v1/rooms', { name: 'Late host' })).json()) as any + const expired = store.addInvite({ code: room.code, agent: 'riotcoder', sub: 'sub-op' }) + store.finishInvite(expired.id, 'done', 'on the call 182 s, nobody came') + + await event('call.initiated', { call_control_id: 'late-host', direction: 'incoming', from: '+15550001111' }) + await event('call.gather.ended', { call_control_id: 'late-host', digits: room.code }) + + expect(store.invite(expired.id)?.note).toBe('requeued after host arrival') + expect(store.invites(room.code, 10).some((invite) => invite.id !== expired.id && invite.agent === 'riotcoder')).toBe(true) + expect(telnyx.calls.some((c) => c.path === '/calls' && c.body.to.includes('sip.rtc.elevenlabs.io'))).toBe(true) +}) + test("a known caller is named, and agents can see which recent lines came from someone trusted", async () => { const keys = telnyxKeys() const telnyx = telnyxRecorder()