From 1a91dc44891c57bb99dc993334ca06d41bd5a09a Mon Sep 17 00:00:00 2001 From: didar Date: Sun, 23 Aug 2026 12:55:25 +0500 Subject: [PATCH] feat: add tenant ID handling for sales voice and telegram sync requests --- services/ai_voice_runtime_service/app.py | 26 ++++++++++++--- .../ai_voice_runtime_service/audiosocket.py | 33 +++++++++++++++++++ .../ai_voice_runtime_service/providers/asr.py | 20 +++++++---- services/telegram_adapter_service/app.py | 7 ++++ 4 files changed, 76 insertions(+), 10 deletions(-) diff --git a/services/ai_voice_runtime_service/app.py b/services/ai_voice_runtime_service/app.py index c7c1b90..edfc7f3 100644 --- a/services/ai_voice_runtime_service/app.py +++ b/services/ai_voice_runtime_service/app.py @@ -92,6 +92,10 @@ def _sales_service_url() -> str: return os.getenv("SALES_SERVICE_URL", "http://localhost:8020").rstrip("/") +def _sales_voice_sync_tenant_id() -> str | None: + return os.getenv("SALES_VOICE_SYNC_TENANT_ID", "").strip() or None + + def _asr_provider_name() -> str: return os.getenv("AI_VOICE_ASR_PROVIDER", "yandex").strip() or "yandex" @@ -188,13 +192,14 @@ def _voice_v2_emotive_ack_ru_only() -> bool: return _bool_env("AI_VOICE_V2_EMOTIVE_ACK_RU_ONLY", True) -def _service_headers() -> dict[str, str]: +def _service_headers(*, tenant_id: str | None = None) -> dict[str, str]: token = issue_app_token( subject="svc:ai-voice-runtime", username="ai-voice-runtime", role="admin", auth_source="service", provider="ai-voice-runtime", + tenant_id=tenant_id, ttl_seconds=300, ) return {"Authorization": f"Bearer {token}"} @@ -219,13 +224,20 @@ def _retry_db_write(fn, *, attempts: int = 6, base_delay_seconds: float = 0.05): raise RuntimeError("DB write retry exhausted without exception") -def _request(method: str, url: str, *, payload: dict[str, Any] | None = None, timeout: float = 10.0) -> dict[str, Any]: +def _request( + method: str, + url: str, + *, + payload: dict[str, Any] | None = None, + timeout: float = 10.0, + tenant_id: str | None = None, +) -> dict[str, Any]: with httpx.Client(timeout=timeout) as client: response = client.request( method, url, json=payload, - headers=_service_headers(), + headers=_service_headers(tenant_id=tenant_id), ) response.raise_for_status() return response.json() @@ -276,7 +288,13 @@ def _interaction_request(method: str, path: str, *, payload: dict[str, Any] | No def _sales_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 8.0) -> dict[str, Any]: - return _request(method, f"{_sales_service_url()}{path}", payload=payload, timeout=timeout) + return _request( + method, + f"{_sales_service_url()}{path}", + payload=payload, + timeout=timeout, + tenant_id=_sales_voice_sync_tenant_id(), + ) def _resolved_voice_language(language_hint: str | None, fallback: str | None = None) -> str: diff --git a/services/ai_voice_runtime_service/audiosocket.py b/services/ai_voice_runtime_service/audiosocket.py index d65d76a..6c2ef30 100644 --- a/services/ai_voice_runtime_service/audiosocket.py +++ b/services/ai_voice_runtime_service/audiosocket.py @@ -82,6 +82,39 @@ def resample_pcm16le( return converted +def resample_pcm16le_stateful( + pcm_bytes: bytes, + *, + input_rate_hz: int, + output_rate_hz: int, + state: object | None, + channels: int = 1, + sample_width_bytes: int = 2, +) -> tuple[bytes, object | None]: + """Like resample_pcm16le, but carries audioop.ratecv state across calls. + + Resampling each chunk of a continuous stream independently (state=None every + call) introduces small discontinuities at every chunk boundary. For a live + audio stream fed to an ASR backend in small batches, that repeated boundary + noise can measurably hurt transcription quality, so callers that push a + stream in chunks should carry state across calls with this function instead. + """ + if not pcm_bytes or input_rate_hz == output_rate_hz: + return pcm_bytes, state + mono_bytes = pcm_bytes + if channels == 2: + mono_bytes = audioop.tomono(pcm_bytes, sample_width_bytes, 0.5, 0.5) + converted, new_state = audioop.ratecv( + mono_bytes, + sample_width_bytes, + 1, + input_rate_hz, + output_rate_hz, + state, + ) + return converted, new_state + + def chunk_audio(pcm_bytes: bytes, *, frame_bytes: int) -> list[bytes]: if frame_bytes <= 0: raise ValueError("frame_bytes must be positive") diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 4890945..1ceac84 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -16,7 +16,11 @@ from dataclasses import dataclass, field import httpx -from services.ai_voice_runtime_service.audiosocket import pcm16le_to_wav_bytes, resample_pcm16le +from services.ai_voice_runtime_service.audiosocket import ( + pcm16le_to_wav_bytes, + resample_pcm16le, + resample_pcm16le_stateful, +) from services.shared.audioop_compat import audioop from services.shared.security import issue_app_token @@ -531,6 +535,8 @@ class _ElevenLabsRealtimeStreamState: final_event: threading.Event = field(default_factory=threading.Event) lock: threading.Lock = field(default_factory=threading.Lock) send_lock: threading.Lock = field(default_factory=threading.Lock) + resample_state: object | None = None + resample_lock: threading.Lock = field(default_factory=threading.Lock) class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): @@ -823,11 +829,13 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): state = self._state(stream_id) pcm_bytes = pcm_8k_chunk if self._sample_rate_hz != 8000: - pcm_bytes = resample_pcm16le( - pcm_8k_chunk, - input_rate_hz=8000, - output_rate_hz=self._sample_rate_hz, - ) + with state.resample_lock: + pcm_bytes, state.resample_state = resample_pcm16le_stateful( + pcm_8k_chunk, + input_rate_hz=8000, + output_rate_hz=self._sample_rate_hz, + state=state.resample_state, + ) self._send_json( state, { diff --git a/services/telegram_adapter_service/app.py b/services/telegram_adapter_service/app.py index 1549540..167a0fd 100644 --- a/services/telegram_adapter_service/app.py +++ b/services/telegram_adapter_service/app.py @@ -235,6 +235,10 @@ def _sales_service_url() -> str: return os.getenv("SALES_SERVICE_URL", "http://localhost:8020").rstrip("/") +def _sales_telegram_sync_tenant_id() -> str | None: + return os.getenv("SALES_TELEGRAM_SYNC_TENANT_ID", "").strip() or None + + def _ai_telegram_enabled() -> bool: return _bool_env("AI_TELEGRAM_ENABLED", False) @@ -256,6 +260,7 @@ def _internal_service_headers( subject: str, username: str, provider: str, + tenant_id: str | None = None, ) -> dict[str, str]: token = issue_app_token( subject=subject, @@ -263,6 +268,7 @@ def _internal_service_headers( role="admin", auth_source="service", provider=provider, + tenant_id=tenant_id, ttl_seconds=300, ) return {"Authorization": f"Bearer {token}"} @@ -290,6 +296,7 @@ def _sales_sync_request(payload: dict[str, Any]) -> None: subject="svc:telegram-adapter", username="telegram-adapter", provider="telegram-adapter", + tenant_id=_sales_telegram_sync_tenant_id(), ), ) response.raise_for_status()