Fix room agent turn taking and late joins

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.

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.

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);

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()