diff --git a/.env.example b/.env.example index cf39859..b5a59ad 100644 --- a/.env.example +++ b/.env.example @@ -60,6 +60,12 @@ AI_VOICE_V2_ACK_MODE=immediate_short AI_VOICE_V2_DUPLEX_ENABLED=1 AI_VOICE_V2_STREAMING_ASR_BACKEND=local_sidecar AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://127.0.0.1:8021 +AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS=4 +AI_VOICE_V2_STREAMING_ASR_MODEL=base +AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE=int8 +AI_VOICE_V2_STREAMING_ASR_DEVICE=cpu +AI_VOICE_V2_STREAMING_ASR_CACHE_DIR=.models/faster_whisper +AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES=ru AI_VOICE_V2_PREBAKED_ACK_ENABLED=1 AI_VOICE_V2_PREBAKED_ACK_DIR=.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS=1 diff --git a/.env.production.template b/.env.production.template index 159ac5b..4c67e74 100644 --- a/.env.production.template +++ b/.env.production.template @@ -54,7 +54,13 @@ AI_VOICE_V2_QUEUE_CODES=voice_lab_ai AI_VOICE_V2_ACK_MODE=immediate_short AI_VOICE_V2_DUPLEX_ENABLED=1 AI_VOICE_V2_STREAMING_ASR_BACKEND=local_sidecar -AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://127.0.0.1:8021 +AI_VOICE_V2_STREAMING_ASR_BASE_URL=http://streaming-asr-sidecar:8021 +AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS=4 +AI_VOICE_V2_STREAMING_ASR_MODEL=base +AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE=int8 +AI_VOICE_V2_STREAMING_ASR_DEVICE=cpu +AI_VOICE_V2_STREAMING_ASR_CACHE_DIR=/models/faster_whisper +AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES=ru AI_VOICE_V2_PREBAKED_ACK_ENABLED=1 AI_VOICE_V2_PREBAKED_ACK_DIR=/app/.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS=1 diff --git a/.gitignore b/.gitignore index b438a13..abbef3d 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,7 @@ .codex_tmp/ .server_backups/ .tmp/ +.models/ test-results/ .testdata/ backups/ diff --git a/deployment/docker-compose.parallel.server.yml b/deployment/docker-compose.parallel.server.yml index 042f834..8ecc862 100644 --- a/deployment/docker-compose.parallel.server.yml +++ b/deployment/docker-compose.parallel.server.yml @@ -22,6 +22,8 @@ x-app-env: &app_env KB_SERVICE_URL: http://kb-service:8000 REPORTING_SERVICE_URL: http://reporting-service:8000 SUPERVISOR_SERVICE_URL: http://supervisor-service:8000 + AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 + AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" x-service-defaults: &service_defaults build: @@ -123,8 +125,38 @@ services: <<: *app_env APP_MODULE: services.ai_orchestrator_service.app:app + streaming-asr-sidecar: + <<: *service_defaults + command: ["sh", "-c", "uvicorn ${APP_MODULE} --host 0.0.0.0 --port 8021"] + environment: + <<: *app_env + APP_MODULE: services.streaming_asr_sidecar_service.app:app + AI_VOICE_V2_STREAMING_ASR_MODEL: base + AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE: int8 + AI_VOICE_V2_STREAMING_ASR_DEVICE: cpu + AI_VOICE_V2_STREAMING_ASR_CACHE_DIR: /models/faster_whisper + AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES: ru + volumes: + - ../.data_pg_parallel:/app/.data_pg_parallel + - ../.models/faster_whisper:/models/faster_whisper + healthcheck: + test: + - CMD + - python + - -c + - | + from urllib.request import urlopen + import sys + sys.exit(0 if urlopen("http://127.0.0.1:8021/health", timeout=2).getcode() == 200 else 1) + interval: 15s + timeout: 5s + retries: 10 + start_period: 10s + ai-voice-runtime-service: <<: *service_defaults + depends_on: + - streaming-asr-sidecar environment: <<: *app_env APP_MODULE: services.ai_voice_runtime_service.app:app diff --git a/deployment/docker-compose.server.registry.yml b/deployment/docker-compose.server.registry.yml index 4e43417..66d3a34 100644 --- a/deployment/docker-compose.server.registry.yml +++ b/deployment/docker-compose.server.registry.yml @@ -22,6 +22,8 @@ x-app-env: &app_env KB_SERVICE_URL: http://kb-service:8000 REPORTING_SERVICE_URL: http://reporting-service:8000 SUPERVISOR_SERVICE_URL: http://supervisor-service:8000 + AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 + AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" x-service-defaults: &service_defaults image: ${APP_IMAGE:?Set APP_IMAGE in deployment/.env.images or shell env} @@ -117,8 +119,38 @@ services: <<: *app_env APP_MODULE: services.ai_orchestrator_service.app:app + streaming-asr-sidecar: + <<: *service_defaults + command: ["sh", "-c", "uvicorn ${APP_MODULE} --host 0.0.0.0 --port 8021"] + environment: + <<: *app_env + APP_MODULE: services.streaming_asr_sidecar_service.app:app + AI_VOICE_V2_STREAMING_ASR_MODEL: base + AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE: int8 + AI_VOICE_V2_STREAMING_ASR_DEVICE: cpu + AI_VOICE_V2_STREAMING_ASR_CACHE_DIR: /models/faster_whisper + AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES: ru + volumes: + - ../.data_local:/app/.data_local + - ../.models/faster_whisper:/models/faster_whisper + healthcheck: + test: + - CMD + - python + - -c + - | + from urllib.request import urlopen + import sys + sys.exit(0 if urlopen("http://127.0.0.1:8021/health", timeout=2).getcode() == 200 else 1) + interval: 15s + timeout: 5s + retries: 10 + start_period: 10s + ai-voice-runtime-service: <<: *service_defaults + depends_on: + - streaming-asr-sidecar environment: <<: *app_env APP_MODULE: services.ai_voice_runtime_service.app:app diff --git a/deployment/docker-compose.server.yml b/deployment/docker-compose.server.yml index 4cc9487..76b487c 100644 --- a/deployment/docker-compose.server.yml +++ b/deployment/docker-compose.server.yml @@ -22,6 +22,8 @@ x-app-env: &app_env KB_SERVICE_URL: http://kb-service:8000 REPORTING_SERVICE_URL: http://reporting-service:8000 SUPERVISOR_SERVICE_URL: http://supervisor-service:8000 + AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 + AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" x-service-defaults: &service_defaults build: @@ -119,8 +121,38 @@ services: <<: *app_env APP_MODULE: services.ai_orchestrator_service.app:app + streaming-asr-sidecar: + <<: *service_defaults + command: ["sh", "-c", "uvicorn ${APP_MODULE} --host 0.0.0.0 --port 8021"] + environment: + <<: *app_env + APP_MODULE: services.streaming_asr_sidecar_service.app:app + AI_VOICE_V2_STREAMING_ASR_MODEL: base + AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE: int8 + AI_VOICE_V2_STREAMING_ASR_DEVICE: cpu + AI_VOICE_V2_STREAMING_ASR_CACHE_DIR: /models/faster_whisper + AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES: ru + volumes: + - ../.data_local:/app/.data_local + - ../.models/faster_whisper:/models/faster_whisper + healthcheck: + test: + - CMD + - python + - -c + - | + from urllib.request import urlopen + import sys + sys.exit(0 if urlopen("http://127.0.0.1:8021/health", timeout=2).getcode() == 200 else 1) + interval: 15s + timeout: 5s + retries: 10 + start_period: 10s + ai-voice-runtime-service: <<: *service_defaults + depends_on: + - streaming-asr-sidecar environment: <<: *app_env APP_MODULE: services.ai_voice_runtime_service.app:app diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index b83d410..95f8e32 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -36,7 +36,8 @@ x-app-env: &app_env AI_VOICE_V2_ACK_MODE: immediate_short AI_VOICE_V2_DUPLEX_ENABLED: "1" AI_VOICE_V2_STREAMING_ASR_BACKEND: local_sidecar - AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://127.0.0.1:8021 + AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 + AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" AI_VOICE_V2_PREBAKED_ACK_ENABLED: "1" AI_VOICE_V2_PREBAKED_ACK_DIR: /app/.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS: "1" @@ -193,9 +194,40 @@ services: ports: - "8017:8000" - ai-voice-runtime-service: + streaming-asr-sidecar: build: .. depends_on: [postgres, rabbitmq] + command: ["sh", "-c", "uvicorn ${APP_MODULE} --host 0.0.0.0 --port 8021"] + environment: + <<: *app_env + APP_MODULE: services.streaming_asr_sidecar_service.app:app + AI_VOICE_V2_STREAMING_ASR_MODEL: base + AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE: int8 + AI_VOICE_V2_STREAMING_ASR_DEVICE: cpu + AI_VOICE_V2_STREAMING_ASR_CACHE_DIR: /models/faster_whisper + AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES: ru + volumes: + - ../.data:/app/.data + - ../.models/faster_whisper:/models/faster_whisper + healthcheck: + test: + - CMD + - python + - -c + - | + from urllib.request import urlopen + import sys + sys.exit(0 if urlopen("http://127.0.0.1:8021/health", timeout=2).getcode() == 200 else 1) + interval: 15s + timeout: 5s + retries: 10 + start_period: 10s + ports: + - "8021:8021" + + ai-voice-runtime-service: + build: .. + depends_on: [postgres, rabbitmq, streaming-asr-sidecar] environment: <<: *app_env APP_MODULE: services.ai_voice_runtime_service.app:app diff --git a/requirements.txt b/requirements.txt index 62abde2..b02c408 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,4 @@ psycopg[binary]==3.2.6 pika==1.3.2 paramiko==3.5.0 python-multipart==0.0.20 +faster-whisper==1.1.1 diff --git a/scripts/local_stack.py b/scripts/local_stack.py index 948aa31..20b88aa 100644 --- a/scripts/local_stack.py +++ b/scripts/local_stack.py @@ -88,6 +88,7 @@ def build_service_specs(env: dict[str, str] | None = None) -> list[dict[str, Any {"name": "telegram", "port": 8007, "module": "services.telegram_adapter_service.app:app"}, {"name": "whatsapp", "port": 8019, "module": "services.whatsapp_adapter_service.app:app"}, {"name": "ai", "port": 8017, "module": "services.ai_orchestrator_service.app:app"}, + {"name": "streaming-asr-sidecar", "port": 8021, "module": "services.streaming_asr_sidecar_service.app:app"}, {"name": "ai-voice-runtime", "port": 8018, "module": "services.ai_voice_runtime_service.app:app"}, {"name": "webchat", "port": 8011, "module": "services.webchat_adapter_service.app:app"}, {"name": "email", "port": 8012, "module": "services.email_adapter_service.app:app"}, @@ -317,6 +318,12 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ env["AI_VOICE_V2_DUPLEX_ENABLED"] = env.get("AI_VOICE_V2_DUPLEX_ENABLED", "1") env["AI_VOICE_V2_STREAMING_ASR_BACKEND"] = env.get("AI_VOICE_V2_STREAMING_ASR_BACKEND", "local_sidecar") env["AI_VOICE_V2_STREAMING_ASR_BASE_URL"] = env.get("AI_VOICE_V2_STREAMING_ASR_BASE_URL", "http://127.0.0.1:8021") + env["AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS"] = env.get("AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS", "4") + env["AI_VOICE_V2_STREAMING_ASR_MODEL"] = env.get("AI_VOICE_V2_STREAMING_ASR_MODEL", "base") + env["AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE"] = env.get("AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE", "int8") + env["AI_VOICE_V2_STREAMING_ASR_DEVICE"] = env.get("AI_VOICE_V2_STREAMING_ASR_DEVICE", "cpu") + env["AI_VOICE_V2_STREAMING_ASR_CACHE_DIR"] = env.get("AI_VOICE_V2_STREAMING_ASR_CACHE_DIR", str(ROOT / ".models" / "faster_whisper")) + env["AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES"] = env.get("AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES", "ru") env["AI_VOICE_V2_PREBAKED_ACK_ENABLED"] = env.get("AI_VOICE_V2_PREBAKED_ACK_ENABLED", "1") env["AI_VOICE_V2_PREBAKED_ACK_DIR"] = env.get("AI_VOICE_V2_PREBAKED_ACK_DIR", str(DATA_DIR / "voice_v2_ack_bank")) env["AI_VOICE_V2_STREAMING_TTS"] = env.get("AI_VOICE_V2_STREAMING_TTS", "1") @@ -338,6 +345,8 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ if spec["name"] == "recording": env["CC_RECORDINGS_DIR"] = str((data_dir / "recordings").resolve()) env["RECORDING_MAX_BYTES"] = env.get("RECORDING_MAX_BYTES", "26214400") + if spec["name"] == "streaming-asr-sidecar": + Path(env["AI_VOICE_V2_STREAMING_ASR_CACHE_DIR"]).mkdir(parents=True, exist_ok=True) if spec["name"] == "gateway": env.update(build_gateway_env(env)) diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 0b5de4b..bb55467 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -6,6 +6,8 @@ from dataclasses import dataclass import httpx +from services.shared.security import issue_app_token + def _api_base() -> str: return (os.getenv("AI_API_BASE", "https://api.openai.com/v1").strip() or "https://api.openai.com/v1").rstrip("/") @@ -44,6 +46,18 @@ def _streaming_asr_timeout_seconds() -> float: return max(value, 0.25) +def _service_headers() -> 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", + ttl_seconds=300, + ) + return {"Authorization": f"Bearer {token}"} + + @dataclass(slots=True) class ASRTranscription: text: str @@ -166,6 +180,7 @@ class LocalSidecarStreamingASRProvider(StreamingASRProvider): method, f"{self._api_base}{path}", json=payload, + headers=_service_headers(), ) response.raise_for_status() except httpx.HTTPError as exc: diff --git a/services/streaming_asr_sidecar_service/__init__.py b/services/streaming_asr_sidecar_service/__init__.py new file mode 100644 index 0000000..af9f719 --- /dev/null +++ b/services/streaming_asr_sidecar_service/__init__.py @@ -0,0 +1 @@ +"""Streaming ASR sidecar service package.""" diff --git a/services/streaming_asr_sidecar_service/app.py b/services/streaming_asr_sidecar_service/app.py new file mode 100644 index 0000000..e394282 --- /dev/null +++ b/services/streaming_asr_sidecar_service/app.py @@ -0,0 +1,576 @@ +from __future__ import annotations + +import audioop +import base64 +import logging +import os +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Protocol + +from fastapi import Depends, FastAPI, HTTPException +from pydantic import BaseModel + +from services.shared.core import Role, new_id +from services.shared.models import HealthResponse +from services.shared.security import require_roles + + +LOGGER = logging.getLogger("uvicorn.error") + +app = FastAPI(title="streaming-asr-sidecar-service", version="1.0.0") + + +def _int_env(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + return int(str(raw).strip()) + except ValueError: + return default + + +def _float_env(name: str, default: float) -> float: + raw = os.getenv(name) + if raw is None: + return default + try: + return float(str(raw).strip()) + except ValueError: + return default + + +def _stream_idle_ttl_seconds() -> float: + return max(_float_env("AI_VOICE_V2_STREAMING_ASR_IDLE_TTL_SECONDS", 45.0), 5.0) + + +def _partial_min_audio_ms() -> int: + return max(_int_env("AI_VOICE_V2_STREAMING_ASR_MIN_AUDIO_MS", 320), 120) + + +def _partial_recompute_interval_ms() -> int: + return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_RECOMPUTE_INTERVAL_MS", 200), 80) + + +def _partial_stability_hold_ms() -> int: + return max(_int_env("AI_VOICE_V2_STREAMING_ASR_PARTIAL_STABILITY_HOLD_MS", 400), 120) + + +def _model_name() -> str: + return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_MODEL", "base") or "base").strip() or "base" + + +def _compute_type() -> str: + return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE", "int8") or "int8").strip() or "int8" + + +def _device() -> str: + return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_DEVICE", "cpu") or "cpu").strip() or "cpu" + + +def _cache_dir() -> str: + return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_CACHE_DIR", "/models/faster_whisper") or "/models/faster_whisper").strip() or "/models/faster_whisper" + + +def _supported_languages() -> set[str]: + raw = str(os.getenv("AI_VOICE_V2_STREAMING_ASR_SUPPORTED_LANGUAGES", "ru") or "ru").strip() + values = {item.strip().lower() for item in raw.split(",") if item.strip()} + return values or {"ru"} + + +def _target_sample_rate_hz() -> int: + return max(_int_env("AI_VOICE_V2_STREAMING_ASR_TARGET_SAMPLE_RATE_HZ", 16000), 8000) + + +def _normalize_language_hint(language_hint: str | None) -> str: + raw = str(language_hint or "ru").strip().lower() or "ru" + if raw in {"ru", "ru-ru"}: + return "ru" + if raw in {"kk", "kz", "kk-kk"}: + return "kk" + return raw + + +def _pcm_duration_ms(pcm_bytes: bytes | bytearray, sample_rate_hz: int) -> int: + if sample_rate_hz <= 0: + return 0 + sample_count = len(pcm_bytes) // 2 + return int((sample_count / float(sample_rate_hz)) * 1000.0) + + +def _resample_pcm16le(pcm_bytes: bytes, *, input_rate_hz: int, output_rate_hz: int) -> bytes: + if not pcm_bytes or input_rate_hz == output_rate_hz: + return pcm_bytes + converted, _ = audioop.ratecv( + pcm_bytes, + 2, + 1, + input_rate_hz, + output_rate_hz, + None, + ) + return converted + + +@dataclass(slots=True) +class SidecarTranscript: + text: str + language: str | None = None + confidence: float | None = None + + +class TranscriptionEngine(Protocol): + def transcribe_pcm( + self, + pcm_bytes: bytes, + *, + sample_rate_hz: int, + language_hint: str | None = None, + ) -> SidecarTranscript: ... + + +class FasterWhisperEngine: + def __init__( + self, + *, + model_name: str, + compute_type: str, + device: str, + cache_dir: str, + target_sample_rate_hz: int, + ) -> None: + self._model_name = model_name + self._compute_type = compute_type + self._device = device + self._cache_dir = cache_dir + self._target_sample_rate_hz = target_sample_rate_hz + self._model = None + self._lock = threading.Lock() + + def _load_model(self): + with self._lock: + if self._model is not None: + return self._model + try: + from faster_whisper import WhisperModel + except Exception as exc: # noqa: BLE001 + raise RuntimeError( + "faster-whisper is not installed; install requirements before starting the streaming ASR sidecar" + ) from exc + self._model = WhisperModel( + self._model_name, + device=self._device, + compute_type=self._compute_type, + download_root=self._cache_dir, + ) + return self._model + + def transcribe_pcm( + self, + pcm_bytes: bytes, + *, + sample_rate_hz: int, + language_hint: str | None = None, + ) -> SidecarTranscript: + if not pcm_bytes: + return SidecarTranscript(text="", language=language_hint, confidence=None) + try: + import numpy as np + except Exception as exc: # noqa: BLE001 + raise RuntimeError("numpy is required for streaming ASR sidecar") from exc + + model = self._load_model() + pcm_model_rate = _resample_pcm16le( + pcm_bytes, + input_rate_hz=sample_rate_hz, + output_rate_hz=self._target_sample_rate_hz, + ) + waveform = np.frombuffer(pcm_model_rate, dtype=np.int16).astype(np.float32) / 32768.0 + segments, info = model.transcribe( + waveform, + language=_normalize_language_hint(language_hint), + beam_size=1, + best_of=1, + temperature=0.0, + vad_filter=False, + word_timestamps=False, + condition_on_previous_text=False, + without_timestamps=True, + ) + parts: list[str] = [] + for segment in segments: + text = str(getattr(segment, "text", "") or "").strip() + if text: + parts.append(text) + text = " ".join(parts).strip() + language = str(getattr(info, "language", "") or language_hint or "").strip() or language_hint + confidence_raw = getattr(info, "language_probability", None) + confidence = float(confidence_raw) if isinstance(confidence_raw, (int, float)) else None + if confidence is not None: + confidence = max(0.0, min(confidence, 1.0)) + return SidecarTranscript(text=text, language=language, confidence=confidence) + + +def _default_engine_factory() -> TranscriptionEngine: + return FasterWhisperEngine( + model_name=_model_name(), + compute_type=_compute_type(), + device=_device(), + cache_dir=_cache_dir(), + target_sample_rate_hz=_target_sample_rate_hz(), + ) + + +_ENGINE_FACTORY = _default_engine_factory +_ENGINE_INSTANCE: TranscriptionEngine | None = None +_ENGINE_LOCK = threading.Lock() + + +def _get_engine() -> TranscriptionEngine: + global _ENGINE_INSTANCE + with _ENGINE_LOCK: + if _ENGINE_INSTANCE is None: + _ENGINE_INSTANCE = _ENGINE_FACTORY() + return _ENGINE_INSTANCE + + +@dataclass(slots=True) +class StreamState: + stream_id: str + session_id: str + language_hint: str + sample_rate_hz: int + encoding: str + pcm_buffer: bytearray = field(default_factory=bytearray) + created_monotonic: float = field(default_factory=time.monotonic) + last_activity_monotonic: float = field(default_factory=time.monotonic) + buffer_version: int = 0 + partial_text: str = "" + partial_language: str | None = None + partial_confidence: float | None = None + partial_buffer_version: int = -1 + partial_updated_monotonic: float = 0.0 + partial_first_seen_monotonic: float = 0.0 + partial_repeat_count: int = 0 + final_transcript: SidecarTranscript | None = None + + def touch(self) -> None: + self.last_activity_monotonic = time.monotonic() + + +class StreamStore: + def __init__(self, *, idle_ttl_seconds: float) -> None: + self._idle_ttl_seconds = idle_ttl_seconds + self._streams: dict[str, StreamState] = {} + self._lock = threading.Lock() + + def cleanup_expired(self) -> list[str]: + now = time.monotonic() + removed: list[str] = [] + with self._lock: + for stream_id, stream in list(self._streams.items()): + if now - stream.last_activity_monotonic < self._idle_ttl_seconds: + continue + removed.append(stream_id) + self._streams.pop(stream_id, None) + return removed + + def create(self, *, session_id: str, language_hint: str, sample_rate_hz: int, encoding: str) -> StreamState: + stream = StreamState( + stream_id=new_id("sasr"), + session_id=session_id, + language_hint=language_hint, + sample_rate_hz=sample_rate_hz, + encoding=encoding, + ) + with self._lock: + self._streams[stream.stream_id] = stream + return stream + + def get(self, stream_id: str) -> StreamState | None: + with self._lock: + stream = self._streams.get(stream_id) + if stream is not None: + stream.touch() + return stream + + def append_pcm(self, stream_id: str, pcm_chunk: bytes) -> StreamState: + with self._lock: + stream = self._streams.get(stream_id) + if stream is None: + raise KeyError(stream_id) + stream.pcm_buffer.extend(pcm_chunk) + stream.buffer_version += 1 + stream.touch() + return stream + + def update_partial( + self, + stream_id: str, + *, + transcript: SidecarTranscript, + buffer_version: int, + ) -> StreamState | None: + now = time.monotonic() + with self._lock: + stream = self._streams.get(stream_id) + if stream is None: + return None + previous_text = stream.partial_text + stream.touch() + if transcript.text: + if transcript.text == previous_text: + stream.partial_repeat_count += 1 + else: + stream.partial_first_seen_monotonic = now + stream.partial_repeat_count = 1 + stream.partial_text = transcript.text + stream.partial_language = transcript.language + stream.partial_confidence = transcript.confidence + stream.partial_updated_monotonic = now + stream.partial_buffer_version = buffer_version + return stream + + def set_final(self, stream_id: str, transcript: SidecarTranscript) -> StreamState | None: + with self._lock: + stream = self._streams.get(stream_id) + if stream is None: + return None + stream.final_transcript = transcript + stream.touch() + return stream + + def delete(self, stream_id: str) -> bool: + with self._lock: + return self._streams.pop(stream_id, None) is not None + + +_STREAMS = StreamStore(idle_ttl_seconds=_stream_idle_ttl_seconds()) + + +class OpenStreamIn(BaseModel): + session_id: str + language_hint: str | None = None + sample_rate_hz: int = 8000 + encoding: str = "pcm_s16le" + + +class OpenStreamOut(BaseModel): + ok: bool = True + stream_id: str + language: str + + +class PushChunkIn(BaseModel): + pcm_b64: str + sample_rate_hz: int = 8000 + encoding: str = "pcm_s16le" + + +class PushChunkOut(BaseModel): + ok: bool = True + stream_id: str + received_bytes: int + duration_ms: int + + +class PartialOut(BaseModel): + text: str + language: str | None = None + confidence: float | None = None + is_final: bool = False + is_stable: bool = False + + +class FinalizeOut(BaseModel): + text: str + language: str | None = None + confidence: float | None = None + + +def _require_internal_actor(_: dict = Depends(require_roles(Role.ADMIN))) -> None: + return None + + +def _cleanup_expired_streams() -> None: + removed = _STREAMS.cleanup_expired() + if removed: + LOGGER.info("streaming_asr.cleanup_expired removed=%s", len(removed)) + + +def _current_partial_is_stable(stream: StreamState) -> bool: + if not stream.partial_text: + return False + if stream.partial_repeat_count >= 2: + return True + return (time.monotonic() - stream.partial_first_seen_monotonic) >= (_partial_stability_hold_ms() / 1000.0) + + +def _copy_stream_snapshot(stream: StreamState) -> tuple[bytes, int, int, str]: + return ( + bytes(stream.pcm_buffer), + stream.buffer_version, + stream.sample_rate_hz, + stream.language_hint, + ) + + +def _transcribe_stream_snapshot(pcm_bytes: bytes, *, sample_rate_hz: int, language_hint: str) -> SidecarTranscript: + return _get_engine().transcribe_pcm( + pcm_bytes, + sample_rate_hz=sample_rate_hz, + language_hint=language_hint, + ) + + +def _current_partial_payload(stream: StreamState) -> PartialOut | None: + if not stream.partial_text: + return None + return PartialOut( + text=stream.partial_text, + language=stream.partial_language, + confidence=stream.partial_confidence, + is_final=False, + is_stable=_current_partial_is_stable(stream), + ) + + +def _language_or_error(language_hint: str | None) -> str: + normalized = _normalize_language_hint(language_hint) + if normalized not in _supported_languages(): + raise HTTPException( + status_code=400, + detail=f"Unsupported streaming ASR language: {normalized}", + ) + return normalized + + +def _validate_audio_params(sample_rate_hz: int, encoding: str) -> None: + if int(sample_rate_hz) != 8000: + raise HTTPException(status_code=400, detail="Only 8000 Hz streaming ASR input is supported") + if str(encoding or "").strip().lower() != "pcm_s16le": + raise HTTPException(status_code=400, detail="Only pcm_s16le encoding is supported") + + +@app.get("/health", response_model=HealthResponse) +def health() -> HealthResponse: + _cleanup_expired_streams() + with _ENGINE_LOCK: + loaded = _ENGINE_INSTANCE is not None + suffix = "loaded" if loaded else "cold" + return HealthResponse(status="ok", service=f"streaming-asr-sidecar ({suffix})") + + +@app.post("/internal/asr/streams", response_model=OpenStreamOut) +def open_stream(payload: OpenStreamIn, _: None = Depends(_require_internal_actor)) -> OpenStreamOut: + _cleanup_expired_streams() + session_id = str(payload.session_id or "").strip() + if not session_id: + raise HTTPException(status_code=400, detail="session_id is required") + language = _language_or_error(payload.language_hint) + _validate_audio_params(payload.sample_rate_hz, payload.encoding) + stream = _STREAMS.create( + session_id=session_id, + language_hint=language, + sample_rate_hz=payload.sample_rate_hz, + encoding=payload.encoding, + ) + return OpenStreamOut(stream_id=stream.stream_id, language=language) + + +@app.post("/internal/asr/streams/{stream_id}/chunks", response_model=PushChunkOut) +def push_chunk(stream_id: str, payload: PushChunkIn, _: None = Depends(_require_internal_actor)) -> PushChunkOut: + _cleanup_expired_streams() + _validate_audio_params(payload.sample_rate_hz, payload.encoding) + try: + pcm_bytes = base64.b64decode(str(payload.pcm_b64 or "").encode("ascii"), validate=True) + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=400, detail="pcm_b64 must be valid base64") from exc + try: + stream = _STREAMS.append_pcm(stream_id, pcm_bytes) + except KeyError as exc: + raise HTTPException(status_code=404, detail="ASR stream not found") from exc + return PushChunkOut( + stream_id=stream_id, + received_bytes=len(pcm_bytes), + duration_ms=_pcm_duration_ms(stream.pcm_buffer, stream.sample_rate_hz), + ) + + +@app.get("/internal/asr/streams/{stream_id}/partial", response_model=PartialOut | dict[str, Any]) +def poll_partial(stream_id: str, _: None = Depends(_require_internal_actor)) -> PartialOut | dict[str, Any]: + _cleanup_expired_streams() + stream = _STREAMS.get(stream_id) + if stream is None: + raise HTTPException(status_code=404, detail="ASR stream not found") + if stream.final_transcript is not None: + return PartialOut( + text=stream.final_transcript.text, + language=stream.final_transcript.language, + confidence=stream.final_transcript.confidence, + is_final=True, + is_stable=True, + ) + if _pcm_duration_ms(stream.pcm_buffer, stream.sample_rate_hz) < _partial_min_audio_ms(): + payload = _current_partial_payload(stream) + return payload.model_dump() if payload else {} + now = time.monotonic() + if ( + stream.partial_text + and stream.partial_buffer_version == stream.buffer_version + and (now - stream.partial_updated_monotonic) < (_partial_recompute_interval_ms() / 1000.0) + ): + payload = _current_partial_payload(stream) + return payload.model_dump() if payload else {} + + pcm_bytes, buffer_version, sample_rate_hz, language_hint = _copy_stream_snapshot(stream) + transcript = _transcribe_stream_snapshot( + pcm_bytes, + sample_rate_hz=sample_rate_hz, + language_hint=language_hint, + ) + stream = _STREAMS.update_partial( + stream_id, + transcript=transcript, + buffer_version=buffer_version, + ) + if stream is None: + raise HTTPException(status_code=404, detail="ASR stream not found") + payload = _current_partial_payload(stream) + return payload.model_dump() if payload else {} + + +@app.post("/internal/asr/streams/{stream_id}/finalize", response_model=FinalizeOut) +def finalize_stream(stream_id: str, _: None = Depends(_require_internal_actor)) -> FinalizeOut: + _cleanup_expired_streams() + stream = _STREAMS.get(stream_id) + if stream is None: + raise HTTPException(status_code=404, detail="ASR stream not found") + if stream.final_transcript is not None: + return FinalizeOut( + text=stream.final_transcript.text, + language=stream.final_transcript.language, + confidence=stream.final_transcript.confidence, + ) + pcm_bytes, _buffer_version, sample_rate_hz, language_hint = _copy_stream_snapshot(stream) + transcript = _transcribe_stream_snapshot( + pcm_bytes, + sample_rate_hz=sample_rate_hz, + language_hint=language_hint, + ) + _STREAMS.set_final(stream_id, transcript) + return FinalizeOut( + text=transcript.text, + language=transcript.language, + confidence=transcript.confidence, + ) + + +@app.delete("/internal/asr/streams/{stream_id}") +def close_stream(stream_id: str, _: None = Depends(_require_internal_actor)) -> dict[str, Any]: + _cleanup_expired_streams() + if not _STREAMS.delete(stream_id): + raise HTTPException(status_code=404, detail="ASR stream not found") + return {"ok": True, "stream_id": stream_id} diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index f77acc0..855053c 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -14,7 +14,13 @@ from services.ai_voice_runtime_service.audiosocket import ( resample_pcm16le, ) from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaActor, MediaRegistration -from services.ai_voice_runtime_service.providers.asr import ASRProvider, ASRTranscription +from services.ai_voice_runtime_service.providers.asr import ( + ASRProvider, + ASRTranscription, + StreamingASRPartial, + StreamingASRProvider, + StreamingASRUnavailable, +) from services.ai_voice_runtime_service.providers.tts import TTSProvider, TTSSynthesis from services.shared.models import VoiceAITurnDecisionOut @@ -1221,3 +1227,272 @@ def test_media_runtime_voice_v2_emotive_ack_avoids_same_variant_back_to_back(): assert first_text in runtime._base_ack_variants("ru", "understanding") assert second_text in runtime._base_ack_variants("ru", "understanding") assert first_text != second_text + + +def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr(): + registrations: dict[str, MediaRegistration] = {} + reply_starts: list[tuple[str, str | None, float]] = [] + turns: list[str] = [] + delivered: list[tuple[str, bool]] = [] + + class _ExplodingBatchASRProvider(ASRProvider): + name = "exploding-batch-asr" + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + raise AssertionError("batch ASR should not be used when streaming sidecar is active") + + class _FakeStreamingASRProvider(StreamingASRProvider): + name = "fake-streaming-sidecar" + supports_streaming = True + + def __init__(self) -> None: + self.chunk_count = 0 + self.events: list[tuple[str, float]] = [] + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + del session_id, language_hint + self.events.append(("open", time.monotonic())) + return "stream-1" + + def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None: + del stream_id + assert pcm_8k_chunk + self.chunk_count += 1 + self.events.append(("push", time.monotonic())) + + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: + del stream_id + self.events.append(("poll", time.monotonic())) + if self.chunk_count >= 2: + return StreamingASRPartial( + text="мне нужен график работы", + language="ru", + confidence=0.84, + is_stable=True, + ) + return None + + def finalize(self, stream_id: str) -> ASRTranscription: + del stream_id + self.events.append(("finalize", time.monotonic())) + time.sleep(0.2) + return ASRTranscription(text="мне нужен график работы", language="ru", confidence=0.84) + + def close_stream(self, stream_id: str) -> None: + del stream_id + self.events.append(("close", time.monotonic())) + + media_uuid = str(uuid.uuid4()) + registrations[media_uuid] = MediaRegistration( + voice_session_id="avs_media_runtime_v2_streaming", + call_id="call_media_runtime_v2_streaming", + interaction_id="int_media_runtime_v2_streaming", + ai_session_id="ais_media_runtime_v2_streaming", + language="ru", + media_uuid=media_uuid, + queue_code="voice_lab_ai", + queue_id="que_voice_lab_ai", + agent_profile="voice_support", + voice_v2_enabled=True, + voice_v2_ack_mode="immediate_short", + voice_v2_streaming_tts=True, + voice_v2_partial_asr=True, + voice_v2_duplex=True, + voice_v2_streaming_asr_backend="local_sidecar", + ) + streaming_provider = _FakeStreamingASRProvider() + + runtime = AudioSocketMediaRuntime( + enabled=True, + host="127.0.0.1", + port=0, + frame_ms=20, + idle_timeout_seconds=2.0, + registration_wait_timeout_seconds=0.5, + min_speech_ms=40, + trailing_silence_ms=40, + max_turn_ms=2000, + asr_provider=_ExplodingBatchASRProvider(), + streaming_asr_provider=streaming_provider, + tts_provider=_StubTTSProvider(), + load_registration_by_media_uuid=lambda value: registrations.get(value), + mark_media_connected=lambda session_id, value: None, + mark_media_ended=lambda session_id, reason: None, + touch_media_frame=lambda session_id: None, + set_state=lambda session_id, state, handoff_reason, metadata: None, + get_pending_greeting=lambda session_id: None, + mark_reply_started=lambda session_id, text, is_greeting, phase: reply_starts.append((text, phase, time.monotonic())), + mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((text, is_greeting)), + plan_reply=lambda session_id, text, metadata, kind: None, + process_turn=lambda session_id, transcript_text, language, barge_in, metadata: ( + turns.append(transcript_text) + or VoiceAITurnDecisionOut( + language=language or "ru", + intent="clarification", + reply_text="Назовите, пожалуйста, город.", + confidence=0.9, + needs_handoff=False, + handoff_reason=None, + case_action="keep_open", + kb_refs=[], + summary_text="reply ready", + model="stub-voice", + latency_ms=1, + status="active", + ) + ), + request_handoff=lambda session_id, customer_request_text, decision: None, + handle_media_error=lambda session_id, message, metadata: None, + ) + + async def _scenario() -> None: + await runtime.start() + port = runtime._server.sockets[0].getsockname()[1] + + reader, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes)) + await writer.drain() + + speech_frame = (1000).to_bytes(2, "little", signed=True) * 160 + silence_frame = b"\x00\x00" * 160 + for _ in range(14): + writer.write(encode_audio_packet(speech_frame)) + for _ in range(2): + writer.write(encode_audio_packet(silence_frame)) + await writer.drain() + + deadline = time.time() + 2.5 + while time.time() < deadline: + if len(reply_starts) >= 2 and turns: + break + await asyncio.sleep(0.05) + + writer.close() + await writer.wait_closed() + await asyncio.sleep(0.2) + await runtime.stop() + + asyncio.run(_scenario()) + + phases = [phase for _, phase, _ in reply_starts] + finalize_started_at = next(ts for name, ts in streaming_provider.events if name == "finalize") + assert turns == ["мне нужен график работы"] + assert "open" in [name for name, _ in streaming_provider.events] + assert "close" in [name for name, _ in streaming_provider.events] + assert phases[:2] == ["ack", "main"] + assert reply_starts[0][2] < finalize_started_at + assert any(text == "Назовите, пожалуйста, город." and phase == "main" for text, phase, _ in reply_starts) + + +def test_media_runtime_voice_v2_falls_back_when_streaming_sidecar_is_unavailable(): + registrations: dict[str, MediaRegistration] = {} + turns: list[str] = [] + + class _BatchASRProvider(ASRProvider): + name = "batch-asr" + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + assert audio_bytes + return ASRTranscription(text="нужен оператор", language=language_hint or "ru", confidence=0.88) + + class _UnavailableStreamingProvider(StreamingASRProvider): + name = "missing-sidecar" + supports_streaming = True + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + del session_id, language_hint + raise StreamingASRUnavailable("sidecar down") + + media_uuid = str(uuid.uuid4()) + registrations[media_uuid] = MediaRegistration( + voice_session_id="avs_media_runtime_v2_fallback", + call_id="call_media_runtime_v2_fallback", + interaction_id="int_media_runtime_v2_fallback", + ai_session_id="ais_media_runtime_v2_fallback", + language="ru", + media_uuid=media_uuid, + queue_code="voice_lab_ai", + queue_id="que_voice_lab_ai", + agent_profile="voice_support", + voice_v2_enabled=True, + voice_v2_ack_mode="immediate_short", + voice_v2_streaming_tts=True, + voice_v2_partial_asr=True, + voice_v2_duplex=True, + voice_v2_streaming_asr_backend="local_sidecar", + ) + + runtime = AudioSocketMediaRuntime( + enabled=True, + host="127.0.0.1", + port=0, + frame_ms=20, + idle_timeout_seconds=2.0, + registration_wait_timeout_seconds=0.5, + min_speech_ms=40, + trailing_silence_ms=40, + max_turn_ms=2000, + asr_provider=_BatchASRProvider(), + streaming_asr_provider=_UnavailableStreamingProvider(), + tts_provider=_StubTTSProvider(), + load_registration_by_media_uuid=lambda value: registrations.get(value), + mark_media_connected=lambda session_id, value: None, + mark_media_ended=lambda session_id, reason: None, + touch_media_frame=lambda session_id: None, + set_state=lambda session_id, state, handoff_reason, metadata: None, + get_pending_greeting=lambda session_id: None, + mark_reply_delivered=lambda session_id, text, is_greeting: None, + plan_reply=lambda session_id, text, metadata, kind: None, + process_turn=lambda session_id, transcript_text, language, barge_in, metadata: ( + turns.append(transcript_text) + or VoiceAITurnDecisionOut( + language=language or "ru", + intent="handoff", + reply_text="Соединяю с оператором.", + confidence=0.9, + needs_handoff=False, + handoff_reason=None, + case_action="keep_open", + kb_refs=[], + summary_text="reply ready", + model="stub-voice", + latency_ms=1, + status="active", + ) + ), + request_handoff=lambda session_id, customer_request_text, decision: None, + handle_media_error=lambda session_id, message, metadata: None, + ) + + async def _scenario() -> None: + await runtime.start() + port = runtime._server.sockets[0].getsockname()[1] + + reader, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes)) + await writer.drain() + + speech_frame = (1000).to_bytes(2, "little", signed=True) * 160 + silence_frame = b"\x00\x00" * 160 + for _ in range(2): + writer.write(encode_audio_packet(speech_frame)) + for _ in range(2): + writer.write(encode_audio_packet(silence_frame)) + await writer.drain() + + deadline = time.time() + 2.0 + while time.time() < deadline: + if turns: + break + await asyncio.sleep(0.05) + + writer.close() + await writer.wait_closed() + await asyncio.sleep(0.2) + await runtime.stop() + + asyncio.run(_scenario()) + + assert turns == ["нужен оператор"] + assert registrations[media_uuid].voice_v2_duplex is False + assert registrations[media_uuid].voice_v2_partial_asr is False diff --git a/tests/test_streaming_asr_sidecar_service.py b/tests/test_streaming_asr_sidecar_service.py new file mode 100644 index 0000000..5b6a09f --- /dev/null +++ b/tests/test_streaming_asr_sidecar_service.py @@ -0,0 +1,181 @@ +import base64 +import time + +from fastapi.testclient import TestClient + +import services.streaming_asr_sidecar_service.app as sidecar_module + + +def _admin_headers() -> dict[str, str]: + return {"X-User": "admin", "X-Role": "admin"} + + +def _pcm_chunk(duration_ms: int = 400, *, amplitude: int = 1000) -> bytes: + sample_count = int(8000 * (duration_ms / 1000.0)) + return int(amplitude).to_bytes(2, "little", signed=True) * sample_count + + +class _FakeEngine: + def __init__(self, *, text: str = "привет") -> None: + self.text = text + self.calls: list[tuple[int, int, str | None]] = [] + + def transcribe_pcm( + self, + pcm_bytes: bytes, + *, + sample_rate_hz: int, + language_hint: str | None = None, + ) -> sidecar_module.SidecarTranscript: + self.calls.append((len(pcm_bytes), sample_rate_hz, language_hint)) + return sidecar_module.SidecarTranscript( + text=self.text, + language=language_hint or "ru", + confidence=0.91, + ) + + +def _reset_sidecar( + monkeypatch, + *, + engine: _FakeEngine | None = None, + idle_ttl_seconds: float = 45.0, + supported_languages: set[str] | None = None, +) -> _FakeEngine: + fake_engine = engine or _FakeEngine() + monkeypatch.setattr(sidecar_module, "_ENGINE_INSTANCE", None) + monkeypatch.setattr(sidecar_module, "_ENGINE_FACTORY", lambda: fake_engine) + monkeypatch.setattr(sidecar_module, "_STREAMS", sidecar_module.StreamStore(idle_ttl_seconds=idle_ttl_seconds)) + monkeypatch.setattr( + sidecar_module, + "_supported_languages", + lambda: set(supported_languages or {"ru"}), + ) + return fake_engine + + +def test_streaming_asr_sidecar_stream_lifecycle(monkeypatch): + engine = _reset_sidecar(monkeypatch) + client = TestClient(sidecar_module.app) + + open_response = client.post( + "/internal/asr/streams", + headers=_admin_headers(), + json={ + "session_id": "avs_sidecar_1", + "language_hint": "ru", + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + assert open_response.status_code == 200 + stream_id = open_response.json()["stream_id"] + + chunk = _pcm_chunk(400) + push_response = client.post( + f"/internal/asr/streams/{stream_id}/chunks", + headers=_admin_headers(), + json={ + "pcm_b64": base64.b64encode(chunk).decode("ascii"), + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + assert push_response.status_code == 200 + assert push_response.json()["received_bytes"] == len(chunk) + + partial_one = client.get( + f"/internal/asr/streams/{stream_id}/partial", + headers=_admin_headers(), + ) + assert partial_one.status_code == 200 + first_payload = partial_one.json() + assert first_payload["text"] == "привет" + assert first_payload["is_stable"] is False + + second_chunk = _pcm_chunk(80) + client.post( + f"/internal/asr/streams/{stream_id}/chunks", + headers=_admin_headers(), + json={ + "pcm_b64": base64.b64encode(second_chunk).decode("ascii"), + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + partial_two = client.get( + f"/internal/asr/streams/{stream_id}/partial", + headers=_admin_headers(), + ) + assert partial_two.status_code == 200 + second_payload = partial_two.json() + assert second_payload["text"] == "привет" + assert second_payload["is_stable"] is True + + finalize_response = client.post( + f"/internal/asr/streams/{stream_id}/finalize", + headers=_admin_headers(), + ) + assert finalize_response.status_code == 200 + assert finalize_response.json() == { + "text": "привет", + "language": "ru", + "confidence": 0.91, + } + + close_response = client.delete( + f"/internal/asr/streams/{stream_id}", + headers=_admin_headers(), + ) + assert close_response.status_code == 200 + assert close_response.json() == {"ok": True, "stream_id": stream_id} + assert len(engine.calls) >= 3 + + +def test_streaming_asr_sidecar_rejects_unsupported_language(monkeypatch): + _reset_sidecar(monkeypatch, supported_languages={"ru"}) + client = TestClient(sidecar_module.app) + + response = client.post( + "/internal/asr/streams", + headers=_admin_headers(), + json={ + "session_id": "avs_sidecar_kk", + "language_hint": "kk", + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + + assert response.status_code == 400 + assert "Unsupported streaming ASR language" in response.json()["detail"] + + +def test_streaming_asr_sidecar_cleans_up_idle_streams(monkeypatch): + _reset_sidecar(monkeypatch, idle_ttl_seconds=0.01) + client = TestClient(sidecar_module.app) + + open_response = client.post( + "/internal/asr/streams", + headers=_admin_headers(), + json={ + "session_id": "avs_sidecar_idle", + "language_hint": "ru", + "sample_rate_hz": 8000, + "encoding": "pcm_s16le", + }, + ) + assert open_response.status_code == 200 + stream_id = open_response.json()["stream_id"] + + with sidecar_module._STREAMS._lock: + sidecar_module._STREAMS._streams[stream_id].last_activity_monotonic = time.monotonic() - 1.0 + + health_response = client.get("/health") + assert health_response.status_code == 200 + + missing_response = client.get( + f"/internal/asr/streams/{stream_id}/partial", + headers=_admin_headers(), + ) + assert missing_response.status_code == 404