feat: add tenant ID handling for sales voice and telegram sync requests

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:

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

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,
             {

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