feat(voice): add streaming asr sidecar for v2 duplex

.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

.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

.gitignore

@@ -10,6 +10,7 @@
 .codex_tmp/
 .server_backups/
 .tmp/
+.models/
 test-results/
 .testdata/
 backups/

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

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

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

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

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

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

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:

services/streaming_asr_sidecar_service/__init__.py

@@ -0,0 +1 @@
+"""Streaming ASR sidecar service package."""

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}

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

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