vad_gate.py

"""
VAD Streaming Gate — Issue #4644

Server-side VAD gate that skips sending silence to the STT provider,
using KeepAlive to maintain the connection and Finalize to flush
pending transcripts on speech→silence transitions.

Modes (VAD_GATE_MODE env var):
  off    — disabled, all audio forwarded (default)
  shadow — VAD runs and logs decisions, but all audio still forwarded
  active — VAD gates audio: silence skipped, KeepAlive sent instead
"""

import audioop
import logging
import os
import threading
import time
from bisect import bisect_right
from collections import deque
from dataclasses import dataclass
from enum import Enum
from typing import Any, Deque, Dict, List, Optional, Tuple

import numpy as np

from utils.metrics import (
    OMI_LIVE_STT_MISALIGNED_FRAMES_TOTAL,
    OMI_VAD_GATE_AUDIO_SECONDS_TOTAL,
    OMI_VAD_GATE_SESSIONS_TOTAL,
)
from utils.observability.fallback import record_fallback
from utils.stt.socket import STTSocket
from utils.stt.vad import (
    VAD_WINDOW_SAMPLES,
    _get_ort_session,  # type: ignore[reportPrivateUsage]  # internal helper, same package
    make_fresh_state,
    run_vad_window,
)

logger = logging.getLogger('vad_gate')

# ---------------------------------------------------------------------------
# Configuration from environment
# ---------------------------------------------------------------------------
VAD_GATE_MODE = os.getenv('VAD_GATE_MODE', 'off')  # off | shadow | active
VAD_GATE_PRE_ROLL_MS = 300
VAD_GATE_HANGOVER_MS = 4000
VAD_GATE_SPEECH_THRESHOLD = 0.65
VAD_GATE_FINALIZE_SILENCE_MS = 300  # Flush DG transcript during hangover after this much silence
VAD_GATE_KEEPALIVE_SEC = 5


def is_gate_enabled() -> bool:
    return VAD_GATE_MODE in ('shadow', 'active')


# ---------------------------------------------------------------------------
# Gate state machine
# ---------------------------------------------------------------------------
class GateState(str, Enum):
    SILENCE = 'silence'
    SPEECH = 'speech'
    HANGOVER = 'hangover'


@dataclass
class GateOutput:
    """Output from processing one audio chunk through the gate."""

    audio_to_send: bytes  # PCM bytes to forward to DG (may be empty)
    should_finalize: bool = False  # call dg_socket.finalize()
    state: GateState = GateState.SILENCE
    is_speech: bool = False  # raw VAD decision for this chunk


# ---------------------------------------------------------------------------
# DG ↔ Wall-clock timestamp mapper
# ---------------------------------------------------------------------------
class WallTimeMapper:
    """Maps STT provider audio-time timestamps to wall-clock-relative timestamps.

    Provider timestamps are continuous (only counting audio actually sent).
    When we skip silence via KeepAlive, provider time compresses vs wall time.
    This mapper tracks checkpoints at each silence→speech transition to
    convert provider timestamps back to wall-clock-relative timestamps.
    """

    _MAX_CHECKPOINTS = 500  # Cap to bound memory for long sessions

    def __init__(self):
        self._lock = threading.Lock()
        # Each checkpoint: (dg_sec, wall_rel_sec) at silence→speech transition
        self._checkpoints: List[Tuple[float, float]] = []
        self._provider_cursor_sec: float = 0.0
        self._sending: bool = False

    def on_audio_sent(self, chunk_duration_sec: float, chunk_wall_rel_sec: float) -> None:
        """Called when audio is actually sent to DG."""
        with self._lock:
            if not self._sending:
                # Enforce monotonicity: the new checkpoint's wall time must be at
                # least prev_wall + (dg_elapsed since prev checkpoint).  Pre-roll
                # subtraction can produce wall times below the previous checkpoint,
                # and simple clamping to prev_wall creates overlapping wall-time
                # ranges that cause non-monotonic remapped timestamps.
                if self._checkpoints:
                    prev_dg, prev_wall = self._checkpoints[-1]
                    min_wall = prev_wall + (self._provider_cursor_sec - prev_dg)
                    chunk_wall_rel_sec = max(chunk_wall_rel_sec, min_wall)
                self._checkpoints.append((self._provider_cursor_sec, chunk_wall_rel_sec))
                # Compact: keep an anchor for early remaps + recent checkpoints.
                if len(self._checkpoints) > self._MAX_CHECKPOINTS:
                    if self._MAX_CHECKPOINTS <= 1:
                        self._checkpoints = self._checkpoints[:1]
                    else:
                        self._checkpoints = [self._checkpoints[0]] + self._checkpoints[-(self._MAX_CHECKPOINTS - 1) :]
                self._sending = True
            self._provider_cursor_sec += chunk_duration_sec

    def on_silence_skipped(self) -> None:
        """Called when silence is skipped (not sent to DG)."""
        with self._lock:
            self._sending = False

    def dg_to_wall_rel(self, dg_sec: float) -> float:
        """Convert DG audio-time to wall-clock-relative time."""
        with self._lock:
            cps = self._checkpoints[:]
        if not cps:
            return dg_sec
        dg_marks = [c[0] for c in cps]
        i = max(bisect_right(dg_marks, dg_sec) - 1, 0)
        cp_dg, cp_wall = cps[i]
        return cp_wall + (dg_sec - cp_dg)


# ---------------------------------------------------------------------------
# VAD Streaming Gate (per-session)
# ---------------------------------------------------------------------------
class VADStreamingGate:
    """Per-session VAD gate that decides whether to send audio to the STT provider.

    Uses ONNX Silero-VAD model's speech probability (not start/end events)
    for robust per-chunk speech detection. Buffers VAD input samples to handle
    chunk sizes smaller than the VAD window (e.g. 16ms at 16kHz = 256 samples).

    The ONNX InferenceSession is shared process-wide (thread-safe).
    Per-connection recurrent state (h/c) and context are stored on this instance.

    Args:
        sample_rate: Input audio sample rate (Hz)
        channels: Number of audio channels
        mode: 'shadow' or 'active'
        uid: User ID for logging
        session_id: Session ID for logging
    """

    def __init__(
        self,
        sample_rate: int = 16000,
        channels: int = 1,
        mode: str = 'active',
        uid: str = '',
        session_id: str = '',
    ):
        self.sample_rate = sample_rate
        self.channels = channels
        self.mode = mode
        self.uid = uid
        self.session_id = session_id
        # All audio reaching the gate MUST be PCM16 LE (2 bytes/sample).
        self._sample_width = 2  # bytes per sample, always int16

        # VAD setup — always resample to 16kHz for best accuracy
        self._vad_sample_rate = 16000
        # Eagerly init the shared ONNX session (fail-fast at gate creation)
        _get_ort_session()
        self._vad_window_samples = VAD_WINDOW_SAMPLES  # 512 for 16kHz (Silero v6)
        self._vad_buffer = np.array([], dtype=np.float32)  # Buffer for cross-chunk accumulation
        self._pcm_remainder = b''  # Trailing bytes of a frame split across chunks
        self._vad_state: np.ndarray[Any, Any]
        self._vad_context: np.ndarray[Any, Any]
        self._vad_state, self._vad_context = make_fresh_state()  # Per-connection ONNX recurrent state + context
        self._vad_inference_lock = threading.Lock()
        self._speech_threshold = VAD_GATE_SPEECH_THRESHOLD

        # State machine
        self._state = GateState.SILENCE
        self._audio_cursor_ms: float = 0.0
        self._last_speech_ms: float = 0.0
        self._pre_roll_ms = VAD_GATE_PRE_ROLL_MS
        self._hangover_ms = VAD_GATE_HANGOVER_MS
        self._finalize_silence_ms = VAD_GATE_FINALIZE_SILENCE_MS
        self._hangover_finalized = False  # True once finalize sent during current hangover

        # Pre-roll buffer: stores recent audio chunks for playback on speech onset.
        # Tracks accumulated duration to respect _pre_roll_ms regardless of chunk size.
        self._pre_roll: Deque[bytes] = deque()
        self._pre_roll_total_ms: float = 0.0

        # Timestamp mapper
        self.dg_wall_mapper = WallTimeMapper()

        # Metrics
        self._chunks_total = 0
        self._chunks_speech = 0
        self._chunks_silence = 0
        self._finalize_count = 0
        self._finalize_errors = 0
        self._bytes_received = 0
        self._bytes_sent = 0
        self._first_audio_wall_time: Optional[float] = None
        self._last_send_wall_time: Optional[float] = None  # For keepalive timing
        self._keepalive_count = 0

        # Fair-use speech accumulator (#5746)
        self._speech_ms_total: float = 0.0
        self._speech_ms_delta: float = 0.0

        # Prometheus accounting only advances once pre-roll audio has a final
        # outcome, so buffered silence cannot be counted as skipped and later sent.
        self._prometheus_bytes_sent = 0
        self._prometheus_bytes_skipped = 0
        if self.mode in ('active', 'shadow'):
            OMI_VAD_GATE_SESSIONS_TOTAL.labels(mode=self.mode).inc()

    def activate(self) -> None:
        """Switch from shadow to active mode (used after speech profile completes).

        Advances the WallTimeMapper cursor to account for all audio sent during
        shadow mode. Without this, the mapper would think provider cursor is at 0
        and over-shift all timestamps after the first gated silence gap.
        """
        if self.mode == 'shadow':
            self.mode = 'active'
            # Reset state machine to start fresh in active mode
            self._state = GateState.SILENCE
            self._pre_roll.clear()
            self._pre_roll_total_ms = 0.0
            self._hangover_finalized = False
            # Reset VAD recurrent state and buffer for clean active-mode start
            self._vad_state, self._vad_context = make_fresh_state()
            self._vad_buffer = np.array([], dtype=np.float32)
            self._pcm_remainder = b''
            # Sync mapper cursor: DG received all audio during shadow phase
            self.dg_wall_mapper._provider_cursor_sec = self._audio_cursor_ms / 1000.0  # type: ignore[reportPrivateUsage]  # sync internal mapper cursor
            logger.info(
                'VADGate activated shadow->active uid=%s session=%s cursor=%.1fms',
                self.uid,
                self.session_id,
                self._audio_cursor_ms,
            )

    def needs_keepalive(self, wall_time: float) -> bool:
        """Check if a keepalive should be sent to prevent STT provider timeout."""
        if self.mode != 'active':
            return False
        ref_time = self._last_send_wall_time or self._first_audio_wall_time
        if ref_time is None:
            return False
        return (wall_time - ref_time) >= VAD_GATE_KEEPALIVE_SEC

    def _convert_for_vad(self, pcm_data: bytes) -> np.ndarray[Any, Any]:
        """Convert audio to float32 at 16kHz mono for VAD."""
        # A client may split PCM16 anywhere, so a chunk can end mid-frame. Carry
        # those bytes into the next chunk: np.frombuffer rejects a partial sample,
        # and dropping it would shift every later sample by one byte.
        data = self._pcm_remainder + pcm_data if self._pcm_remainder else pcm_data
        frame_bytes = self._sample_width * self.channels
        leftover = len(data) % frame_bytes
        if leftover:
            self._pcm_remainder = data[len(data) - leftover :]
            data = data[: len(data) - leftover]
        else:
            self._pcm_remainder = b''
        if not data:
            return np.array([], dtype=np.float32)

        # Convert to mono if stereo
        if self.channels == 2:
            data = audioop.tomono(data, self._sample_width, 0.5, 0.5)

        data_int16 = np.frombuffer(data, dtype=np.int16)

        # Resample to 16kHz if needed
        if self.sample_rate != self._vad_sample_rate:
            # Simple linear interpolation resampling
            ratio = self._vad_sample_rate / self.sample_rate
            n_out = int(len(data_int16) * ratio)
            indices = np.linspace(0, len(data_int16) - 1, n_out)
            data_int16 = np.interp(indices, np.arange(len(data_int16)), data_int16.astype(np.float64)).astype(np.int16)

        return data_int16.astype(np.float32) / 32768.0

    def _run_vad(self, pcm_data: bytes) -> bool:
        """Run ONNX Silero VAD on audio chunk. Returns True if speech detected.

        Uses the shared ONNX InferenceSession with per-connection recurrent
        state (h/c stored on this instance). No model pool needed — ONNX
        sessions are stateless and thread-safe for different input data.

        Buffers samples across chunks to handle cases where chunk size < window size.
        """
        with self._vad_inference_lock:
            float_data = self._convert_for_vad(pcm_data)

            # Always append to buffer for cross-chunk accumulation
            self._vad_buffer = np.concatenate([self._vad_buffer, float_data])
            del float_data

            is_speech = False
            if len(self._vad_buffer) >= self._vad_window_samples:
                # Process all complete windows in buffer
                while len(self._vad_buffer) >= self._vad_window_samples:
                    window = self._vad_buffer[: self._vad_window_samples]
                    self._vad_buffer = self._vad_buffer[self._vad_window_samples :]

                    prob, self._vad_state, self._vad_context = run_vad_window(
                        window, self._vad_state, self._vad_context
                    )
                    if prob > self._speech_threshold:
                        is_speech = True

            # Keep buffer bounded (max 1 window of leftover)
            if len(self._vad_buffer) > self._vad_window_samples:
                self._vad_buffer = self._vad_buffer[-self._vad_window_samples :]

            return is_speech

    def process_audio(self, pcm_data: bytes, wall_time: float) -> GateOutput:
        """Process an audio chunk through the VAD gate.

        Args:
            pcm_data: Raw PCM16 audio bytes
            wall_time: Wall-clock timestamp of this chunk

        Returns:
            GateOutput with audio to send and control signals
        """
        if self._first_audio_wall_time is None:
            self._first_audio_wall_time = wall_time

        self._chunks_total += 1
        self._bytes_received += len(pcm_data)

        # Track audio time
        n_samples = len(pcm_data) // (self._sample_width * self.channels)
        chunk_ms = (n_samples * 1000.0) / self.sample_rate
        self._audio_cursor_ms += chunk_ms

        # Run VAD
        is_speech = self._run_vad(pcm_data)

        if is_speech:
            self._last_speech_ms = self._audio_cursor_ms
            self._chunks_speech += 1
        else:
            self._chunks_silence += 1

        # Fair-use speech accumulator (#5746)
        if is_speech and self.mode == 'active':
            self._speech_ms_total += chunk_ms
            self._speech_ms_delta += chunk_ms

        # Shadow mode: log but send everything
        if self.mode == 'shadow':
            self._bytes_sent += len(pcm_data)
            self._last_send_wall_time = wall_time
            output = GateOutput(
                audio_to_send=pcm_data,
                should_finalize=False,
                state=self._state,
                is_speech=is_speech,
            )
            self._record_prometheus_audio()
            return output

        # Active mode: state machine
        prev_state = self._state
        output = self._update_state(pcm_data, is_speech, wall_time)

        if prev_state != self._state:
            logger.debug(
                'VADGate state %s->%s uid=%s session=%s speech=%s cursor=%.1fms',
                prev_state.value,
                self._state.value,
                self.uid,
                self.session_id,
                is_speech,
                self._audio_cursor_ms,
            )

        self._record_prometheus_audio()
        return output

    def _record_prometheus_audio(self) -> None:
        """Record newly finalized sent/skipped audio without double-counting pre-roll."""
        if self.mode not in ('active', 'shadow'):
            return
        bytes_per_second = self._sample_width * self.channels * self.sample_rate

        sent_delta = self._bytes_sent - self._prometheus_bytes_sent
        if sent_delta:
            OMI_VAD_GATE_AUDIO_SECONDS_TOTAL.labels(outcome='sent', mode=self.mode).inc(sent_delta / bytes_per_second)
            self._prometheus_bytes_sent = self._bytes_sent

        pending_bytes = sum(len(chunk) for chunk in self._pre_roll)
        skipped_bytes = max(0, self._bytes_received - self._bytes_sent - pending_bytes)
        skipped_delta = skipped_bytes - self._prometheus_bytes_skipped
        if skipped_delta:
            OMI_VAD_GATE_AUDIO_SECONDS_TOTAL.labels(outcome='skipped', mode=self.mode).inc(
                skipped_delta / bytes_per_second
            )
            self._prometheus_bytes_skipped = skipped_bytes

    def _update_state(self, pcm_data: bytes, is_speech: bool, wall_time: float) -> GateOutput:
        """State machine transition logic."""
        wall_rel = wall_time - self._first_audio_wall_time if self._first_audio_wall_time else 0.0
        chunk_duration_sec = len(pcm_data) / (self._sample_width * self.channels * self.sample_rate)
        chunk_ms = chunk_duration_sec * 1000.0

        if self._state == GateState.SILENCE:
            # Buffer for pre-roll (time-based eviction)
            self._pre_roll.append(pcm_data)
            self._pre_roll_total_ms += chunk_ms
            while self._pre_roll_total_ms > self._pre_roll_ms and len(self._pre_roll) > 1:
                evicted = self._pre_roll.popleft()
                evicted_ms = (len(evicted) / (self._sample_width * self.channels * self.sample_rate)) * 1000.0
                self._pre_roll_total_ms -= evicted_ms

            if is_speech:
                # Transition: SILENCE → SPEECH
                self._state = GateState.SPEECH
                # Emit pre-roll + current chunk
                pre_roll_audio = b''.join(self._pre_roll)
                self._pre_roll.clear()
                self._pre_roll_total_ms = 0.0

                # Record mapper checkpoint for pre-roll start
                pre_roll_duration = len(pre_roll_audio) / (self._sample_width * self.channels * self.sample_rate)
                pre_roll_wall_rel = max(0.0, wall_rel - pre_roll_duration + chunk_duration_sec)
                self.dg_wall_mapper.on_audio_sent(pre_roll_duration, pre_roll_wall_rel)
                self._bytes_sent += len(pre_roll_audio)
                self._last_send_wall_time = wall_time

                return GateOutput(
                    audio_to_send=pre_roll_audio,
                    should_finalize=False,
                    state=GateState.SPEECH,
                    is_speech=True,
                )
            else:
                # Stay in SILENCE: audio buffered in pre-roll (not yet skipped/sent)
                self.dg_wall_mapper.on_silence_skipped()
                return GateOutput(
                    audio_to_send=b'',
                    should_finalize=False,
                    state=GateState.SILENCE,
                    is_speech=False,
                )

        elif self._state == GateState.SPEECH:
            # Send audio to DG
            self.dg_wall_mapper.on_audio_sent(chunk_duration_sec, wall_rel)
            self._bytes_sent += len(pcm_data)
            self._last_send_wall_time = wall_time

            if not is_speech:
                # Transition: SPEECH → HANGOVER
                self._state = GateState.HANGOVER
                self._hangover_finalized = False

            return GateOutput(
                audio_to_send=pcm_data,
                should_finalize=False,
                state=self._state,
                is_speech=is_speech,
            )

        elif self._state == GateState.HANGOVER:
            time_since_speech_ms = self._audio_cursor_ms - self._last_speech_ms

            if is_speech:
                # Speech resumed: HANGOVER → SPEECH (no finalize needed)
                self._state = GateState.SPEECH
                self._hangover_finalized = False
                self.dg_wall_mapper.on_audio_sent(chunk_duration_sec, wall_rel)
                self._bytes_sent += len(pcm_data)
                self._last_send_wall_time = wall_time
                return GateOutput(
                    audio_to_send=pcm_data,
                    should_finalize=False,
                    state=GateState.SPEECH,
                    is_speech=True,
                )

            if time_since_speech_ms > self._hangover_ms:
                # Hangover expired: HANGOVER → SILENCE
                self._state = GateState.SILENCE
                need_finalize = not self._hangover_finalized
                if need_finalize:
                    self._finalize_count += 1
                self._hangover_finalized = False
                self._pre_roll.clear()
                self._pre_roll_total_ms = 0.0
                self._pre_roll.append(pcm_data)
                chunk_ms_local = (len(pcm_data) / (self._sample_width * self.channels * self.sample_rate)) * 1000.0
                self._pre_roll_total_ms = chunk_ms_local
                # pcm_data is buffered in pre-roll and will count as skipped if never sent
                self.dg_wall_mapper.on_silence_skipped()
                return GateOutput(
                    audio_to_send=b'',
                    should_finalize=need_finalize,
                    state=GateState.SILENCE,
                    is_speech=False,
                )

            # Mid-hangover finalize: flush DG transcript early while keeping audio flowing
            should_finalize_now = False
            if not self._hangover_finalized and time_since_speech_ms >= self._finalize_silence_ms:
                should_finalize_now = True
                self._hangover_finalized = True
                self._finalize_count += 1

            # Still in hangover: send audio
            self.dg_wall_mapper.on_audio_sent(chunk_duration_sec, wall_rel)
            self._bytes_sent += len(pcm_data)
            self._last_send_wall_time = wall_time
            return GateOutput(
                audio_to_send=pcm_data,
                should_finalize=should_finalize_now,
                state=GateState.HANGOVER,
                is_speech=False,
            )

        # Fallback: send everything
        return GateOutput(audio_to_send=pcm_data, is_speech=is_speech)

    def consume_speech_ms_delta(self) -> int:
        """Consume and reset the speech_ms delta since last call.

        Used by the usage recording loop to periodically flush speech_ms
        to Redis/Firestore for fair-use tracking (#5746).
        Thread-safe: called from the same asyncio task that feeds audio.
        """
        delta = int(self._speech_ms_delta)
        self._speech_ms_delta = 0.0
        return delta

    def get_metrics(self) -> Dict[str, Any]:
        """Return gate metrics for logging/monitoring."""
        total = self._chunks_total or 1
        bytes_skipped = max(0, self._bytes_received - self._bytes_sent)
        total_bytes = self._bytes_received or 1
        return {
            'chunks_total': self._chunks_total,
            'chunks_speech': self._chunks_speech,
            'chunks_silence': self._chunks_silence,
            'silence_ratio': self._chunks_silence / total,
            'finalize_count': self._finalize_count,
            'finalize_errors': self._finalize_errors,
            'bytes_received': self._bytes_received,
            'bytes_sent': self._bytes_sent,
            'bytes_skipped': bytes_skipped,
            'bytes_saved_ratio': bytes_skipped / total_bytes,
            'keepalive_count': self._keepalive_count,
            'speech_ms_total': self._speech_ms_total,
            'state': self._state.value,
            'mode': self.mode,
        }

    def to_json_log(self) -> Dict[str, Any]:
        """Return JSON-safe metrics with derived quality/cost fields."""
        metrics = self.get_metrics()
        total = metrics['chunks_total'] or 1
        return {
            'event': 'vad_gate_metrics',
            'uid': self.uid,
            'session_id': self.session_id,
            'session_duration_sec': self._audio_cursor_ms / 1000.0,
            'speech_ratio': metrics['chunks_speech'] / total,
            'estimated_savings_pct': metrics['bytes_saved_ratio'] * 100.0,
            **metrics,
        }

    def remap_segments(self, segments: List[Dict[str, Any]]) -> None:
        """Remap STT provider timestamps to wall-clock-relative if gate is active."""
        if self.mode == 'active':
            for seg in segments:
                seg['start'] = self.dg_wall_mapper.dg_to_wall_rel(seg['start'])
                seg['end'] = self.dg_wall_mapper.dg_to_wall_rel(seg['end'])

    def record_keepalive(self, wall_time: float) -> None:
        """Record a keepalive send using the gate public API."""
        self._keepalive_count += 1
        self._last_send_wall_time = wall_time


# ---------------------------------------------------------------------------
# Gated STT Socket — wraps any STTSocket with VAD gate
# ---------------------------------------------------------------------------
class GatedSTTSocket(STTSocket):
    """Wraps an STTSocket with built-in VAD gate.

    When gate is active:
      - send() runs VAD internally, only forwards speech audio to the STT provider
      - Automatically calls finalize() on speech→silence transitions
      - finish() flushes pending transcript before closing
    When gate is None or mode='shadow':
      - Acts as transparent pass-through

    This keeps all VAD logic out of transcribe.py.
    """

    def __init__(
        self, stt_connection: STTSocket, gate: Optional['VADStreamingGate'] = None, passthrough_audio: bool = False
    ):
        self._conn = stt_connection
        self._gate = gate
        self._passthrough_audio = passthrough_audio
        # Audio capture for transcript quality validation (off by default)
        self._capture_dir = os.getenv('VAD_GATE_AUDIO_CAPTURE_DIR', '')
        self._raw_file = None
        self._gated_file = None
        if self._capture_dir and gate:
            os.makedirs(self._capture_dir, exist_ok=True)
            session_id = gate.session_id or 'unknown'
            self._raw_file = open(os.path.join(self._capture_dir, f'{session_id}_raw.pcm'), 'wb')
            self._gated_file = open(os.path.join(self._capture_dir, f'{session_id}_gated.pcm'), 'wb')

    @property
    def is_connection_dead(self) -> bool:
        if isinstance(self._conn, STTSocket):  # type: ignore[reportUnnecessaryIsInstance]  # runtime safety check
            return self._conn.is_connection_dead
        return False

    @property
    def death_reason(self) -> Optional[str]:
        return self._conn.death_reason

    @property
    def typed_death_reason(self) -> Optional[str]:
        """Proxy the wrapped socket's typed rejection (None when untyped)."""
        return getattr(self._conn, 'typed_death_reason', None)

    def _counted(self, audio: bytes) -> bytes:
        """Count frames that are not whole 16-bit samples, as the provider receives them.

        The gate re-chunks, so a count taken before it does not describe what a provider
        gets. Deepgram accepts a misaligned frame silently; Velma rejects it and closes the
        session, so the live Deepgram path is where that risk is measurable.
        """
        if len(audio) % 2:
            OMI_LIVE_STT_MISALIGNED_FRAMES_TOTAL.labels(
                provider=type(self._conn).__name__,
                stage='provider_send',
            ).inc()
        return audio

    def send(self, data: bytes, wall_time: Optional[float] = None) -> bool:
        """Send audio through VAD gate and report whether it was accepted."""
        if self.is_connection_dead:
            return False
        if self._gate is None:
            return self._conn.send(self._counted(data))

        now = wall_time or time.time()
        try:
            gate_out = self._gate.process_audio(data, now)
        except Exception:
            logger.exception('VAD gate process error, falling back to direct send uid=%s', self._gate.uid)
            record_fallback(
                component='vad',
                from_mode='gated',
                to_mode='direct',
                reason='other',
                outcome='degraded',
            )
            self._gate.mode = 'off'  # Disable timestamp remapping in stream_transcript wrapper
            self._gate = None  # Disable gate for rest of session
            return self._conn.send(data)
        if self._raw_file:
            self._raw_file.write(data)
        if self._gated_file and gate_out.audio_to_send:
            self._gated_file.write(gate_out.audio_to_send)
        if self._passthrough_audio:
            accepted = self._conn.send(self._counted(data))
        elif gate_out.audio_to_send:
            accepted = self._conn.send(self._counted(gate_out.audio_to_send))
        else:
            # Deliberately filtered silence is accepted by the gate; it is not
            # a provider enqueue failure and must not terminate the session.
            accepted = True
        if gate_out.should_finalize:
            try:
                self._conn.finalize()
            except Exception:
                self._gate._finalize_errors += 1  # type: ignore[reportPrivateUsage]  # internal counter
                logger.warning('finalize failed uid=%s session=%s', self._gate.uid, self._gate.session_id)
                # A failed speech-boundary flush means the provider may have
                # dropped the pending utterance. Report the send as rejected so
                # the shared live-STT boundary delivers a terminal failure to
                # the client instead of continuing as if transcription worked.
                return False
        return accepted

    def finalize(self) -> None:
        """Flush pending transcript."""
        self._conn.finalize()

    def finish(self) -> None:
        """Close STT connection. Flushes first if gate is active."""
        if self._gate is not None and self._gate.mode == 'active':
            try:
                self._conn.finalize()
            except Exception:
                self._gate._finalize_errors += 1  # type: ignore[reportPrivateUsage]  # internal counter
                logger.warning('finalize in finish() failed uid=%s session=%s', self._gate.uid, self._gate.session_id)
        self._conn.finish()
        for f in (self._raw_file, self._gated_file):
            if f:
                try:
                    f.close()
                except Exception:
                    pass

    def remap_segments(self, segments: List[Dict[str, Any]]) -> None:
        """Remap STT provider timestamps from audio-time to wall-clock-relative time."""
        if self._gate is not None:
            self._gate.remap_segments(segments)

    def get_metrics(self) -> Optional[Dict[str, Any]]:
        """Return gate metrics, or None if no gate."""
        if self._gate is not None:
            return self._gate.get_metrics()
        return None

    @property
    def is_gated(self) -> bool:
        return self._gate is not None


# Backward-compatibility aliases
GatedDeepgramSocket = GatedSTTSocket
DgWallMapper = WallTimeMapper