From 165fde453eb57bcc4600a0ea7357cbacf1cd5a86 Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Mon, 19 Oct 2026 11:26:00 +0500 Subject: [PATCH 1/4] feat: The incoming calls and the ai greeting text have been supported in the chatbot --- README.md | 49 +- docs/README.md | 50 +- setup.py | 2 +- tests/calls/test_call_runtime.py | 8 + tests/calls/test_call_service.py | 506 ++++++++++++++- tests/calls/test_call_state_machine.py | 46 ++ tests/calls/test_realtime_voice.py | 28 + tests/test_calls_imports.py | 4 +- whatsapp_chatbot_python/calls/__init__.py | 10 +- whatsapp_chatbot_python/calls/contracts.py | 8 + whatsapp_chatbot_python/calls/models.py | 34 + .../calls/realtime_voice.py | 17 +- whatsapp_chatbot_python/calls/runtime.py | 9 +- whatsapp_chatbot_python/calls/service.py | 590 +++++++++++++++++- .../calls/state_machine.py | 34 +- 15 files changed, 1337 insertions(+), 58 deletions(-) diff --git a/README.md b/README.md index 41113ed..24e59cd 100644 --- a/README.md +++ b/README.md @@ -541,20 +541,25 @@ credentials: ```python import asyncio import logging -import os from whatsapp_chatbot_python.calls import ( CallEvent, CallSession, CallStateMachine, ) from whatsapp_chatbot_python.calls.service import WhatsAppCallService +id_instance = "ID_INSTANCE" +api_token_instance = "API_TOKEN" +openai_api_key = "OPENAI_KEY" +realtime_model = "gpt-realtime-2.1" +realtime_voice = "marin" + service = WhatsAppCallService( api_url="https://api.green-api.com", - id_instance=os.environ["ID_INSTANCE"], - api_token_instance=os.environ["API_TOKEN_INSTANCE"], - openai_api_key=os.environ["OPENAI_API_KEY"], - realtime_model=os.environ["REALTIME_MODEL"], - realtime_voice=os.environ["REALTIME_VOICE"], + id_instance=id_instance, + api_token_instance=api_token_instance, + openai_api_key=openai_api_key, + realtime_model=realtime_model, + realtime_voice=realtime_voice, ring_timeout_seconds=30, talk_timeout_seconds=120, shutdown_timeout_seconds=10, @@ -583,6 +588,38 @@ async def call_once(): asyncio.run(call_once()) ``` +### Accept an incoming call + +The application owns its queue, offer deduplication and notification parsing. An incoming offer can share the same queue +and worker as outgoing calls. Create its deadline when the offer arrives, using the time remaining from the offer +timestamp. For example, a 60-second application limit becomes `monotonic() + remaining_seconds`; the library has no +fixed incoming ring limit. The deadline includes time spent in the queue and in call setup. Coordinators can type their +executor as `IncomingCallExecutor`, which extends the existing `CallExecutor` with the two incoming call-control +methods. + +```python +from time import monotonic +from whatsapp_chatbot_python.calls import CallEvent, CallSession, CallStateMachine + +session = CallSession.incoming( + sender_id=caller_id, + call_id=call_id, + language="en", + incoming_deadline=monotonic() + remaining_seconds, + greeting_text="Hello! This is a WhatsApp demo call. What would you like to discuss?", +) +fsm = CallStateMachine() +fsm.apply(session, CallEvent.ENQUEUED) +app_queue.put(session) + +if monotonic() >= session.incoming_deadline: + fsm.apply(session, CallEvent.INCOMING_EXPIRED) + await service.reject_incoming(session.call_id, session.sender_id) +else: + fsm.apply(session, CallEvent.DEQUEUED) + result = await service.execute(session, transition_under_app_lock) +``` + ## Service methods documentation [Service methods documentation](https://green-api.com/en/docs/api/) diff --git a/docs/README.md b/docs/README.md index 139425c..ad7d44e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -533,20 +533,25 @@ python -m pip install 'whatsapp-chatbot-python[voip]' ```python import asyncio import logging -import os from whatsapp_chatbot_python.calls import ( CallEvent, CallSession, CallStateMachine, ) from whatsapp_chatbot_python.calls.service import WhatsAppCallService +id_instance = "ID_INSTANCE" +api_token_instance = "API_TOKEN" +openai_api_key = "OPENAI_KEY" +realtime_model = "gpt-realtime-2.1" +realtime_voice = "marin" + service = WhatsAppCallService( api_url="https://api.green-api.com", - id_instance=os.environ["ID_INSTANCE"], - api_token_instance=os.environ["API_TOKEN_INSTANCE"], - openai_api_key=os.environ["OPENAI_API_KEY"], - realtime_model=os.environ["REALTIME_MODEL"], - realtime_voice=os.environ["REALTIME_VOICE"], + id_instance=id_instance, + api_token_instance=api_token_instance, + openai_api_key=openai_api_key, + realtime_model=realtime_model, + realtime_voice=realtime_voice, ring_timeout_seconds=30, talk_timeout_seconds=120, shutdown_timeout_seconds=10, @@ -567,7 +572,6 @@ async def call_once(): print(session.state, session.end_reason, session.error_code) if result.recording_path is not None: print("MP3:", result.recording_path) - # Upload or copy the recording here, before removing it. finally: if result.recording_path is not None: result.recording_path.unlink(missing_ok=True) @@ -575,6 +579,38 @@ async def call_once(): asyncio.run(call_once()) ``` +### Принять входящий звонок + +Очередь, дедупликация предложений звонка и разбор уведомлений принадлежат приложению. Входящий звонок может использовать +того же работника очереди, что и исходящий. При получении `offer` приложение вычисляет оставшееся время по времени +уведомления и передаёт абсолютный срок `monotonic() + remaining_seconds`. Например, лимит 60 секунд задаёт приложение; +библиотека не содержит фиксированного лимита входящего дозвона. Срок включает ожидание в очереди и подготовку вызова. +Координатор может использовать тип `IncomingCallExecutor`: он расширяет существующий `CallExecutor` двумя методами +управления входящим вызовом. + +```python +from time import monotonic +from whatsapp_chatbot_python.calls import CallEvent, CallSession, CallStateMachine + +session = CallSession.incoming( + sender_id=caller_id, + call_id=call_id, + language="ru", + incoming_deadline=monotonic() + remaining_seconds, + greeting_text="Здравствуйте! Это демонстрационный звонок WhatsApp. Что хотите обсудить?", +) +fsm = CallStateMachine() +fsm.apply(session, CallEvent.ENQUEUED) +app_queue.put(session) + +if monotonic() >= session.incoming_deadline: + fsm.apply(session, CallEvent.INCOMING_EXPIRED) + await service.reject_incoming(session.call_id, session.sender_id) +else: + fsm.apply(session, CallEvent.DEQUEUED) + result = await service.execute(session, transition_under_app_lock) +``` + ## Документация по методам сервиса [Документация по методам сервиса](https://green-api.com/docs/api/) diff --git a/setup.py b/setup.py index 057f4e5..6bb9242 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="whatsapp-chatbot-python", - version="1.0.0", + version="1.1.0", description=( "This library helps you easily create" " a Python chatbot with WhatsApp API." diff --git a/tests/calls/test_call_runtime.py b/tests/calls/test_call_runtime.py index d95baec..0fd4606 100644 --- a/tests/calls/test_call_runtime.py +++ b/tests/calls/test_call_runtime.py @@ -2,6 +2,7 @@ from whatsapp_chatbot_python.calls.runtime import CallRuntime, RuntimeEvent, RuntimeEventType import asyncio import unittest +from time import monotonic class CallRuntimeTest(unittest.IsolatedAsyncioTestCase): @@ -25,6 +26,13 @@ async def test_empty_queue_returns_none_when_timer_expires(self) -> None: runtime.start_ringing() self.assertIsNone(await runtime.next_event(CallState.RINGING)) + async def test_incoming_deadline_is_absolute(self) -> None: + runtime = CallRuntime(30, 1, 1) + deadline = monotonic() + 0.01 + runtime.start_ringing(deadline=deadline) + self.assertEqual(runtime.ring_deadline, deadline) + self.assertIsNone(await runtime.next_event(CallState.RINGING)) + async def test_cancellation_is_not_swallowed_when_event_becomes_ready(self) -> None: waiting = asyncio.Event() diff --git a/tests/calls/test_call_service.py b/tests/calls/test_call_service.py index 79a16f0..4e683f6 100644 --- a/tests/calls/test_call_service.py +++ b/tests/calls/test_call_service.py @@ -1,11 +1,13 @@ from pathlib import Path from tempfile import NamedTemporaryFile from threading import Thread -from unittest.mock import patch +from unittest.mock import AsyncMock, Mock, patch +from time import monotonic import asyncio import logging import unittest +from whatsapp_api_client_python.tools.voip import CallInfo from whatsapp_chatbot_python.calls.models import CallEndReason, CallEvent, CallSession, CallState from whatsapp_chatbot_python.calls.service import WhatsAppCallService from whatsapp_chatbot_python.calls.state_machine import CallStateMachine @@ -81,11 +83,14 @@ def __init__(self, client): self.closed = False self.audio_factory = None self.devices = [] + self.state = None def on(self, name, callback): self.listeners.setdefault(name, []).append(callback) def emit(self, name, detail=None): + if name == "state": + self.state = detail for callback in self.listeners.get(name, []): callback(detail) @@ -96,7 +101,7 @@ async def openAsync(self, *, timeout): raise RuntimeError("callsRtc connection refused") asyncio.get_running_loop().call_soon( - self.emit, "state", CallStateDetail(self.client.initial_state) + self.emit, "state", CallStateDetail(self.client.initial_state, info=self.client.initial_info) ) async def startAudioAsync(self): @@ -107,12 +112,12 @@ async def startAudioAsync(self): raise RuntimeError("audio bridge failed") if self.client.answer and not self.client.answer_after_bridge: - self.emit("state", CallStateDetail("on-call")) + self.emit("state", CallStateDetail("on-call", info=self.client.initial_info)) if self.client.answer and self.client.answer_after_bridge: asyncio.get_running_loop().call_later( 0.001, self.emit, "state", - CallStateDetail("on-call"), + CallStateDetail("on-call", info=self.client.initial_info), ) if self.client.reconnect: @@ -137,13 +142,15 @@ async def closeAsync(self): class CallStateDetail: - def __init__(self, state, reason=None): + def __init__(self, state, reason=None, info=None): self.state = state self.reason = reason + self.info = info class FakeClient: initial_state = "idle" + initial_info = None answer = True remote_hangup = True bridge_error = False @@ -157,8 +164,13 @@ def __init__(self, id_instance, api_token_instance, *, host): self.voip = self self.calls = FakeCalls(self) self.actions = [] + self.credentials = (id_instance, api_token_instance, host) + self.session = self self.instances.append(self) + def close(self): + pass + def connect(self, *, audio_factory): self.calls.audio_factory = audio_factory return self.calls @@ -166,6 +178,16 @@ def connect(self, *, audio_factory): async def dialAsync(self, target): self.actions.append(("dial", target)) + async def acceptAsync(self): + self.actions.append("accept") + self.calls.emit("state", CallStateDetail("on-call", info=self.initial_info)) + + async def rejectAsync(self): + self.actions.append("reject") + + async def getStateAsync(self): + return self.calls.state + async def hangUpAsync(self): self.actions.append("hangup") @@ -174,6 +196,7 @@ class CallServiceTest(unittest.IsolatedAsyncioTestCase): def setUp(self): for name, value in { "initial_state": "idle", + "initial_info": None, "answer": True, "remote_hangup": True, "bridge_error": False, @@ -411,6 +434,479 @@ def finish(self): self.assertIsNone(result.recording_path) self.assertFalse(path.exists()) + def make_incoming_session(self, *, seconds=60, call_id="call-1"): + fsm = CallStateMachine() + session = CallSession.incoming( + sender_id="sender@c.us", call_id=call_id, language="ru", + incoming_deadline=monotonic() + seconds, + greeting_text="Здравствуйте! Что хотите обсудить?", + ) + fsm.apply(session, CallEvent.ENQUEUED) + fsm.apply(session, CallEvent.DEQUEUED) + return session, fsm.apply + + async def test_incoming_accepts_matching_call_before_audio(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + await self.make_service().execute(session, transition) + + self.assertEqual(FakeClient.instances[0].actions[:3], ["open", "accept", "bridge"]) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(session.end_reason, CallEndReason.REMOTE_HANGUP) + self.assertTrue(session.incoming_accept_completed) + self.assertIsNotNone(session.answered_at) + self.assertEqual(FakeVoice.instances[0].options["greeting_text"], session.greeting_text) + self.assertEqual(FakeVoice.instances[0].greetings, 1) + + async def test_voice_service_uses_explicit_credentials_despite_environment(self): + import os + + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + with patch.dict(os.environ, { + "ID_INSTANCE": "ambient-instance", + "API_TOKEN_INSTANCE": "ambient-token", + "OPENAI_API_KEY": "ambient-openai-key", + }): + await self.make_service().execute(session, transition) + + self.assertEqual( + FakeClient.instances[0].credentials, + ("1", "token", "https://example.invalid"), + ) + self.assertEqual(FakeVoice.instances[0].options["api_key"], "test-key") + + async def test_incoming_other_call_is_never_accepted_or_ended(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("new-call", "sender@c.us", None) + session, transition = self.make_incoming_session() + + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + self.assertEqual(FakeClient.instances[0].actions, ["open"]) + + async def test_incoming_ended_before_execute_uses_no_resources(self): + session, transition = self.make_incoming_session() + transition(session, CallEvent.CALLER_LEFT) + + result = await self.make_service().execute(session, transition) + + self.assertIsNone(result.recording_path) + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(FakeClient.instances, []) + self.assertEqual(FakeVoice.instances, []) + + async def test_incoming_deadline_cancels_pending_accept(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session(seconds=0.03) + + async def slow_accept(client): + client.actions.append("accept") + await asyncio.Event().wait() + + with patch.object(FakeClient, "acceptAsync", slow_accept): + await asyncio.wait_for(self.make_service().execute(session, transition), timeout=1) + + self.assertEqual(session.state, CallState.RING_TIMEOUT) + self.assertEqual(session.end_reason, CallEndReason.INCOMING_EXPIRED) + self.assertEqual(FakeClient.instances[0].actions, ["open", "accept", "reject"]) + + async def test_matching_on_call_before_deadline_wins_over_pending_accept(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session(seconds=0.2) + + async def accepted_without_rest_response(client): + client.actions.append("accept") + client.calls.emit("state", CallStateDetail("on-call", info=client.initial_info)) + await asyncio.Event().wait() + + async def bridge_after_ring_deadline(calls): + calls.client.actions.append("bridge") + await asyncio.sleep(0.25) + calls.emit("state", CallStateDetail("idle", "hangup")) + + with patch.object(FakeClient, "acceptAsync", accepted_without_rest_response), \ + patch.object(FakeCalls, "startAudioAsync", bridge_after_ring_deadline): + await asyncio.wait_for(self.make_service().execute(session, transition), timeout=1) + + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(session.end_reason, CallEndReason.REMOTE_HANGUP) + self.assertIsNotNone(session.answered_at) + self.assertFalse(session.incoming_accept_completed) + self.assertEqual(FakeClient.instances[0].actions[:3], ["open", "accept", "bridge"]) + + async def test_on_call_after_caller_left_does_not_reclassify_unanswered_offer(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + service = self.make_service() + accept_started = asyncio.Event() + + async def late_answer(client): + client.actions.append("accept") + accept_started.set() + await asyncio.Event().wait() + + with patch.object(FakeClient, "acceptAsync", late_answer): + task = asyncio.create_task(service.execute(session, transition)) + await asyncio.wait_for(accept_started.wait(), timeout=1) + self.assertTrue(service.notify_incoming_ended("call-1")) + await asyncio.sleep(0) + FakeClient.instances[0].calls.emit( + "state", CallStateDetail("on-call", info=FakeClient.initial_info), + ) + await asyncio.wait_for(task, timeout=1) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + + async def test_incoming_ended_during_execute_handoff_uses_no_resources(self): + import whatsapp_chatbot_python.calls.service as call_service + + session, transition = self.make_incoming_session() + original_execution = call_service._CallExecution + service = self.make_service() + + def end_during_handoff(*args): + execution = original_execution(*args) + self.assertFalse(service.notify_incoming_ended("call-1")) + transition(session, CallEvent.CALLER_LEFT) + return execution + + with patch.object(call_service, "_CallExecution", side_effect=end_during_handoff): + await service.execute(session, transition) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(FakeClient.instances, []) + self.assertEqual(FakeVoice.instances, []) + + async def test_accept_error_after_caller_left_is_not_internal_failure(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def failed_accept(client): + client.actions.append("accept") + client.calls.emit("state", CallStateDetail("idle")) + raise RuntimeError("offer gone") + + with patch.object(FakeClient, "acceptAsync", failed_accept): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + self.assertNotIn("bridge", FakeClient.instances[0].actions) + + async def test_accept_error_after_new_call_does_not_touch_new_call(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def failed_accept(client): + client.actions.append("accept") + client.calls.state = CallStateDetail( + "inc-call", info=CallInfo("call-2", "sender@c.us", None), + ) + raise RuntimeError("offer replaced") + + with patch.object(FakeClient, "acceptAsync", failed_accept): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + self.assertEqual(FakeClient.instances[0].actions, ["open", "accept"]) + + async def test_accept_error_while_offer_is_live_remains_failure(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def failed_accept(client): + client.actions.append("accept") + raise RuntimeError("accept unavailable") + + with patch.object(FakeClient, "acceptAsync", failed_accept): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.end_reason, CallEndReason.ERROR) + self.assertEqual(FakeClient.instances[0].actions, ["open", "accept", "reject"]) + + async def test_accept_error_with_unavailable_state_remains_failure(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def failed_accept(client): + client.actions.append("accept") + raise RuntimeError("response lost") + + async def unavailable_state(client): + raise RuntimeError("state unavailable") + + with patch.object(FakeClient, "acceptAsync", failed_accept), \ + patch.object(FakeClient, "getStateAsync", unavailable_state): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.end_reason, CallEndReason.ERROR) + + async def test_accept_error_with_matching_on_call_preserves_answer(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def failed_after_answer(client): + client.actions.append("accept") + client.calls.emit("state", CallStateDetail("on-call", info=client.initial_info)) + raise RuntimeError("response lost") + + with patch.object(FakeClient, "acceptAsync", failed_after_answer): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertIsNotNone(session.answered_at) + self.assertNotEqual(session.end_reason, CallEndReason.ERROR) + + async def test_accept_error_with_server_confirmed_on_call_preserves_answer(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def failed_accept(client): + client.actions.append("accept") + raise RuntimeError("response lost") + + async def server_state(client): + return CallStateDetail("on-call", info=client.initial_info) + + with patch.object(FakeClient, "acceptAsync", failed_accept), \ + patch.object(FakeClient, "getStateAsync", server_state): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertIsNotNone(session.answered_at) + + async def test_observed_answer_is_applied_before_external_caller_left(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def answer_then_leave(client): + client.actions.append("accept") + client.calls.emit("state", CallStateDetail("on-call", info=client.initial_info)) + self.assertEqual(session.state, CallState.CONNECTING) + transition(session, CallEvent.CALLER_LEFT) + raise RuntimeError("response lost") + + with patch.object(FakeClient, "acceptAsync", answer_then_leave): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(session.end_reason, CallEndReason.REMOTE_HANGUP) + + async def test_incoming_deadline_cancels_voice_preparation(self): + session, transition = self.make_incoming_session(seconds=0.03) + service = self.make_service() + + async def slow_start(voice): + await asyncio.Event().wait() + + with patch.object(FakeVoice, "start", slow_start), \ + patch.object(service, "reject_incoming", new_callable=AsyncMock) as reject: + await asyncio.wait_for(service.execute(session, transition), timeout=1) + + self.assertEqual(session.state, CallState.RING_TIMEOUT) + self.assertEqual(session.end_reason, CallEndReason.INCOMING_EXPIRED) + reject.assert_awaited_once_with("call-1", "sender@c.us") + self.assertTrue(FakeVoice.instances[0].closed) + self.assertEqual(FakeClient.instances, []) + + async def test_incoming_caller_leaving_interrupts_voice_preparation(self): + session, transition = self.make_incoming_session() + service = self.make_service() + preparing = asyncio.Event() + + async def slow_start(voice): + preparing.set() + await asyncio.Event().wait() + + with patch.object(FakeVoice, "start", slow_start): + task = asyncio.create_task(service.execute(session, transition)) + try: + await asyncio.wait_for(preparing.wait(), timeout=1) + self.assertTrue(service.notify_incoming_ended("call-1")) + await asyncio.wait_for(task, timeout=1) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + self.assertEqual(FakeClient.instances, []) + + async def test_incoming_deadline_cancels_calls_connection_open(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session(seconds=0.03) + + async def slow_open(calls, *, timeout): + calls.client.actions.append("open") + await asyncio.Event().wait() + + with patch.object(FakeCalls, "openAsync", slow_open): + await asyncio.wait_for(self.make_service().execute(session, transition), timeout=1) + + self.assertEqual(session.state, CallState.RING_TIMEOUT) + self.assertEqual(session.end_reason, CallEndReason.INCOMING_EXPIRED) + self.assertEqual(FakeClient.instances[0].actions, ["open"]) + + async def test_incoming_caller_left_signal_wakes_active_call(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + FakeClient.remote_hangup = False + session, transition = self.make_incoming_session() + service = self.make_service() + task = asyncio.create_task(service.execute(session, transition)) + + try: + async with asyncio.timeout(1): + while session.state != CallState.IN_CALL: + await asyncio.sleep(0) + self.assertFalse(service.notify_incoming_ended("older-call")) + self.assertTrue(service.notify_incoming_ended("call-1")) + await asyncio.wait_for(task, timeout=1) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertEqual(session.end_reason, CallEndReason.REMOTE_HANGUP) + self.assertFalse(service.notify_incoming_ended("call-1")) + + async def test_incoming_bridge_failure_hangs_up_matching_call(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + FakeClient.bridge_error = True + FakeClient.remote_hangup = False + session, transition = self.make_incoming_session() + + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(FakeClient.instances[0].actions.count("hangup"), 1) + + async def test_incoming_caller_leaves_while_accept_is_pending(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + service = self.make_service() + accept_started = asyncio.Event() + release_accept = asyncio.Event() + + async def pending_accept(client): + client.actions.append("accept") + accept_started.set() + await release_accept.wait() + client.calls.emit("state", CallStateDetail("on-call", info=client.initial_info)) + + with patch.object(FakeClient, "acceptAsync", pending_accept): + task = asyncio.create_task(service.execute(session, transition)) + try: + await asyncio.wait_for(accept_started.wait(), timeout=1) + self.assertTrue(service.notify_incoming_ended("call-1")) + await asyncio.sleep(0) + release_accept.set() + await asyncio.wait_for(task, timeout=1) + finally: + release_accept.set() + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + self.assertNotIn("bridge", FakeClient.instances[0].actions) + self.assertEqual(FakeClient.instances[0].actions.count("hangup"), 1) + + async def test_new_call_after_accept_does_not_get_hung_up(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + + async def replaced_after_accept(client): + client.actions.append("accept") + client.calls.emit("state", CallStateDetail( + "on-call", info=CallInfo("call-2", "sender@c.us", None), + )) + + with patch.object(FakeClient, "acceptAsync", replaced_after_accept): + await self.make_service().execute(session, transition) + + self.assertEqual(session.state, CallState.REJECTED) + self.assertEqual(session.end_reason, CallEndReason.CALLER_LEFT) + self.assertEqual(FakeClient.instances[0].actions, ["open", "accept"]) + + async def test_shutdown_interrupts_pending_incoming_accept(self): + FakeClient.initial_state = "inc-call" + FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + session, transition = self.make_incoming_session() + service = self.make_service() + accept_started = asyncio.Event() + + async def pending_accept(client): + client.actions.append("accept") + accept_started.set() + await asyncio.Event().wait() + + with patch.object(FakeClient, "acceptAsync", pending_accept): + task = asyncio.create_task(service.execute(session, transition)) + try: + await asyncio.wait_for(accept_started.wait(), timeout=1) + service.request_stop() + await asyncio.wait_for(task, timeout=1) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) + self.assertEqual(FakeClient.instances[0].actions, ["open", "accept", "reject"]) + + async def test_reject_incoming_checks_both_call_id_and_sender(self): + api = Mock() + api.voip.getStateAsync = AsyncMock(return_value=CallStateDetail( + "inc-call", info=CallInfo("new-call", "sender@c.us", None), + )) + api.voip.rejectAsync = AsyncMock() + with patch("whatsapp_chatbot_python.calls.service.GreenAPI", return_value=api): + service = self.make_service() + self.assertFalse(await service.reject_incoming("call-1", "sender@c.us")) + api.voip.rejectAsync.assert_not_awaited() + + api.voip.getStateAsync.return_value = CallStateDetail( + "inc-call", info=CallInfo("call-1", "other@c.us", None), + ) + self.assertFalse(await service.reject_incoming("call-1", "sender@c.us")) + api.voip.rejectAsync.assert_not_awaited() + + api.voip.getStateAsync.return_value = CallStateDetail( + "inc-call", info=CallInfo("call-1", "sender@c.us", None), + ) + self.assertTrue(await service.reject_incoming("call-1", "sender@c.us")) + api.voip.rejectAsync.assert_awaited_once() + self.assertEqual(api.session.close.call_count, 3) + async def test_stop_before_execute_does_not_start_a_call(self): session, transition = self.make_session() diff --git a/tests/calls/test_call_state_machine.py b/tests/calls/test_call_state_machine.py index 78b5eb9..99eb904 100644 --- a/tests/calls/test_call_state_machine.py +++ b/tests/calls/test_call_state_machine.py @@ -2,12 +2,14 @@ import unittest from whatsapp_chatbot_python.calls.models import ( + CallDirection, CallEndReason, CallEvent, CallSession, CallState, TERMINAL_STATES, ) +from time import monotonic class CallStateMachineTest(unittest.TestCase): @@ -173,6 +175,50 @@ def test_all_terminal_states_ignore_late_events(self) -> None: self.assertFalse(result.applied) self.assertEqual(session.state, state) + def test_incoming_uses_existing_queue_and_answer_transitions(self) -> None: + session = CallSession.incoming( + sender_id="sender@c.us", call_id="call-1", language="ru", + incoming_deadline=monotonic() + 60, + ) + self.assertEqual(session.direction, CallDirection.INCOMING) + self.fsm.apply(session, CallEvent.ENQUEUED) + self.fsm.apply(session, CallEvent.DEQUEUED) + self.assertEqual(session.state, CallState.RINGING) + self.assertIsNone(session.dialing_at) + self.assertIsNone(session.answered_at) + + self.fsm.apply(session, CallEvent.INCOMING_ACCEPTED) + self.assertEqual(session.state, CallState.RINGING) + self.assertTrue(session.incoming_accept_completed) + self.assertIsNone(session.answered_at) + + self.fsm.apply(session, CallEvent.REMOTE_ACCEPTED) + self.assertEqual(session.state, CallState.CONNECTING) + self.fsm.apply(session, CallEvent.BRIDGE_READY) + self.assertEqual(session.state, CallState.IN_CALL) + self.assertIsNotNone(session.answered_at) + + def test_incoming_expiry_and_caller_left_from_queue(self) -> None: + for event, state, reason in ( + (CallEvent.INCOMING_EXPIRED, CallState.RING_TIMEOUT, CallEndReason.INCOMING_EXPIRED), + (CallEvent.CALLER_LEFT, CallState.REJECTED, CallEndReason.CALLER_LEFT), + ): + with self.subTest(event=event): + session = CallSession.incoming( + sender_id="sender@c.us", call_id="call-1", language="en", + incoming_deadline=monotonic() + 60, + ) + self.fsm.apply(session, CallEvent.ENQUEUED) + self.fsm.apply(session, event) + self.assertEqual((session.state, session.end_reason), (state, reason)) + self.assertFalse(self.fsm.apply(session, CallEvent.DEQUEUED).applied) + + def test_incoming_only_events_do_not_change_outgoing_calls(self) -> None: + self.apply(CallEvent.ENQUEUED) + for event in (CallEvent.INCOMING_EXPIRED, CallEvent.CALLER_LEFT, CallEvent.INCOMING_ACCEPTED): + self.assertFalse(self.fsm.apply(self.session, event).applied) + self.assertEqual(self.session.state, CallState.QUEUED) + if __name__ == "__main__": unittest.main() diff --git a/tests/calls/test_realtime_voice.py b/tests/calls/test_realtime_voice.py index c9f1401..91b78fb 100644 --- a/tests/calls/test_realtime_voice.py +++ b/tests/calls/test_realtime_voice.py @@ -1,4 +1,5 @@ from fractions import Fraction +import os from types import SimpleNamespace from unittest.mock import ANY, patch from whatsapp_chatbot_python.calls.playback_state_machine import PlaybackState @@ -115,6 +116,17 @@ def make_client(**kwargs): language="ru", on_error=self.failures.append, ) + def test_missing_explicit_api_key_does_not_fall_back_to_environment(self): + with patch.dict(os.environ, {"OPENAI_API_KEY": "ambient-key"}): + with patch("whatsapp_chatbot_python.calls.realtime_voice.AsyncOpenAI") as openai_client: + with self.assertRaisesRegex(ValueError, "OpenAI API key is required"): + VoiceBotSession( + api_key=None, model="gpt-realtime-2.1", voice="marin", + language="en", on_error=self.failures.append, + ) + + openai_client.assert_not_called() + async def asyncTearDown(self): await self.voice.close() @@ -128,6 +140,22 @@ async def test_session_configuration_and_single_greeting(self): self.assertEqual([a for a, _ in actions], ["update", "greet"]) self.assertEqual(actions[0][1]["session"]["audio"]["input"]["format"]["rate"], 24_000) + async def test_custom_greeting_is_used_once(self): + voice = VoiceBotSession( + api_key="test", model="gpt-realtime-2.1", voice="marin", + language="ru", greeting_text="Здравствуйте! Что хотите обсудить?", + on_error=self.failures.append, + ) + try: + await voice.start() + await voice.greet_once() + await voice.greet_once() + actions = self.client.connection.actions + self.assertEqual([action for action, _ in actions], ["update", "greet"]) + self.assertIn("Здравствуйте! Что хотите обсудить?", actions[1][1]["response"]["instructions"]) + finally: + await voice.close() + async def test_supported_languages_configure_voice_and_greeting(self): for code, language in { "ru": "Russian", "en": "English", "he": "Hebrew", diff --git a/tests/test_calls_imports.py b/tests/test_calls_imports.py index 7c0a413..bf8e929 100644 --- a/tests/test_calls_imports.py +++ b/tests/test_calls_imports.py @@ -31,8 +31,10 @@ def find_spec(self, fullname, path=None, target=None): def test_contracts_do_not_load_call_implementation(self): self.run_script(''' import sys -from whatsapp_chatbot_python.calls import CallSession, CallStateMachine +from whatsapp_chatbot_python.calls import CallDirection, CallSession, CallStateMachine, IncomingCallExecutor assert CallSession("sender", "chat", "en").state == "triggered" +assert CallSession("sender", "chat", "en").direction == CallDirection.OUTGOING +assert IncomingCallExecutor.__name__ == "IncomingCallExecutor" assert "whatsapp_chatbot_python.calls.service" not in sys.modules assert "whatsapp_chatbot_python.calls.realtime_voice" not in sys.modules assert "openai" not in sys.modules diff --git a/whatsapp_chatbot_python/calls/__init__.py b/whatsapp_chatbot_python/calls/__init__.py index 198bfe1..5fa2e80 100644 --- a/whatsapp_chatbot_python/calls/__init__.py +++ b/whatsapp_chatbot_python/calls/__init__.py @@ -11,11 +11,12 @@ if sys.version_info < (3, 11): raise ImportError("whatsapp_chatbot_python.calls requires Python 3.11 or newer") -from .contracts import CallExecutor, TransitionCallback +from .contracts import CallExecutor, IncomingCallExecutor, TransitionCallback from .models import ( ACTIVE_STATES, TERMINAL_STATES, WAITING_STATES, + CallDirection, CallEndReason, CallEvent, CallExecutionResult, @@ -27,8 +28,7 @@ from .state_machine import CallStateMachine __all__ = [ - "CallExecutor", "TransitionCallback", "CallEndReason", "CallEvent", - "CallExecutionResult", "CallSession", "CallState", "StateTransition", - "TransitionOutcome", "CallStateMachine", "ACTIVE_STATES", - "TERMINAL_STATES", "WAITING_STATES", + "CallExecutor", "IncomingCallExecutor", "TransitionCallback", "CallDirection", "CallEndReason", "CallEvent", + "CallExecutionResult", "CallSession", "CallState", "StateTransition", "TransitionOutcome", "CallStateMachine", + "ACTIVE_STATES", "TERMINAL_STATES", "WAITING_STATES", ] diff --git a/whatsapp_chatbot_python/calls/contracts.py b/whatsapp_chatbot_python/calls/contracts.py index fded04d..3ccc7ed 100644 --- a/whatsapp_chatbot_python/calls/contracts.py +++ b/whatsapp_chatbot_python/calls/contracts.py @@ -26,3 +26,11 @@ async def execute( ) -> CallExecutionResult: ... def request_stop(self) -> None: ... + + +class IncomingCallExecutor(CallExecutor, Protocol): + """Call control for application-owned queues; the app owns caller-left transitions.""" + + def notify_incoming_ended(self, call_id: str) -> bool: ... + + async def reject_incoming(self, call_id: str, sender_id: str) -> bool: ... diff --git a/whatsapp_chatbot_python/calls/models.py b/whatsapp_chatbot_python/calls/models.py index dde8092..edf86f4 100644 --- a/whatsapp_chatbot_python/calls/models.py +++ b/whatsapp_chatbot_python/calls/models.py @@ -1,7 +1,9 @@ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone +from time import monotonic from enum import StrEnum +from math import isfinite from pathlib import Path from uuid import uuid4 @@ -26,6 +28,11 @@ class CallState(StrEnum): FAILED = "failed" +class CallDirection(StrEnum): + OUTGOING = "outgoing" + INCOMING = "incoming" + + class CallEvent(StrEnum): ENQUEUED = "enqueued" DEQUEUED = "dequeued" @@ -39,6 +46,9 @@ class CallEvent(StrEnum): HANGUP_CONFIRMED = "hangup_confirmed" SHUTDOWN_REQUESTED = "shutdown_requested" INTERNAL_ERROR = "internal_error" + INCOMING_ACCEPTED = "incoming_accepted" + INCOMING_EXPIRED = "incoming_expired" + CALLER_LEFT = "caller_left" class CallEndReason(StrEnum): @@ -49,6 +59,8 @@ class CallEndReason(StrEnum): ERROR = "error" SHUTDOWN = "shutdown" USER_CANCELLED = "user_cancelled" + INCOMING_EXPIRED = "incoming_expired" + CALLER_LEFT = "caller_left" class TransitionOutcome(StrEnum): @@ -98,6 +110,28 @@ class CallSession: ended_at: datetime | None = None bridge_ready: bool = False remote_accepted: bool = False + direction: CallDirection = CallDirection.OUTGOING + incoming_deadline: float | None = None + greeting_text: str | None = None + incoming_accept_completed: bool = False + + @classmethod + def incoming( + cls, *, sender_id: str, call_id: str, language: str, + incoming_deadline: float, chat_id: str | None = None, + greeting_text: str | None = None, + ) -> CallSession: + if not sender_id or not call_id: + raise ValueError("Incoming calls require sender_id and call_id") + + if not isfinite(incoming_deadline) or incoming_deadline <= monotonic(): + raise ValueError("Incoming call deadline has expired") + + return cls( + sender_id=sender_id, chat_id=chat_id or sender_id, language=language, + call_id=call_id, direction=CallDirection.INCOMING, + incoming_deadline=incoming_deadline, greeting_text=greeting_text, + ) @dataclass(frozen=True, slots=True) diff --git a/whatsapp_chatbot_python/calls/realtime_voice.py b/whatsapp_chatbot_python/calls/realtime_voice.py index 34e45a3..630530d 100644 --- a/whatsapp_chatbot_python/calls/realtime_voice.py +++ b/whatsapp_chatbot_python/calls/realtime_voice.py @@ -134,13 +134,18 @@ def __init__( model: str, voice: str, language: str, + greeting_text: str | None = None, on_error: Callable[[Exception], None], recorder: CallRecorder | None = None, ) -> None: + if not isinstance(api_key, str) or not api_key.strip(): + raise ValueError("OpenAI API key is required for voice calls") + self._client = AsyncOpenAI(api_key=api_key) self._model = model self._voice = voice self._language = language + self._greeting_text = greeting_text self._on_error = on_error self.recorder = recorder self._manager = None @@ -218,10 +223,16 @@ async def greet_once(self) -> None: self._greeted = True + language = LANGUAGE_NAMES.get(self._language, "English") + + instructions = ( + f'Greet the caller: "{self._greeting_text}" in {language}.' + if self._greeting_text else + f"Greet the caller briefly in {language}." + ) + await self._connection.response.create(response={ - "instructions": ( - f"Greet the caller briefly in {LANGUAGE_NAMES.get(self._language, 'English')}." - ), + "instructions": instructions, }) async def new_output_track(self) -> BotOutputTrack: diff --git a/whatsapp_chatbot_python/calls/runtime.py b/whatsapp_chatbot_python/calls/runtime.py index 508daeb..388486c 100644 --- a/whatsapp_chatbot_python/calls/runtime.py +++ b/whatsapp_chatbot_python/calls/runtime.py @@ -14,6 +14,7 @@ class RuntimeEventType(StrEnum): END_CALL = "end_call" DISCONNECT = "disconnect" SHUTDOWN = "shutdown" + CALLER_LEFT = "caller_left" class RuntimeTimer(StrEnum): @@ -58,6 +59,10 @@ def disconnect(cls, permanent: bool) -> RuntimeEvent: def shutdown(cls) -> RuntimeEvent: return cls(RuntimeEventType.SHUTDOWN) + @classmethod + def caller_left(cls) -> RuntimeEvent: + return cls(RuntimeEventType.CALLER_LEFT) + @dataclass(slots=True) class CallRuntime: @@ -69,8 +74,8 @@ class CallRuntime: talk_deadline: float | None = None bridge_deadline: float | None = None - def start_ringing(self) -> None: - self.ring_deadline = monotonic() + self.ring_timeout_seconds + def start_ringing(self, *, deadline: float | None = None) -> None: + self.ring_deadline = deadline if deadline is not None else monotonic() + self.ring_timeout_seconds def remote_accepted(self, bridge_ready: bool) -> None: if self.talk_deadline is None: diff --git a/whatsapp_chatbot_python/calls/service.py b/whatsapp_chatbot_python/calls/service.py index 6a30c60..e748792 100644 --- a/whatsapp_chatbot_python/calls/service.py +++ b/whatsapp_chatbot_python/calls/service.py @@ -1,14 +1,15 @@ from __future__ import annotations from contextlib import suppress -from dataclasses import dataclass +from dataclasses import dataclass, field from functools import partial from pathlib import Path +from time import monotonic from whatsapp_api_client_python.API import GreenAPI -from whatsapp_api_client_python.tools.voip import CallAudio, CallsConnection +from whatsapp_api_client_python.tools.voip import CallAudio, CallsConnection, CallState as ApiCallState from .contracts import TransitionCallback from .realtime_voice import VoiceBotSession from .recording import CallRecorder -from .models import CallEvent, CallExecutionResult, CallSession, CallState, TERMINAL_STATES +from .models import CallDirection, CallEvent, CallExecutionResult, CallSession, CallState, TERMINAL_STATES from .runtime import CallRuntime, RuntimeEvent, RuntimeEventType, RuntimeTimer import asyncio import logging @@ -19,6 +20,23 @@ INITIAL_STATE_TIMEOUT_SECONDS = 10 +MATCHING_CALL_TIMEOUT_SECONDS = 5 + + +class _IncomingCallerLeft(Exception): + pass + + +class _IncomingDeadlineExceeded(TimeoutError): + pass + + +class _IncomingShutdown(Exception): + pass + + +_ANSWER_CONFIRMED = object() + @dataclass(slots=True) class _CallExecution: @@ -31,10 +49,16 @@ class _CallExecution: voice: VoiceBotSession | None = None api: GreenAPI | None = None calls: CallsConnection | None = None - initial_state: asyncio.Future[str] | None = None + initial_state: asyncio.Future[ApiCallState] | None = None bridge_task: asyncio.Task[None] | None = None - dial_started: bool = False - dialed: bool = False + command_started: bool = False + command_completed: bool = False + incoming_ended: bool = False + incoming_left_at: float | None = None + incoming_end_event: asyncio.Event = field(default_factory=asyncio.Event) + incoming_answered_at: float | None = None + incoming_answer_event: asyncio.Event = field(default_factory=asyncio.Event) + stop_event: asyncio.Event = field(default_factory=asyncio.Event) execution_completed: bool = False cancellation: asyncio.CancelledError | None = None recording_path: Path | None = None @@ -55,6 +79,14 @@ def __init__( shutdown_timeout_seconds: int, logger: logging.Logger, ) -> None: + for name, value in ( + ("id_instance", id_instance), + ("api_token_instance", api_token_instance), + ("openai_api_key", openai_api_key), + ): + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{name} is required for voice calls") + self._api_url = api_url self._id_instance = id_instance self._api_token_instance = api_token_instance @@ -67,6 +99,7 @@ def __init__( self._logger = logger self._active_loop: asyncio.AbstractEventLoop | None = None self._active_runtime: CallRuntime | None = None + self._active_execution: _CallExecution | None = None self._stop_requested = False def request_stop(self) -> None: @@ -74,8 +107,67 @@ def request_stop(self) -> None: loop = self._active_loop runtime = self._active_runtime - if loop is not None and runtime is not None and loop.is_running(): - loop.call_soon_threadsafe(runtime.events.put_nowait, RuntimeEvent.shutdown()) + if loop is not None and loop.is_running(): + execution = self._active_execution + + def signal() -> None: + if self._active_execution is execution and execution is not None: + execution.stop_event.set() + + if runtime is not None: + runtime.events.put_nowait(RuntimeEvent.shutdown()) + + loop.call_soon_threadsafe(signal) + + def notify_incoming_ended(self, call_id: str) -> bool: + """Wake the active incoming call after an external caller-ended notification. + + The caller must also transition the matching session under its queue lock, + even when this returns False during handoff. A stale call ID cannot stop + the next execution for the same sender. + """ + execution = self._active_execution + loop = self._active_loop + + if (execution is None or loop is None or not loop.is_running() + or execution.session.direction != CallDirection.INCOMING + or execution.session.call_id != call_id + ): + + return False + + def signal() -> None: + if self._active_execution is not execution: + return + + execution.incoming_ended = True + execution.incoming_left_at = monotonic() + + execution.incoming_end_event.set() + + if execution.runtime is not None: + execution.runtime.events.put_nowait(RuntimeEvent.caller_left()) + + loop.call_soon_threadsafe(signal) + + return True + + async def reject_incoming(self, call_id: str, sender_id: str) -> bool: + """Reject a waiting offer only while it still matches the live call.""" + + api = GreenAPI(self._id_instance, self._api_token_instance, host=self._api_url) + + try: + state = await asyncio.wait_for(api.voip.getStateAsync(), MATCHING_CALL_TIMEOUT_SECONDS) + + if not self._matches_incoming(state, call_id, sender_id, "inc-call"): + return False + + await asyncio.wait_for(api.voip.rejectAsync(), MATCHING_CALL_TIMEOUT_SECONDS) + + return True + finally: + api.session.close() async def execute( self, @@ -84,14 +176,51 @@ async def execute( ) -> CallExecutionResult: if self._stop_requested: transition(session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + if session.direction == CallDirection.INCOMING and session.call_id: + with suppress(Exception): + await self.reject_incoming(session.call_id, session.sender_id) + return CallExecutionResult() + if session.direction == CallDirection.INCOMING: + if not session.call_id or session.incoming_deadline is None: + raise ValueError("Incoming calls require call_id and incoming_deadline") + + if session.state in TERMINAL_STATES: + return CallExecutionResult() + + if monotonic() >= session.incoming_deadline: + transition(session, CallEvent.INCOMING_EXPIRED) + + with suppress(Exception): + await self.reject_incoming(session.call_id, session.sender_id) + + return CallExecutionResult() + + if session.state != CallState.RINGING: + raise ValueError("Incoming call must be dequeued before execution") + execution = _CallExecution(session, transition, CallRecorder(self._logger)) + self._active_loop = asyncio.get_running_loop() + self._active_execution = execution try: + # The application can end the offer between the first state check + # and this execution becoming visible to notify_incoming_ended(). + if session.direction == CallDirection.INCOMING and session.state in TERMINAL_STATES: + return CallExecutionResult() + await self._execute_call(execution) except asyncio.CancelledError as error: self._record_cancellation(execution, error) + raise + except Exception: + if session.direction == CallDirection.INCOMING: + transition(session, CallEvent.INTERNAL_ERROR, error_code="setup_failed") + + with suppress(Exception): + await self.reject_incoming(session.call_id, session.sender_id) + raise finally: await self._finalize_call(execution) @@ -100,7 +229,49 @@ async def execute( async def _execute_call(self, execution: _CallExecution) -> None: # Setup failures propagate to the caller; call failures become FSM events. - await self._prepare_call(execution) + if execution.session.direction == CallDirection.INCOMING: + try: + await self._await_incoming_operation(execution, self._prepare_call(execution)) + except _IncomingCallerLeft: + execution.transition(execution.session, CallEvent.CALLER_LEFT) + execution.execution_completed = True + return + except _IncomingDeadlineExceeded: + execution.transition(execution.session, CallEvent.INCOMING_EXPIRED) + + with suppress(Exception): + await self.reject_incoming(execution.session.call_id, execution.session.sender_id) + + execution.execution_completed = True + + return + except _IncomingShutdown: + execution.transition( + execution.session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown", + ) + + with suppress(Exception): + await self.reject_incoming(execution.session.call_id, execution.session.sender_id) + + execution.execution_completed = True + + return + if execution.incoming_ended: + execution.transition(execution.session, CallEvent.CALLER_LEFT) + execution.execution_completed = True + return + if self._stop_requested: + execution.transition( + execution.session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown", + ) + + await self._safe_finish_matching_incoming(execution.api, execution.session) + + execution.execution_completed = True + + return + else: + await self._prepare_call(execution) try: await self._run_call(execution) @@ -109,6 +280,74 @@ async def _execute_call(self, execution: _CallExecution) -> None: execution.execution_completed = True + @staticmethod + async def _await_incoming_operation( + execution: _CallExecution, operation, *, answer_confirms: bool = False, + ): + """Stop setup or signalling when the caller leaves or the offer expires.""" + if execution.incoming_ended or execution.session.state in TERMINAL_STATES: + operation.close() + raise _IncomingCallerLeft + + if execution.stop_event.is_set(): + operation.close() + raise _IncomingShutdown + + remaining = execution.session.incoming_deadline - monotonic() + + if remaining <= 0: + operation.close() + raise _IncomingDeadlineExceeded + + task = asyncio.create_task(operation) + ended = asyncio.create_task(execution.incoming_end_event.wait()) + stopped = asyncio.create_task(execution.stop_event.wait()) + answered = asyncio.create_task(execution.incoming_answer_event.wait()) if answer_confirms else None + + try: + pending = {task, ended, stopped} + + if answered is not None: + pending.add(answered) + + done, _ = await asyncio.wait( + pending, timeout=remaining, return_when=asyncio.FIRST_COMPLETED, + ) + + if task in done: + try: + return await task + except Exception: + if answer_confirms and WhatsAppCallService._answer_before_deadline(execution): + return _ANSWER_CONFIRMED + + raise + + if ended in done: + raise _IncomingCallerLeft + + if stopped in done: + raise _IncomingShutdown + + if answer_confirms and WhatsAppCallService._answer_before_deadline(execution): + return _ANSWER_CONFIRMED + + raise _IncomingDeadlineExceeded + finally: + ended.cancel() + stopped.cancel() + + if answered is not None: + answered.cancel() + + if not task.done(): + task.cancel() + + await asyncio.gather( + task, ended, stopped, *([answered] if answered is not None else []), + return_exceptions=True, + ) + async def _prepare_call(self, execution: _CallExecution) -> None: runtime = CallRuntime( ring_timeout_seconds=self._ring_timeout, @@ -117,7 +356,6 @@ async def _prepare_call(self, execution: _CallExecution) -> None: ) execution.runtime = runtime - self._active_loop = asyncio.get_running_loop() self._active_runtime = runtime if self._stop_requested: @@ -128,6 +366,7 @@ async def _prepare_call(self, execution: _CallExecution) -> None: model=self._realtime_model, voice=self._realtime_voice, language=execution.session.language, + greeting_text=execution.session.greeting_text, on_error=lambda error: runtime.events.put_nowait(RuntimeEvent.voice_failed(error)), recorder=execution.recorder, ) @@ -181,11 +420,75 @@ def _register_listeners(self, execution: _CallExecution) -> None: @staticmethod def _on_state(execution: _CallExecution, state) -> None: if not execution.initial_state.done(): - execution.initial_state.set_result(state.state) + execution.initial_state.set_result(state) + + if execution.session.direction == CallDirection.INCOMING: + if state.state == "on-call" and WhatsAppCallService._matches_incoming( + state, execution.session.call_id, execution.session.sender_id, "on-call", + ): + if execution.incoming_answered_at is None: + execution.incoming_answered_at = monotonic() + + execution.incoming_answer_event.set() + + # Apply the answer through the same locked transition callback + # used by the application's caller-ended notification. Otherwise + # that notification could classify an already answered call as + # an unanswered offer while this event waits in the RTC queue. + if execution.command_started: + WhatsAppCallService._apply_observed_incoming_answer(execution) + + if state.state == "idle" and not execution.command_completed: + execution.incoming_ended = True + execution.incoming_left_at = monotonic() + execution.incoming_end_event.set() + + if state.state in {"inc-call", "on-call"} and not WhatsAppCallService._matches_incoming( + state, execution.session.call_id, execution.session.sender_id, state.state, + ): + execution.incoming_ended = True + execution.incoming_left_at = monotonic() + + execution.incoming_end_event.set() + execution.runtime.events.put_nowait(RuntimeEvent.caller_left()) + + return - if execution.dial_started: + if execution.command_started: execution.runtime.events.put_nowait(RuntimeEvent.server_state(state.state, state.reason)) + @staticmethod + def _matches_incoming(state: ApiCallState | None, call_id: str | None, sender_id: str, kind: str) -> bool: + info = getattr(state, "info", None) + + return ( + getattr(state, "state", None) == kind and info is not None + and info.id == call_id and info.wid == sender_id + ) + + @staticmethod + def _answer_before_deadline(execution: _CallExecution) -> bool: + observed_at = execution.incoming_answered_at + + return ( + observed_at is not None + and observed_at < execution.session.incoming_deadline + and (execution.incoming_left_at is None or observed_at < execution.incoming_left_at) + ) + + @staticmethod + def _apply_observed_incoming_answer(execution: _CallExecution) -> bool: + if not WhatsAppCallService._answer_before_deadline(execution): + return False + + session = execution.session + + if session.state == CallState.RINGING: + execution.transition(session, CallEvent.REMOTE_ACCEPTED) + execution.runtime.remote_accepted(session.bridge_ready) + + return session.remote_accepted + @staticmethod def _on_disconnect(execution: _CallExecution, detail) -> None: permanent = bool(detail.get("permanent")) @@ -206,21 +509,54 @@ def _on_bridge_done(execution: _CallExecution, task: asyncio.Task[None]) -> None execution.runtime.events.put_nowait(event) async def _run_call(self, execution: _CallExecution) -> None: - await execution.calls.openAsync(timeout=INITIAL_STATE_TIMEOUT_SECONDS) + if execution.session.direction == CallDirection.INCOMING: + execution.runtime.start_ringing(deadline=execution.session.incoming_deadline) + try: + await self._await_incoming_operation( + execution, execution.calls.openAsync(timeout=INITIAL_STATE_TIMEOUT_SECONDS), + ) + + await self._await_incoming_operation( + execution, + asyncio.wait_for(execution.initial_state, timeout=INITIAL_STATE_TIMEOUT_SECONDS), + ) + except _IncomingCallerLeft: + execution.transition(execution.session, CallEvent.CALLER_LEFT) + return + + except _IncomingShutdown: + execution.transition( + execution.session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown", + ) + + await self._safe_finish_matching_incoming(execution.api, execution.session) + + return + + except _IncomingDeadlineExceeded: + execution.transition(execution.session, CallEvent.INCOMING_EXPIRED) + await self._safe_finish_matching_incoming(execution.api, execution.session) + return + + await self._run_incoming(execution) + + return + + await execution.calls.openAsync(timeout=INITIAL_STATE_TIMEOUT_SECONDS) state = await asyncio.wait_for(execution.initial_state, timeout=INITIAL_STATE_TIMEOUT_SECONDS) - if state != "idle": - raise RuntimeError(f"Instance already has a call: {state}") + if state.state != "idle": + raise RuntimeError(f"Instance already has a call: {state.state}") if self._stop_requested: execution.transition(execution.session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") else: - execution.dial_started = True + execution.command_started = True await execution.api.voip.dialAsync(execution.session.chat_id) - execution.dialed = True + execution.command_completed = True execution.transition(execution.session, CallEvent.DIAL_ACCEPTED) execution.runtime.start_ringing() @@ -232,10 +568,138 @@ async def _run_call(self, execution: _CallExecution) -> None: await self._process_events(execution) + async def _run_incoming(self, execution: _CallExecution) -> None: + session = execution.session + state = execution.calls.state + + if session.state in TERMINAL_STATES: + await self._safe_finish_matching_incoming(execution.api, session) + return + + if execution.incoming_ended: + self._apply_observed_incoming_answer(execution) + execution.transition(session, CallEvent.CALLER_LEFT) + return + + if not self._matches_incoming(state, session.call_id, session.sender_id, "inc-call"): + execution.transition(session, CallEvent.CALLER_LEFT) + return + + if monotonic() >= session.incoming_deadline: + execution.transition(session, CallEvent.INCOMING_EXPIRED) + await self._safe_finish_matching_incoming(execution.api, session) + return + + if self._stop_requested: + execution.transition(session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + await self._safe_finish_matching_incoming(execution.api, session) + return + + execution.command_started = True + + try: + accept_result = await self._await_incoming_operation( + execution, execution.api.voip.acceptAsync(), answer_confirms=True, + ) + except _IncomingCallerLeft: + self._apply_observed_incoming_answer(execution) + execution.transition(session, CallEvent.CALLER_LEFT) + await self._safe_finish_matching_incoming(execution.api, session) + + return + except _IncomingShutdown: + execution.transition(session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + await self._safe_finish_matching_incoming(execution.api, session) + return + except _IncomingDeadlineExceeded: + if not self._apply_observed_incoming_answer(execution): + execution.transition(session, CallEvent.INCOMING_EXPIRED) + await self._safe_finish_matching_incoming(execution.api, session) + return + + accept_result = _ANSWER_CONFIRMED + except Exception as error: + if not await self._reconcile_failed_incoming_accept(execution, error): + return + + accept_result = _ANSWER_CONFIRMED + + if accept_result is not _ANSWER_CONFIRMED: + execution.command_completed = True + execution.transition(session, CallEvent.INCOMING_ACCEPTED) + self._apply_observed_incoming_answer(execution) + + if self._stop_requested: + execution.transition(session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") + await self._safe_finish_matching_incoming(execution.api, session) + return + + if execution.incoming_ended or session.state in TERMINAL_STATES: + if execution.incoming_ended: + execution.transition(session, CallEvent.CALLER_LEFT) + + await self._safe_finish_matching_incoming(execution.api, session) + + return + + if not session.remote_accepted and monotonic() >= session.incoming_deadline: + execution.transition(session, CallEvent.INCOMING_EXPIRED) + await self._safe_finish_matching_incoming(execution.api, session) + return + + execution.bridge_task = asyncio.create_task(execution.calls.startAudioAsync()) + execution.bridge_task.add_done_callback(partial(self._on_bridge_done, execution)) + + # The state notification may have arrived before acceptAsync returned. + state = execution.calls.state + if state is not None and state.state == "on-call": + execution.runtime.events.put_nowait(RuntimeEvent.server_state("on-call", state.reason)) + + await self._process_events(execution) + + async def _reconcile_failed_incoming_accept( + self, execution: _CallExecution, accept_error: Exception, + ) -> bool: + """Classify an accept error using the current call, if it can be read.""" + self._apply_observed_incoming_answer(execution) + + session = execution.session + + if execution.incoming_ended or session.state in TERMINAL_STATES: + execution.transition(session, CallEvent.CALLER_LEFT) + return False + + try: + state = await asyncio.wait_for( + execution.api.voip.getStateAsync(), MATCHING_CALL_TIMEOUT_SECONDS, + ) + except Exception: + raise accept_error + + if self._matches_incoming(state, session.call_id, session.sender_id, "on-call"): + # The REST response failed, but the server confirms this call was answered. + execution.incoming_answered_at = monotonic() + + execution.transition(session, CallEvent.REMOTE_ACCEPTED) + execution.runtime.remote_accepted(session.bridge_ready) + + return True + if not self._matches_incoming(state, session.call_id, session.sender_id, "inc-call"): + execution.transition(session, CallEvent.CALLER_LEFT) + return False + + raise accept_error + async def _process_events(self, execution: _CallExecution) -> None: while execution.session.state not in TERMINAL_STATES: + if await self._finish_interrupted_incoming(execution): + break + event = await execution.runtime.next_event(execution.session.state) + if await self._finish_interrupted_incoming(execution): + break + if event is None: await self._handle_timeout( execution.runtime, execution.api, execution.session, execution.transition, @@ -246,6 +710,26 @@ async def _process_events(self, execution: _CallExecution) -> None: execution.recorder, execution.session, execution.transition, ) + async def _finish_interrupted_incoming(self, execution: _CallExecution) -> bool: + session = execution.session + + if session.direction != CallDirection.INCOMING: + return False + + self._apply_observed_incoming_answer(execution) + + if execution.incoming_ended: + execution.transition(session, CallEvent.CALLER_LEFT) + await self._safe_finish_matching_incoming(execution.api, session) + return True + + if session.state == CallState.RINGING and monotonic() >= session.incoming_deadline: + execution.transition(session, CallEvent.INCOMING_EXPIRED) + await self._safe_finish_matching_incoming(execution.api, session) + return True + + return False + async def _handle_call_failure(self, execution: _CallExecution, error: Exception) -> None: self._logger.error( "VoIP execution failed: session=%s error=%s", @@ -258,7 +742,9 @@ async def _handle_call_failure(self, execution: _CallExecution, error: Exception error_code=type(error).__name__, ) - if execution.dialed: + if execution.session.direction == CallDirection.INCOMING: + await self._safe_finish_matching_incoming(execution.api, execution.session) + elif execution.command_completed: await self._safe_hang_up(execution.api) @staticmethod @@ -322,17 +808,26 @@ async def _close_connections(self, execution: _CallExecution) -> None: ) async def _hang_up_after_cancellation(self, execution: _CallExecution) -> None: - # The dial request may have reached the server before its response arrived. - if execution.cancellation is not None and execution.dial_started: + # The command may have reached the server before its response arrived. + if execution.cancellation is not None and ( + execution.command_started or execution.session.direction == CallDirection.INCOMING + ): with suppress(Exception): + cleanup = ( + self._finish_matching_incoming(execution.api, execution.session) + if execution.session.direction == CallDirection.INCOMING + else self._safe_hang_up(execution.api) + ) + await self._wait_for_cleanup( - asyncio.wait_for(self._safe_hang_up(execution.api), timeout=self._shutdown_timeout), + asyncio.wait_for(cleanup, timeout=self._shutdown_timeout), execution, ) def _finish_recording(self, execution: _CallExecution) -> None: self._active_runtime = None self._active_loop = None + self._active_execution = None path = execution.recorder.finish() keep_recording = ( @@ -388,9 +883,15 @@ async def _handle_event( if event.permanent: raise RuntimeError("callsRtc connection permanently closed") # The library reconnects and rebuilds an active audio bridge. + elif event.type == RuntimeEventType.CALLER_LEFT: + transition(call_session, CallEvent.CALLER_LEFT) elif event.type == RuntimeEventType.SHUTDOWN: transition(call_session, CallEvent.SHUTDOWN_REQUESTED, error_code="shutdown") - await self._safe_hang_up(api) + + if call_session.direction == CallDirection.INCOMING: + await self._safe_finish_matching_incoming(api, call_session) + else: + await self._safe_hang_up(api) async def _handle_timeout( self, @@ -402,8 +903,12 @@ async def _handle_timeout( timer = runtime.expired_timer(call_session.state) if timer == RuntimeTimer.RING: - transition(call_session, CallEvent.RING_TIMER_EXPIRED) - await self._hang_up(api, call_session, transition) + if call_session.direction == CallDirection.INCOMING: + transition(call_session, CallEvent.INCOMING_EXPIRED) + await self._safe_finish_matching_incoming(api, call_session) + else: + transition(call_session, CallEvent.RING_TIMER_EXPIRED) + await self._hang_up(api, call_session, transition) elif timer == RuntimeTimer.TALK: transition(call_session, CallEvent.TALK_TIMER_EXPIRED) await self._hang_up(api, call_session, transition) @@ -413,7 +918,10 @@ async def _handle_timeout( error_code="bridge_negotiation_timeout", ) - await self._safe_hang_up(api) + if call_session.direction == CallDirection.INCOMING: + await self._safe_finish_matching_incoming(api, call_session) + else: + await self._safe_hang_up(api) @staticmethod async def _hang_up( @@ -422,7 +930,10 @@ async def _hang_up( transition: TransitionCallback, ) -> None: try: - await api.voip.hangUpAsync() + if call_session.direction == CallDirection.INCOMING: + await WhatsAppCallService._finish_matching_incoming(api, call_session) + else: + await api.voip.hangUpAsync() except Exception as error: transition( call_session, CallEvent.INTERNAL_ERROR, @@ -433,6 +944,33 @@ async def _hang_up( transition(call_session, CallEvent.HANGUP_CONFIRMED) + @staticmethod + async def _finish_matching_incoming(api: GreenAPI | None, session: CallSession) -> None: + if api is None: + return + + state = await asyncio.wait_for(api.voip.getStateAsync(), MATCHING_CALL_TIMEOUT_SECONDS) + + if WhatsAppCallService._matches_incoming( + state, session.call_id, session.sender_id, "inc-call", + ): + await asyncio.wait_for(api.voip.rejectAsync(), MATCHING_CALL_TIMEOUT_SECONDS) + elif WhatsAppCallService._matches_incoming( + state, session.call_id, session.sender_id, "on-call", + ): + await asyncio.wait_for(api.voip.hangUpAsync(), MATCHING_CALL_TIMEOUT_SECONDS) + + async def _safe_finish_matching_incoming( + self, api: GreenAPI | None, session: CallSession, + ) -> None: + try: + await self._finish_matching_incoming(api, session) + except Exception as error: + self._logger.warning( + "Unable to finish incoming call: session=%s error=%s", + session.session_id, type(error).__name__, + ) + @staticmethod async def _safe_hang_up(api: GreenAPI) -> None: with suppress(Exception): diff --git a/whatsapp_chatbot_python/calls/state_machine.py b/whatsapp_chatbot_python/calls/state_machine.py index 91fc7d4..f55b541 100644 --- a/whatsapp_chatbot_python/calls/state_machine.py +++ b/whatsapp_chatbot_python/calls/state_machine.py @@ -3,6 +3,7 @@ from dataclasses import dataclass from .models import ( + CallDirection, CallEndReason, CallEvent, CallSession, @@ -49,10 +50,16 @@ def _ending_target(session: CallSession) -> CallState: }.get(session.end_reason, CallState.FAILED) +def _dequeued_target(session: CallSession) -> CallState: + return CallState.RINGING if session.direction == CallDirection.INCOMING else CallState.DIALING + + _TRANSITION_RULES: dict[tuple[CallState, CallEvent], TransitionRule] = { (CallState.TRIGGERED, CallEvent.ENQUEUED): TransitionRule(CallState.QUEUED), - (CallState.QUEUED, CallEvent.DEQUEUED): TransitionRule(CallState.DIALING), + (CallState.QUEUED, CallEvent.DEQUEUED): TransitionRule(_dequeued_target), (CallState.QUEUED, CallEvent.CANCEL_REQUESTED): TransitionRule(CallState.CANCELLED), + (CallState.QUEUED, CallEvent.INCOMING_EXPIRED): TransitionRule(CallState.RING_TIMEOUT), + (CallState.QUEUED, CallEvent.CALLER_LEFT): TransitionRule(CallState.REJECTED), (CallState.DIALING, CallEvent.DIAL_ACCEPTED): TransitionRule(CallState.RINGING), (CallState.DIALING, CallEvent.REMOTE_IDLE): TransitionRule(CallState.REJECTED), @@ -69,8 +76,13 @@ def _ending_target(session: CallSession) -> CallState: (CallState.RINGING, CallEvent.RING_TIMER_EXPIRED): TransitionRule( CallState.ENDING ), + (CallState.RINGING, CallEvent.INCOMING_ACCEPTED): TransitionRule(CallState.RINGING), + (CallState.RINGING, CallEvent.INCOMING_EXPIRED): TransitionRule(CallState.RING_TIMEOUT), + (CallState.RINGING, CallEvent.CALLER_LEFT): TransitionRule(CallState.REJECTED), (CallState.CONNECTING, CallEvent.BRIDGE_READY): TransitionRule(CallState.IN_CALL), + (CallState.CONNECTING, CallEvent.INCOMING_ACCEPTED): TransitionRule(CallState.CONNECTING), + (CallState.CONNECTING, CallEvent.CALLER_LEFT): TransitionRule(CallState.REMOTE_ENDED), # The remote party already answered; report a completed call even if the # bridge never became ready. No recording is created until IN_CALL. @@ -83,6 +95,8 @@ def _ending_target(session: CallSession) -> CallState: ), (CallState.IN_CALL, CallEvent.REMOTE_IDLE): TransitionRule(CallState.REMOTE_ENDED), + (CallState.IN_CALL, CallEvent.INCOMING_ACCEPTED): TransitionRule(CallState.IN_CALL), + (CallState.IN_CALL, CallEvent.CALLER_LEFT): TransitionRule(CallState.REMOTE_ENDED), (CallState.IN_CALL, CallEvent.TALK_TIMER_EXPIRED): TransitionRule(CallState.ENDING), (CallState.ENDING, CallEvent.HANGUP_CONFIRMED): TransitionRule(_ending_target), (CallState.ENDING, CallEvent.REMOTE_IDLE): TransitionRule(_ending_target), @@ -124,6 +138,11 @@ def apply( if previous in TERMINAL_STATES: return self._ignored(previous, event) + if event in { + CallEvent.INCOMING_ACCEPTED, CallEvent.INCOMING_EXPIRED, CallEvent.CALLER_LEFT, + } and session.direction != CallDirection.INCOMING: + return self._ignored(previous, event) + rule = _TRANSITION_RULES.get((previous, event)) if rule is None: @@ -167,7 +186,18 @@ def _apply_event_data(session: CallSession, event: CallEvent) -> None: if event == CallEvent.ENQUEUED: session.queued_at = now elif event == CallEvent.DEQUEUED: - session.dialing_at = now + if session.direction == CallDirection.OUTGOING: + session.dialing_at = now + elif event == CallEvent.INCOMING_ACCEPTED: + session.incoming_accept_completed = True + elif event == CallEvent.INCOMING_EXPIRED: + session.end_reason = CallEndReason.INCOMING_EXPIRED + elif event == CallEvent.CALLER_LEFT: + session.end_reason = ( + CallEndReason.REMOTE_HANGUP + if session.state in {CallState.CONNECTING, CallState.IN_CALL} + else CallEndReason.CALLER_LEFT + ) elif event == CallEvent.CANCEL_REQUESTED: session.end_reason = CallEndReason.USER_CANCELLED elif event == CallEvent.REMOTE_ACCEPTED: From b581462dcf354586d667595a340690452dff18b2 Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Mon, 19 Oct 2026 15:45:38 +0500 Subject: [PATCH 2/4] feat: Some interfaces and contracts have been improved; Breaking changes in VoIP --- examples/call.py | 8 +- tests/test_calls_imports.py | 3 +- tests/test_incoming_call_notifications.py | 112 ++++++++++++++++ tests/typing/incoming_call_contract.py | 30 +++++ whatsapp_chatbot_python/__init__.py | 3 + whatsapp_chatbot_python/manager/handler.py | 25 ++-- .../manager/incoming_call.py | 122 ++++++++++++++++++ whatsapp_chatbot_python/manager/observer.py | 100 +++++++++++--- whatsapp_chatbot_python/manager/router.py | 20 +-- 9 files changed, 380 insertions(+), 43 deletions(-) create mode 100644 tests/test_incoming_call_notifications.py create mode 100644 tests/typing/incoming_call_contract.py create mode 100644 whatsapp_chatbot_python/manager/incoming_call.py diff --git a/examples/call.py b/examples/call.py index dfc1664..d42142c 100644 --- a/examples/call.py +++ b/examples/call.py @@ -1,4 +1,4 @@ -from whatsapp_chatbot_python import GreenAPIBot, Notification +from whatsapp_chatbot_python import GreenAPIBot, IncomingCallNotification bot = GreenAPIBot( "1101000001", "d75b3a66374942c5b3c019c698abc2067e151558acbd412345" @@ -6,7 +6,9 @@ @bot.router.incoming_call() -def message_handler(notification: Notification) -> None: - print("Call from:", notification.event["from"], "with status", notification.event["status"]) +def incoming_call_handler(notification: IncomingCallNotification) -> None: + call = notification.call + print("Call from:", call.sender, "with status", call.status) + # notification.event (or notification.raw_event) still contains the original webhook. bot.run_forever() diff --git a/tests/test_calls_imports.py b/tests/test_calls_imports.py index bf8e929..3384bdb 100644 --- a/tests/test_calls_imports.py +++ b/tests/test_calls_imports.py @@ -23,7 +23,8 @@ def find_spec(self, fullname, path=None, target=None): raise ImportError("optional dependency blocked: " + fullname) sys.meta_path.insert(0, BlockOptionalImports()) -from whatsapp_chatbot_python import GreenAPIBot, Notification +from whatsapp_chatbot_python import GreenAPIBot, IncomingCall, IncomingCallNotification, Notification +assert IncomingCallNotification.__name__ == "IncomingCallNotification" assert "whatsapp_chatbot_python.calls" not in sys.modules ''') diff --git a/tests/test_incoming_call_notifications.py b/tests/test_incoming_call_notifications.py new file mode 100644 index 0000000..61b3265 --- /dev/null +++ b/tests/test_incoming_call_notifications.py @@ -0,0 +1,112 @@ +import logging +import unittest +from unittest.mock import Mock + +from whatsapp_chatbot_python import IncomingCall, IncomingCallNotification, Notification +from whatsapp_chatbot_python.manager.router import Router + + +class IncomingCallNotificationTest(unittest.TestCase): + def setUp(self): + self.api = Mock() + self.router = Router(self.api, logging.getLogger(__name__)) + + @staticmethod + def event(status="offer", timestamp=1_700_000_000): + return { + "typeWebhook": "incomingCall", "from": "caller@c.us", + "idMessage": "call-1", "status": status, "timestamp": timestamp, + } + + def test_typed_notification_keeps_original_webhook_and_message_methods_work(self): + received = [] + self.router.incoming_call.add_handler(received.append) + event = self.event(status="future-status") + + self.router.route_event(event) + + notification = received[0] + self.assertIsInstance(notification, IncomingCallNotification) + self.assertIsInstance(notification, Notification) + self.assertIs(notification.event, event) + self.assertIs(notification.raw_event, event) + self.assertIs(notification.call.raw_event, event) + self.assertEqual(notification.call, IncomingCall("caller@c.us", "call-1", "future-status", 1_700_000_000.0)) + self.assertEqual(notification.call.chat_id, "caller@c.us") + self.assertEqual((notification.sender, notification.chat, notification.get_id_message()), + ("caller@c.us", "caller@c.us", "call-1")) + self.assertIsNone(notification.message_text) + self.assertIsNone(notification.get_message_data()) + + notification.answer("Hello") + self.api.sending.sendMessage.assert_called_once_with( + "caller@c.us", "Hello", None, None, None, None, + ) + + def test_call_and_message_observers_share_state_only_with_each_other(self): + state_manager = self.router.message.state_manager + self.assertIs(self.router.incoming_call.state_manager, state_manager) + self.assertIsNot(self.router.outgoing_message.state_manager, state_manager) + state_manager.set_state("caller@c.us", "MENU") + state_manager.set_state_data("caller@c.us", {"language_code": "ru"}) + received = [] + + @self.router.incoming_call(status="offer", state="MENU", from_sender="caller@c.us") + def on_offer(notification): + received.append(notification.state_manager.get_state_data(notification.call.sender)) + + self.router.route_event(self.event()) + self.assertEqual(received, [{"language_code": "ru"}]) + + state_manager.update_state("caller@c.us", "DONE") + self.router.route_event(self.event()) + self.assertEqual(len(received), 1) + + def test_status_filter_accepts_one_or_many_values_and_unknown_status_is_routed(self): + offers, ended, other = [], [], [] + self.router.incoming_call.add_handler(offers.append, status="offer") + self.router.incoming_call.add_handler(ended.append, status=["declined", "hungUp"]) + self.router.incoming_call.add_handler(other.append) + + for status in ("offer", "declined", "hungUp", "new-status"): + self.router.route_event(self.event(status=status)) + + self.assertEqual([item.call.status for item in offers], ["offer"]) + self.assertEqual([item.call.status for item in ended], ["declined", "hungUp"]) + self.assertEqual([item.call.status for item in other], ["new-status"]) + + def test_invalid_call_filters_fail_at_registration(self): + for filters in ({"status": None}, {"status": []}, {"status": 123}, + {"status": ["offer", ""]}, {"command": "start"}): + with self.subTest(filters=filters), self.assertRaises(ValueError): + self.router.incoming_call.add_handler(lambda notification: None, **filters) + self.assertEqual(self.router.incoming_call.handlers, []) + + def test_timestamps_and_missing_fields_are_safe(self): + for value, expected in ((1_700_000_000, 1_700_000_000.0), + (1_700_000_000_000, 1_700_000_000.0), + (1_700_000_000_000_000, 1_700_000_000.0), + (946_684_800, 946_684_800.0), + (4_102_444_800, None), + (1_700_000_000_000_000_000, None), + (12_345_678_901_234_567_890, None), + (True, None), (float("nan"), None), (10 ** 1000, None)): + with self.subTest(timestamp=value): + self.assertEqual(IncomingCall.from_event(self.event(timestamp=value)).timestamp, expected) + self.assertEqual(IncomingCall.from_event({}), IncomingCall(None, None, None, None)) + + def test_existing_message_notification_remains_unchanged(self): + received = [] + self.router.message.add_handler(received.append) + event = { + "typeWebhook": "incomingMessageReceived", + "senderData": {"sender": "caller@c.us", "chatId": "caller@c.us"}, + "messageData": {"typeMessage": "textMessage", "textMessageData": {"textMessage": "Hi"}}, + } + self.router.route_event(event) + self.assertIs(type(received[0]), Notification) + self.assertEqual(received[0].message_text, "Hi") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/typing/incoming_call_contract.py b/tests/typing/incoming_call_contract.py new file mode 100644 index 0000000..a945029 --- /dev/null +++ b/tests/typing/incoming_call_contract.py @@ -0,0 +1,30 @@ +"""Static contract for incoming-call handlers; checked by mypy in CI.""" + +from typing import Callable, Optional + +from typing_extensions import assert_type + +from whatsapp_chatbot_python import GreenAPIBot, IncomingCallNotification, Notification + + +def register_handlers(bot: GreenAPIBot) -> None: + @bot.router.incoming_call(status="offer") + def on_offer(notification: IncomingCallNotification) -> None: + assert_type(notification.call.sender, Optional[str]) + + assert_type(on_offer, Callable[[IncomingCallNotification], None]) + bot.router.incoming_call.add_handler(on_offer) + + def on_general(notification: Notification) -> None: + pass + + bot.router.incoming_call.add_handler(on_general) + + @bot.router.incoming_call() # type: ignore[arg-type] + def wrong(value: int) -> None: + pass + + def wrong_direct(value: int) -> None: + pass + + bot.router.incoming_call.add_handler(wrong_direct) # type: ignore[arg-type] diff --git a/whatsapp_chatbot_python/__init__.py b/whatsapp_chatbot_python/__init__.py index cb17e73..6117d83 100644 --- a/whatsapp_chatbot_python/__init__.py +++ b/whatsapp_chatbot_python/__init__.py @@ -1,5 +1,6 @@ from .bot import Bot, GreenAPI, GreenAPIBot, GreenAPIError, GreenAPIBotError from .manager.handler import Notification +from .manager.incoming_call import IncomingCall, IncomingCallNotification from .manager.state import BaseStates __all__ = [ @@ -9,5 +10,7 @@ "GreenAPIError", "GreenAPIBotError", "Notification", + "IncomingCall", + "IncomingCallNotification", "BaseStates" ] diff --git a/whatsapp_chatbot_python/manager/handler.py b/whatsapp_chatbot_python/manager/handler.py index 09061f3..2c2958f 100644 --- a/whatsapp_chatbot_python/manager/handler.py +++ b/whatsapp_chatbot_python/manager/handler.py @@ -1,14 +1,14 @@ import json import logging from abc import ABC, abstractmethod -from typing import Any, Callable, Dict, List, Optional, TYPE_CHECKING, Union +from typing import Any, Callable, Dict, Generic, List, Optional, TYPE_CHECKING, TypeVar, Union from whatsapp_api_client_python.response import Response from ..filters import filters as event_filters if TYPE_CHECKING: - from .observer import Observer + from .observer import AbstractObserver from .state import AbstractStateManager from ..bot import GreenAPI @@ -167,27 +167,28 @@ def answer_with_poll( ) HandlerType = Callable[[Notification], Any] +NotificationT = TypeVar("NotificationT", bound=Notification) -class AbstractHandler(ABC): - handler: HandlerType +class AbstractHandler(ABC, Generic[NotificationT]): + handler: Callable[[NotificationT], Any] filters: Dict[str, Any] @abstractmethod - def check_event(self, notification: Notification) -> bool: + def check_event(self, notification: NotificationT) -> bool: pass @abstractmethod - def execute_handler(self, observer: "Observer") -> bool: + def execute_handler(self, observer: "AbstractObserver[NotificationT]") -> bool: pass -class Handler(AbstractHandler): - def __init__(self, handler: HandlerType, **filters: Any): +class Handler(AbstractHandler[NotificationT], Generic[NotificationT]): + def __init__(self, handler: Callable[[NotificationT], Any], **filters: Any): self.handler = handler self.filters = filters - def check_event(self, notification: Notification) -> bool: + def check_event(self, notification: NotificationT) -> bool: for filter_name in self.filters.keys(): filter_ = event_filters.get(filter_name) if filter_: @@ -198,10 +199,8 @@ def check_event(self, notification: Notification) -> bool: return True - def execute_handler(self, observer: "Observer") -> bool: - notification = Notification( - observer.event, observer.router.api, observer.state_manager - ) + def execute_handler(self, observer: "AbstractObserver[NotificationT]") -> bool: + notification = observer.make_notification() filters = json.dumps(self.filters, ensure_ascii=False) diff --git a/whatsapp_chatbot_python/manager/incoming_call.py b/whatsapp_chatbot_python/manager/incoming_call.py new file mode 100644 index 0000000..c7385b2 --- /dev/null +++ b/whatsapp_chatbot_python/manager/incoming_call.py @@ -0,0 +1,122 @@ +"""Typed access to an incoming-call webhook without VoIP dependencies.""" + +from dataclasses import dataclass, field +from math import isfinite +from typing import Any, Dict, Optional, TYPE_CHECKING + +from .handler import Handler, Notification + +if TYPE_CHECKING: + from .state import AbstractStateManager + from ..bot import GreenAPI + + +MIN_SECONDS = 946_684_800 # 2000-01-01 UTC + +MAX_SECONDS = 4_102_444_800 # 2100-01-01 UTC, exclusive + + +def _optional_text(value: Any) -> Optional[str]: + return value if isinstance(value, str) and value else None + + +def _unix_seconds(value: Any) -> Optional[float]: + """Accept seconds, milliseconds or microseconds for dates in 2000–2100.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + + try: + value = float(value) + except OverflowError: + return None + + if not isfinite(value) or value <= 0: + return None + + # Keep the supported units and plausible range explicit. A larger value is + # not automatically another time unit (for example, nanoseconds). + if MIN_SECONDS <= value < MAX_SECONDS: + return value + + if 1_000 * MIN_SECONDS <= value < 1_000 * MAX_SECONDS: + return value / 1_000 + + if 1_000_000 * MIN_SECONDS <= value < 1_000_000 * MAX_SECONDS: + return value / 1_000_000 + + return None + + +@dataclass(frozen=True) +class IncomingCall: + sender: Optional[str] + call_id: Optional[str] + status: Optional[str] + timestamp: Optional[float] + raw_event: Optional[Dict[str, Any]] = field(default=None, compare=False, repr=False) + + @property + def chat_id(self) -> Optional[str]: + """The caller's address is the chat address for a direct call.""" + return self.sender + + @classmethod + def from_event(cls, event: Dict[str, Any]) -> "IncomingCall": + return cls( + sender=_optional_text(event.get("from")), + call_id=_optional_text(event.get("idMessage")), + status=_optional_text(event.get("status")), + timestamp=_unix_seconds(event.get("timestamp")), + raw_event=event, + ) + + +class IncomingCallNotification(Notification): + """Call fields plus the original webhook and the user's state manager.""" + + call: IncomingCall + + def __init__( + self, event: Dict[str, Any], api: "GreenAPI", + state_manager: "AbstractStateManager", + ) -> None: + super().__init__(event, api, state_manager) + self.call = IncomingCall.from_event(event) + + @property + def raw_event(self) -> dict: + return self.event + + def get_chat(self) -> Optional[str]: + return self.call.chat_id + + def get_sender(self) -> Optional[str]: + return self.call.sender + + def get_id_message(self) -> Optional[str]: + return self.call.call_id + + def get_message_text(self) -> None: + return None + + def get_message_data(self) -> None: + return None + + +class IncomingCallHandler(Handler[IncomingCallNotification]): + """Keep status filtering local to incoming calls.""" + + def check_event(self, notification: IncomingCallNotification) -> bool: + accepted_statuses = self.filters.get("status") + + if accepted_statuses is not None: + if isinstance(accepted_statuses, str): + accepted_statuses = [accepted_statuses] + + if notification.call.status not in accepted_statuses: + return False + + return super().check_event(notification) + + +__all__ = ["IncomingCall", "IncomingCallNotification", "IncomingCallHandler"] diff --git a/whatsapp_chatbot_python/manager/observer.py b/whatsapp_chatbot_python/manager/observer.py index 13fda71..3640191 100644 --- a/whatsapp_chatbot_python/manager/observer.py +++ b/whatsapp_chatbot_python/manager/observer.py @@ -1,26 +1,34 @@ import logging from abc import ABC, abstractmethod -from typing import Any, Callable, List, TYPE_CHECKING +from typing import Any, Callable, Generic, List, Optional, TYPE_CHECKING, Type, TypeVar -from .handler import AbstractHandler, Handler, HandlerType +from .handler import AbstractHandler, Handler, Notification, NotificationT +from .incoming_call import IncomingCallHandler, IncomingCallNotification from .state import AbstractStateManager, StateManager if TYPE_CHECKING: from .router import Router +ReturnT = TypeVar("ReturnT") -class AbstractObserver(ABC): + +class AbstractObserver(ABC, Generic[NotificationT]): + router: "Router" event: dict - handlers: List[AbstractHandler] + handlers: List[AbstractHandler[NotificationT]] state_manager: AbstractStateManager + @abstractmethod + def make_notification(self) -> NotificationT: + pass + def update_event(self, event: dict) -> None: self.event = event self.propagate_event() @abstractmethod - def add_handler(self, handler: HandlerType, **filters: Any) -> None: + def add_handler(self, handler: Callable[[NotificationT], Any], **filters: Any) -> None: pass @abstractmethod @@ -28,15 +36,21 @@ def propagate_event(self) -> None: pass -class Observer(AbstractObserver): - def __init__(self, router: "Router"): +class _TypedObserver(AbstractObserver[NotificationT], Generic[NotificationT]): + def __init__( + self, router: "Router", notification_type: Type[NotificationT], + state_manager: Optional[AbstractStateManager] = None, + ): self.router = router - self.event = {} - self.handlers = [] - self.state_manager = StateManager() + self.handlers: List[AbstractHandler[NotificationT]] = [] + self.state_manager = state_manager if state_manager is not None else StateManager() + self._notification_type = notification_type + + def make_notification(self) -> NotificationT: + return self._notification_type(self.event, self.router.api, self.state_manager) - def add_handler(self, handler: HandlerType, **filters: Any) -> None: + def add_handler(self, handler: Callable[[NotificationT], Any], **filters: Any) -> None: self.handlers.append(Handler(handler, **filters)) def propagate_event(self) -> None: @@ -65,17 +79,27 @@ def propagate_event(self) -> None: ) ) - def __call__(self, **filters: Any) -> Callable[[HandlerType], HandlerType]: - def wrapper(handler: HandlerType) -> HandlerType: + def __call__( + self, **filters: Any, + ) -> Callable[[Callable[[NotificationT], ReturnT]], Callable[[NotificationT], ReturnT]]: + def wrapper( + handler: Callable[[NotificationT], ReturnT], + ) -> Callable[[NotificationT], ReturnT]: self.add_handler(handler, **filters) - return handler return wrapper +class Observer(_TypedObserver[Notification]): + def __init__( + self, router: "Router", state_manager: Optional[AbstractStateManager] = None, + ): + super().__init__(router, Notification, state_manager) + + class ButtonObserver(Observer): - def add_handler(self, handler: HandlerType, **filters: Any) -> None: + def add_handler(self, handler: Callable[[Notification], Any], **filters: Any) -> None: message_types = [ "buttonsResponseMessage", "templateButtonsReplyMessage", @@ -89,21 +113,60 @@ def add_handler(self, handler: HandlerType, **filters: Any) -> None: self.router.message.add_handler(handler, **filters) +class IncomingCallObserver(_TypedObserver[IncomingCallNotification]): + """Incoming-call notifications share message state for the same sender.""" + + _allowed_filters = frozenset({"status", "from_sender", "from_chat", "state"}) + + def __init__( + self, router: "Router", state_manager: Optional[AbstractStateManager] = None, + ): + super().__init__( + router, IncomingCallNotification, state_manager, + ) + + def add_handler( + self, handler: Callable[[IncomingCallNotification], Any], **filters: Any, + ) -> None: + unsupported = set(filters) - self._allowed_filters + if unsupported: + raise ValueError( + "Unsupported incoming-call filters: " + ", ".join(sorted(unsupported)) + ) + + if "status" in filters: + status = filters["status"] + + if isinstance(status, str): + statuses = [status] + elif isinstance(status, (list, tuple)): + statuses = list(status) + else: + statuses = [] + + if not statuses or not all(isinstance(value, str) and value for value in statuses): + raise ValueError( + "Incoming-call status must be a string or a nonempty list of strings" + ) + + self.handlers.append(IncomingCallHandler(handler, **filters)) + + class PollObserver(Observer): - def add_handler(self, handler: HandlerType, **filters: Any) -> None: + def add_handler(self, handler: Callable[[Notification], Any], **filters: Any) -> None: filters["type_message"] = "pollMessage" self.router.message.add_handler(handler, **filters) class PollUpdateObserver(Observer): - def add_handler(self, handler: HandlerType, **filters: Any) -> None: + def add_handler(self, handler: Callable[[Notification], Any], **filters: Any) -> None: filters["type_message"] = "pollUpdateMessage" self.router.message.add_handler(handler, **filters) def add_handler_with_stanza( - self, handler: HandlerType, stanza: str, **filters: Any + self, handler: Callable[[Notification], Any], stanza: str, **filters: Any ) -> None: filters.update({ "stanza": stanza, @@ -116,6 +179,7 @@ def add_handler_with_stanza( __all__ = [ "AbstractObserver", "ButtonObserver", + "IncomingCallObserver", "Observer", "PollObserver", "PollUpdateObserver" diff --git a/whatsapp_chatbot_python/manager/router.py b/whatsapp_chatbot_python/manager/router.py index 4604d2c..c6e912c 100644 --- a/whatsapp_chatbot_python/manager/router.py +++ b/whatsapp_chatbot_python/manager/router.py @@ -1,10 +1,11 @@ import json import logging -from typing import Dict, TYPE_CHECKING +from typing import Any, Dict, TYPE_CHECKING from .observer import ( AbstractObserver, ButtonObserver, + IncomingCallObserver, Observer, PollObserver, PollUpdateObserver @@ -19,18 +20,21 @@ def __init__(self, api: "GreenAPI", logger: logging.Logger): self.api = api self.logger = logger - self.message: AbstractObserver = Observer(self) - self.outgoing_message: AbstractObserver = Observer(self) - self.outgoing_api_message: AbstractObserver = Observer(self) - self.outgoing_message_status: AbstractObserver = Observer(self) - self.incoming_call: AbstractObserver = Observer(self) + self.message: Observer = Observer(self) + self.outgoing_message: Observer = Observer(self) + self.outgoing_api_message: Observer = Observer(self) + self.outgoing_message_status: Observer = Observer(self) - self.buttons: AbstractObserver = ButtonObserver(self) + self.incoming_call: IncomingCallObserver = IncomingCallObserver( + self, state_manager=self.message.state_manager, + ) + + self.buttons: ButtonObserver = ButtonObserver(self) self.poll_message: PollObserver = PollObserver(self) self.poll_update_message: PollUpdateObserver = PollUpdateObserver(self) - self.observers: Dict[str, AbstractObserver] = { + self.observers: Dict[str, AbstractObserver[Any]] = { "incomingMessageReceived": self.message, "outgoingMessageReceived": self.outgoing_message, "outgoingAPIMessageReceived": self.outgoing_api_message, From 6c9bb2b5287297362c2b0c75693a383de0e688d4 Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Tue, 20 Oct 2026 08:57:14 +0500 Subject: [PATCH 3/4] fix: The issue with LIDs has been fixed --- tests/calls/test_call_service.py | 42 +++++++++++++++++++----- whatsapp_chatbot_python/calls/service.py | 25 +++++++------- 2 files changed, 46 insertions(+), 21 deletions(-) diff --git a/tests/calls/test_call_service.py b/tests/calls/test_call_service.py index 4e683f6..cc6637b 100644 --- a/tests/calls/test_call_service.py +++ b/tests/calls/test_call_service.py @@ -460,6 +460,27 @@ async def test_incoming_accepts_matching_call_before_audio(self): self.assertEqual(FakeVoice.instances[0].options["greeting_text"], session.greeting_text) self.assertEqual(FakeVoice.instances[0].greetings, 1) + async def test_incoming_accepts_same_call_with_pn_and_lid_addresses(self): + FakeClient.initial_state = "inc-call" + + for webhook_sender, rtc_sender in ( + ("sender@c.us", "caller@lid"), + ("sender@lid", "sender@lid"), + ("sender@lid", "caller@c.us"), + ): + with self.subTest(webhook_sender=webhook_sender, rtc_sender=rtc_sender): + FakeClient.instances.clear() + FakeVoice.instances.clear() + FakeClient.initial_info = CallInfo("call-1", rtc_sender, None) + session, transition = self.make_incoming_session() + session.sender_id = webhook_sender + + await self.make_service().execute(session, transition) + + self.assertEqual(FakeClient.instances[0].actions[:3], ["open", "accept", "bridge"]) + self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertIsNotNone(session.answered_at) + async def test_voice_service_uses_explicit_credentials_despite_environment(self): import os @@ -685,7 +706,7 @@ async def failed_accept(client): raise RuntimeError("response lost") async def server_state(client): - return CallStateDetail("on-call", info=client.initial_info) + return CallStateDetail("on-call", info=CallInfo("call-1", "caller@lid", None)) with patch.object(FakeClient, "acceptAsync", failed_accept), \ patch.object(FakeClient, "getStateAsync", server_state): @@ -795,7 +816,7 @@ async def test_incoming_caller_left_signal_wakes_active_call(self): async def test_incoming_bridge_failure_hangs_up_matching_call(self): FakeClient.initial_state = "inc-call" - FakeClient.initial_info = CallInfo("call-1", "sender@c.us", None) + FakeClient.initial_info = CallInfo("call-1", "caller@lid", None) FakeClient.bridge_error = True FakeClient.remote_hangup = False session, transition = self.make_incoming_session() @@ -883,7 +904,7 @@ async def pending_accept(client): self.assertEqual(session.end_reason, CallEndReason.SHUTDOWN) self.assertEqual(FakeClient.instances[0].actions, ["open", "accept", "reject"]) - async def test_reject_incoming_checks_both_call_id_and_sender(self): + async def test_reject_incoming_matches_call_id_across_pn_and_lid(self): api = Mock() api.voip.getStateAsync = AsyncMock(return_value=CallStateDetail( "inc-call", info=CallInfo("new-call", "sender@c.us", None), @@ -895,17 +916,20 @@ async def test_reject_incoming_checks_both_call_id_and_sender(self): api.voip.rejectAsync.assert_not_awaited() api.voip.getStateAsync.return_value = CallStateDetail( - "inc-call", info=CallInfo("call-1", "other@c.us", None), + "inc-call", info=CallInfo("call-1", "caller@lid", None), ) - self.assertFalse(await service.reject_incoming("call-1", "sender@c.us")) - api.voip.rejectAsync.assert_not_awaited() + self.assertTrue(await service.reject_incoming("call-1", "sender@c.us")) + api.voip.rejectAsync.assert_awaited_once() api.voip.getStateAsync.return_value = CallStateDetail( - "inc-call", info=CallInfo("call-1", "sender@c.us", None), + "inc-call", info=CallInfo("call-2", "caller@lid", None), ) - self.assertTrue(await service.reject_incoming("call-1", "sender@c.us")) + self.assertFalse(await service.reject_incoming("call-1", "sender@c.us")) + api.voip.rejectAsync.assert_awaited_once() + + self.assertFalse(await service.reject_incoming("", "sender@c.us")) api.voip.rejectAsync.assert_awaited_once() - self.assertEqual(api.session.close.call_count, 3) + self.assertEqual(api.session.close.call_count, 4) async def test_stop_before_execute_does_not_start_a_call(self): session, transition = self.make_session() diff --git a/whatsapp_chatbot_python/calls/service.py b/whatsapp_chatbot_python/calls/service.py index e748792..8935cf3 100644 --- a/whatsapp_chatbot_python/calls/service.py +++ b/whatsapp_chatbot_python/calls/service.py @@ -153,14 +153,14 @@ def signal() -> None: return True async def reject_incoming(self, call_id: str, sender_id: str) -> bool: - """Reject a waiting offer only while it still matches the live call.""" + """Reject a waiting offer with this call ID on the configured instance.""" api = GreenAPI(self._id_instance, self._api_token_instance, host=self._api_url) try: state = await asyncio.wait_for(api.voip.getStateAsync(), MATCHING_CALL_TIMEOUT_SECONDS) - if not self._matches_incoming(state, call_id, sender_id, "inc-call"): + if not self._matches_incoming(state, call_id, "inc-call"): return False await asyncio.wait_for(api.voip.rejectAsync(), MATCHING_CALL_TIMEOUT_SECONDS) @@ -424,7 +424,7 @@ def _on_state(execution: _CallExecution, state) -> None: if execution.session.direction == CallDirection.INCOMING: if state.state == "on-call" and WhatsAppCallService._matches_incoming( - state, execution.session.call_id, execution.session.sender_id, "on-call", + state, execution.session.call_id, "on-call", ): if execution.incoming_answered_at is None: execution.incoming_answered_at = monotonic() @@ -444,7 +444,7 @@ def _on_state(execution: _CallExecution, state) -> None: execution.incoming_end_event.set() if state.state in {"inc-call", "on-call"} and not WhatsAppCallService._matches_incoming( - state, execution.session.call_id, execution.session.sender_id, state.state, + state, execution.session.call_id, state.state, ): execution.incoming_ended = True execution.incoming_left_at = monotonic() @@ -458,12 +458,13 @@ def _on_state(execution: _CallExecution, state) -> None: execution.runtime.events.put_nowait(RuntimeEvent.server_state(state.state, state.reason)) @staticmethod - def _matches_incoming(state: ApiCallState | None, call_id: str | None, sender_id: str, kind: str) -> bool: + def _matches_incoming(state: ApiCallState | None, call_id: str | None, kind: str) -> bool: info = getattr(state, "info", None) return ( - getattr(state, "state", None) == kind and info is not None - and info.id == call_id and info.wid == sender_id + isinstance(call_id, str) and bool(call_id) + and getattr(state, "state", None) == kind and info is not None + and info.id == call_id ) @staticmethod @@ -581,7 +582,7 @@ async def _run_incoming(self, execution: _CallExecution) -> None: execution.transition(session, CallEvent.CALLER_LEFT) return - if not self._matches_incoming(state, session.call_id, session.sender_id, "inc-call"): + if not self._matches_incoming(state, session.call_id, "inc-call"): execution.transition(session, CallEvent.CALLER_LEFT) return @@ -676,7 +677,7 @@ async def _reconcile_failed_incoming_accept( except Exception: raise accept_error - if self._matches_incoming(state, session.call_id, session.sender_id, "on-call"): + if self._matches_incoming(state, session.call_id, "on-call"): # The REST response failed, but the server confirms this call was answered. execution.incoming_answered_at = monotonic() @@ -684,7 +685,7 @@ async def _reconcile_failed_incoming_accept( execution.runtime.remote_accepted(session.bridge_ready) return True - if not self._matches_incoming(state, session.call_id, session.sender_id, "inc-call"): + if not self._matches_incoming(state, session.call_id, "inc-call"): execution.transition(session, CallEvent.CALLER_LEFT) return False @@ -952,11 +953,11 @@ async def _finish_matching_incoming(api: GreenAPI | None, session: CallSession) state = await asyncio.wait_for(api.voip.getStateAsync(), MATCHING_CALL_TIMEOUT_SECONDS) if WhatsAppCallService._matches_incoming( - state, session.call_id, session.sender_id, "inc-call", + state, session.call_id, "inc-call", ): await asyncio.wait_for(api.voip.rejectAsync(), MATCHING_CALL_TIMEOUT_SECONDS) elif WhatsAppCallService._matches_incoming( - state, session.call_id, session.sender_id, "on-call", + state, session.call_id, "on-call", ): await asyncio.wait_for(api.voip.hangUpAsync(), MATCHING_CALL_TIMEOUT_SECONDS) From c30c988a2fee891847b992a040915b806a7aa6b9 Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Tue, 20 Oct 2026 09:31:02 +0500 Subject: [PATCH 4/4] fix: The issue with the MP3 audio quality has been fixed --- tests/calls/test_call_recording.py | 47 ++++++++++++++++++- tests/calls/test_realtime_voice.py | 6 +-- .../calls/realtime_voice.py | 8 +++- whatsapp_chatbot_python/calls/recording.py | 42 ++++++++++++++--- 4 files changed, 91 insertions(+), 12 deletions(-) diff --git a/tests/calls/test_call_recording.py b/tests/calls/test_call_recording.py index 2a589c8..24b1098 100644 --- a/tests/calls/test_call_recording.py +++ b/tests/calls/test_call_recording.py @@ -1,7 +1,7 @@ from array import array from unittest.mock import patch from queue import Full -from whatsapp_chatbot_python.calls.recording import CallRecorder, RATE +from whatsapp_chatbot_python.calls.recording import CallRecorder, MP3_BIT_RATE, RATE import logging import unittest import av @@ -40,6 +40,7 @@ def test_mixes_both_sides_and_keeps_pause_and_reconnect(self) -> None: try: with av.open(str(path)) as container: + self.assertEqual(container.streams.audio[0].bit_rate, MP3_BIT_RATE) frames = [frame for packet in container.demux(audio=0) for frame in packet.decode()] converted = av.AudioResampler(format="s16", layout="mono", rate=RATE) @@ -61,6 +62,50 @@ def level(start, end): finally: path.unlink(missing_ok=True) + def test_frame_jitter_does_not_split_audio_but_media_gap_is_preserved(self) -> None: + clock = [0.0] + recorder = CallRecorder(logging.getLogger("recording-test")) + tracks = {} + + def capture(_path): + for side in ("caller", "bot"): + stream = getattr(recorder, f"_{side}") + stream.seek(0) + tracks[side] = array("h") + tracks[side].frombytes(stream.read()) + + with patch("whatsapp_chatbot_python.calls.recording.monotonic", side_effect=lambda: clock[0]), \ + patch.object(recorder, "_encode", side_effect=capture): + recorder.set_generation(1) + recorder.start() + recorder.add_bot(pcm(3000, 480), 1) + clock[0] = 0.045 # A delayed callback must not insert 25 ms of silence. + recorder.add_bot(pcm(4000, 480), 1) + + clock[0] = 0.02 + recorder.add_caller(pcm(1000, 480), 1, media_time=10.0) + clock[0] = 0.065 + recorder.add_caller(pcm(2000, 480), 1, media_time=10.02) + clock[0] = 0.075 + recorder.add_caller(pcm(3000, 480), 1, media_time=10.04) + clock[0] = 0.15 + recorder.add_caller(pcm(4000, 480), 1, media_time=10.10) + + recorder.set_generation(2) + clock[0] = 0.22 + recorder.add_caller(pcm(5000, 480), 2, media_time=0.0) + path = recorder.finish() + + self.assertIsNotNone(path) + try: + self.assertEqual(tracks["bot"], array("h", [3000] * 480 + [4000] * 480)) + self.assertEqual(tracks["caller"][:1440], array("h", [1000] * 480 + [2000] * 480 + [3000] * 480)) + self.assertEqual(tracks["caller"][1440:2400], array("h", [0] * 960)) + self.assertEqual(tracks["caller"][2400:2880], array("h", [4000] * 480)) + self.assertEqual(tracks["caller"][4800:5280], array("h", [5000] * 480)) + finally: + path.unlink(missing_ok=True) + def test_no_audio_or_queue_overflow_disables_recording(self) -> None: recorder = CallRecorder(logging.getLogger("recording-test")) diff --git a/tests/calls/test_realtime_voice.py b/tests/calls/test_realtime_voice.py index 91b78fb..8c2cada 100644 --- a/tests/calls/test_realtime_voice.py +++ b/tests/calls/test_realtime_voice.py @@ -422,9 +422,9 @@ def __init__(self): def set_generation(self, generation): self.generation = generation - def add_caller(self, pcm, generation): + def add_caller(self, pcm, generation, *, media_time=None): if generation == self.generation: - self.caller.append(pcm) + self.caller.append((pcm, media_time)) def add_bot(self, pcm, generation): if generation == self.generation: @@ -458,7 +458,7 @@ def add_bot(self, pcm, generation): await sink.close() voice._queue_audio("item-1", bytes([1, 2]) * 480) await old_track.recv() - self.assertEqual(recorder.caller, [bytes(FRAME_BYTES)]) + self.assertEqual(recorder.caller, [(bytes(FRAME_BYTES), 0.0)]) self.assertEqual(recorder.bot, [bytes([1, 2]) * 480]) await voice.new_output_track() await old_track.recv() diff --git a/whatsapp_chatbot_python/calls/realtime_voice.py b/whatsapp_chatbot_python/calls/realtime_voice.py index 630530d..7fbfc4f 100644 --- a/whatsapp_chatbot_python/calls/realtime_voice.py +++ b/whatsapp_chatbot_python/calls/realtime_voice.py @@ -108,7 +108,13 @@ async def _consume(self, track: MediaStreamTrack) -> None: pcm = bytes(converted.planes[0])[:converted.samples * 2] if self._voice.recorder is not None: - self._voice.recorder.add_caller(pcm, self._generation) + media_time = ( + float(converted.pts * converted.time_base) + if converted.pts is not None and converted.time_base is not None + else None + ) + + self._voice.recorder.add_caller(pcm, self._generation, media_time=media_time) await self._voice.append_input(pcm) except MediaStreamError: diff --git a/whatsapp_chatbot_python/calls/recording.py b/whatsapp_chatbot_python/calls/recording.py index f44fa12..84b6168 100644 --- a/whatsapp_chatbot_python/calls/recording.py +++ b/whatsapp_chatbot_python/calls/recording.py @@ -2,6 +2,7 @@ from __future__ import annotations from array import array +from math import isfinite from pathlib import Path from queue import Full, Queue from tempfile import TemporaryFile, NamedTemporaryFile @@ -19,6 +20,8 @@ MIX_SAMPLES = 24 * 1000 +MP3_BIT_RATE = 96 * 1000 + class CallRecorder: def __init__(self, logger: logging.Logger) -> None: @@ -31,6 +34,8 @@ def __init__(self, logger: logging.Logger) -> None: self._caller = TemporaryFile() self._bot = TemporaryFile() self._lengths = {"caller": 0, "bot": 0} + self._next_offsets: dict[str, int | None] = {"caller": None, "bot": None} + self._caller_media_anchor: tuple[float, int] | None = None self._has_audio = False def start(self) -> None: @@ -41,15 +46,18 @@ def start(self) -> None: self._thread.start() def set_generation(self, generation: int) -> None: - self._generation = generation + if generation != self._generation: + self._generation = generation + self._next_offsets = {"caller": None, "bot": None} + self._caller_media_anchor = None - def add_caller(self, pcm: bytes, generation: int) -> None: - self._add("caller", pcm, generation) + def add_caller(self, pcm: bytes, generation: int, *, media_time: float | None = None) -> None: + self._add("caller", pcm, generation, media_time=media_time) def add_bot(self, pcm: bytes, generation: int) -> None: self._add("bot", pcm, generation) - def _add(self, side: str, pcm: bytes, generation: int) -> None: + def _add(self, side: str, pcm: bytes, generation: int, *, media_time: float | None = None) -> None: if self._started_at is None or self._failed.is_set() or generation != self._generation or not pcm: return @@ -57,17 +65,36 @@ def _add(self, side: str, pcm: bytes, generation: int) -> None: self._failed.set() return - # Incoming frames arrive after capture; place their start at the beginning - # of the received interval. The outgoing frame starts when recv() yields it. + samples = len(pcm) // 2 + + # Arrival time anchors each track. Later chunks follow sample counts so + # scheduling jitter cannot create holes or overwrite adjacent samples. elapsed = monotonic() - self._started_at if side == "caller": - elapsed -= len(pcm) / (2 * RATE) + elapsed -= samples / RATE offset = max(0, round(elapsed * RATE)) + next_offset = self._next_offsets[side] + + if next_offset is not None: + offset = next_offset + + anchor = self._caller_media_anchor + + if side == "caller" and media_time is not None and isfinite(media_time): + if anchor is None: + anchor = (media_time, offset) + else: + offset = max(offset, anchor[1] + round((media_time - anchor[0]) * RATE)) try: self._queue.put_nowait((side, offset, pcm)) + + self._next_offsets[side] = offset + samples + + if side == "caller": + self._caller_media_anchor = anchor except Full: self._failed.set() self._logger.warning("VoIP recording queue overflow") @@ -130,6 +157,7 @@ def _encode(self, path: Path) -> None: with av.open(str(path), "w") as container: stream = container.add_stream("libmp3lame", rate=RATE) stream.layout = "mono" + stream.bit_rate = MP3_BIT_RATE for _ in range(0, length, MIX_SAMPLES): count = min(MIX_SAMPLES, length - _)