Stop the voice test tripping answering-machine detection

tests/live/test_voice.py

@@ -50,6 +50,12 @@ def _call_me_text() -> str:
 TIMEOUT_S = float(os.environ.get("LIVE_VOICE_TIMEOUT", "220"))
 POLL_EVERY_S = 6.0
 TERMINAL_FAILURE_STATUSES = {"canceled", "failed"}
+# A call can end normally and still never carry a conversation - answering-machine
+# detection hanging up on the driver ends it `completed`, hangup_reason=voicemail.
+# Transcript rows can still land during teardown, so allow a short grace period
+# before giving up rather than polling a finished call for the full timeout.
+ENDED_STATUSES = {"completed"}
+ENDED_GRACE_S = float(os.environ.get("LIVE_VOICE_ENDED_GRACE", "15"))
 
 pytestmark = pytest.mark.skipif(
     not (REMOTE_KEY and AUT_KEY and REAL),
@@ -106,6 +112,7 @@ def _wait_for_two_way_call(remote, number_id, call_id):
     """Block until the call transcript shows BOTH the agent and the driver spoke."""
     deadline = time.monotonic() + TIMEOUT_S
     last = ""
+    ended_at = None
     while time.monotonic() < deadline:
         transcript_state = ""
         try:
@@ -125,6 +132,11 @@ def _wait_for_two_way_call(remote, number_id, call_id):
         last = f"{progress}; {state}"
         if status in TERMINAL_FAILURE_STATUSES:
             pytest.fail(f"call ended before a two-way conversation ({last})")
+        if status in ENDED_STATUSES:
+            if ended_at is None:
+                ended_at = time.monotonic()
+            elif time.monotonic() - ended_at > ENDED_GRACE_S:
+                pytest.fail(f"call ended without a two-way conversation ({last})")
         time.sleep(POLL_EVERY_S)
     pytest.fail(f"agent never held a two-way call within {TIMEOUT_S:.0f}s ({last})")
 

tests/live/voice_driver.py

@@ -47,8 +47,12 @@
     "VOICE_DRIVER_LINE",
     "Hi, this is a quick test call. Please reply out loud with one short sentence, then say goodbye.",
 )
-# Speak shortly after the pipeline is ready so the agent's greeting lands first.
-SPEAK_AFTER_S = float(os.environ.get("VOICE_DRIVER_SPEAK_AFTER", "3"))
+# Answering-machine detection scores whoever answers: a greeting longer than the
+# carrier's `greeting_duration_millis` (3.5s) reads as a voicemail announcement
+# and the call is hung up before the agent ever speaks. Answer the way a person
+# does - one word, then silence - and hold the prompt until that window closes.
+GREETING = os.environ.get("VOICE_DRIVER_GREETING", "Hello?")
+SPEAK_AFTER_S = float(os.environ.get("VOICE_DRIVER_SPEAK_AFTER", "5"))
 # Then give the agent a turn and hang up — a dropped WS does NOT end the call, so we
 # must send an explicit stop or the leg lingers until the server max-duration cap.
 LISTEN_S = float(os.environ.get("VOICE_DRIVER_LISTEN", "12"))
@@ -75,16 +79,20 @@ async def phone_media_ws(ws: WebSocket) -> None:
     spoke = asyncio.Event()
     convo: asyncio.Task | None = None
 
+    async def _say(text: str) -> None:
+        await ws.send_text(json.dumps({"event": "text", "delta": text}))
+        await ws.send_text(json.dumps({"event": "text", "done": True}))
+        log.info("spoke: %s", text)
+
     async def _speak(text: str) -> None:
         if spoke.is_set():
             return
         spoke.set()
-        await ws.send_text(json.dumps({"event": "text", "delta": text}))
-        await ws.send_text(json.dumps({"event": "text", "done": True}))
-        log.info("spoke: %s", text)
+        await _say(text)
 
     async def _run_turn() -> None:
         # Speak one line, give the agent a turn, then hang up so the call ends fast.
+        await _say(GREETING)
         await asyncio.sleep(SPEAK_AFTER_S)
         await _speak(LINE)
         await asyncio.sleep(LISTEN_S)