from __future__ import annotations

import logging
import os
import time
from collections.abc import Iterator
from dataclasses import dataclass
from queue import Queue
from threading import Event
from typing import Any, TypeAlias

import numpy as np
import torch

from speech_to_speech.api.openai_realtime.runtime_config import RuntimeConfig
from speech_to_speech.baseHandler import BaseHandler
from speech_to_speech.pipeline.events import SpeechStartedEvent, SpeechStoppedEvent
from speech_to_speech.pipeline.handler_types import VADIn, VADOut
from speech_to_speech.pipeline.messages import VADAudio
from speech_to_speech.pipeline.queue_types import TextEventItem
from speech_to_speech.pipeline.speculative_turns import SpeculativeTurnTracker
from speech_to_speech.utils.utils import int2float
from speech_to_speech.VAD.vad_iterator import VADIterator

logger = logging.getLogger(__name__)

VADInput: TypeAlias = bytes | tuple[bytes, RuntimeConfig]


@dataclass
class _PendingShortSegment:
    audio: np.ndarray
    active_ms: float
    start_ms: int
    end_ms: int


# Fragments with less active speech than this are treated as noise and never
# held for stitching, so sub-threshold bursts cannot sum past min_speech_ms
# and fire a false barge-in.
_SHORT_SEGMENT_MIN_FRAGMENT_MS = 100


# Optional import for audio enhancement
try:
    from df.enhance import enhance, init_df

    HAS_DF = True
except (ImportError, ModuleNotFoundError) as e:
    HAS_DF = False
    logger.warning(f"DeepFilterNet not available for audio enhancement: {e}")


class VADHandler(BaseHandler[VADIn, VADOut]):
    """
    Handles voice activity detection. When voice activity is detected, audio will be accumulated until the end of speech is detected and then passed
    to the following part.
    """

    def setup(
        self,
        should_listen: Event,
        thresh: float = 0.6,
        sample_rate: int = 16000,
        min_silence_ms: int = 64,
        min_speech_ms: int = 384,
        min_speech_continuation_ms: int = 192,
        max_speech_ms: float = float("inf"),
        speech_pad_ms: int = 30,
        audio_enhancement: bool = False,
        enable_realtime_transcription: bool = False,
        realtime_processing_pause: float = 0.5,
        text_output_queue: Queue[TextEventItem] | None = None,
        speculative_turns: SpeculativeTurnTracker | None = None,
        speculative_reopen_ms: int = 1000,
        unanswered_reopen_ms: int = 7000,
        short_segment_merge_ms: int = 0,
    ) -> None:
        self.should_listen = should_listen
        self.sample_rate = sample_rate
        self.min_silence_ms = min_silence_ms
        self.min_speech_ms = min_speech_ms
        self.min_speech_continuation_ms = self._resolve_min_speech_continuation_ms(
            self.min_speech_ms,
            min_speech_continuation_ms,
        )
        self.max_speech_ms = max_speech_ms
        self.enable_realtime_transcription = enable_realtime_transcription
        self.realtime_processing_pause = realtime_processing_pause
        self.text_output_queue = text_output_queue
        self.speculative_turns = speculative_turns
        self.speculative_reopen_ms = speculative_reopen_ms
        self.unanswered_reopen_ms = max(self.speculative_reopen_ms, unanswered_reopen_ms)
        self.short_segment_merge_ms = max(0, short_segment_merge_ms)
        self._last_turn_detection: dict | None = None
        hub_dir = torch.hub.get_dir()
        model_path = os.path.join(hub_dir, "snakers4_silero-vad_master", "silero_vad.jit")
        if os.path.isfile(model_path):
            self.model = torch.jit.load(model_path)
            self.model.eval()
        else:
            self.model, _ = torch.hub.load(
                "snakers4/silero-vad",
                "silero_vad",
                trust_repo=True,
                skip_validation=True,
            )
        self.iterator = VADIterator(
            self.model,
            threshold=thresh,
            sampling_rate=sample_rate,
            min_silence_duration_ms=min_silence_ms,
            speech_pad_ms=speech_pad_ms,
        )
        self.audio_enhancement = audio_enhancement
        if audio_enhancement:
            if not HAS_DF:
                logger.error(
                    "Audio enhancement requested but DeepFilterNet is not available. Disabling audio enhancement."
                )
                self.audio_enhancement = False
            else:
                self.enhanced_model, self.df_state, _ = init_df()

        # State for progressive audio release
        self.last_process_time: float = 0.0

        # Cumulative sample counter for audio_start_ms / audio_end_ms
        self._total_samples: int = 0

        # Throttled logging state (summary once per second)
        self._last_log_time = 0.0
        self._log_chunks = 0
        self._log_speech_starts = 0
        self._log_speech_ends = 0
        self._log_progressive_yields = 0
        self._speech_started_emitted = False
        self._turn_counter = 0
        self._current_turn_id: str | None = None
        self._current_turn_revision: int | None = None
        self._speculative_audio_prefix: np.ndarray | None = None
        self._last_final_wall_time: float | None = None
        self._last_final_audio_ms: int | None = None
        self._pending_reopen_candidate: tuple[str, int, int] | None = None
        self._pending_short_segment: _PendingShortSegment | None = None

    @property
    def _audio_ms(self) -> int:
        """Cumulative audio received so far, in milliseconds."""
        return int(self._total_samples / self.sample_rate * 1000)

    def _apply_runtime_turn_detection(self, runtime_config: RuntimeConfig | None = None) -> None:
        """Check RuntimeConfig for turn_detection changes and apply them."""
        audio = runtime_config.session.audio if runtime_config else None
        audio_input = audio.input if audio is not None else None
        if not runtime_config or not audio_input or not audio_input.turn_detection:
            return
        td_raw = audio_input.turn_detection

        # Convert Pydantic models (e.g. OpenAI SDK ServerVad) to dict;
        # plain dicts pass through unchanged.
        if hasattr(td_raw, "model_dump"):
            td = td_raw.model_dump(exclude_none=True)
        elif isinstance(td_raw, dict):
            td = td_raw
        else:
            logger.warning(f"Unexpected turn_detection type: {type(td_raw)}")
            return

        # Compare normalized snapshot (identity on td_raw vs stored dict was wrong after first apply).
        if td == self._last_turn_detection:
            return

        self._last_turn_detection = dict(td)

        if "threshold" in td:
            self.iterator.threshold = td["threshold"]
            logger.info(f"VAD threshold updated to {td['threshold']}")
        if "silence_duration_ms" in td:
            self.iterator.min_silence_samples = self.sample_rate * td["silence_duration_ms"] / 1000
            logger.info(f"VAD silence duration updated to {td['silence_duration_ms']}ms")

    def _start_new_turn(self) -> tuple[str, int]:
        self._cancel_pending_reopen()
        self._turn_counter += 1
        self._current_turn_id = f"turn_{self._turn_counter}"
        self._current_turn_revision = 0
        self._speculative_audio_prefix = None
        self._last_final_wall_time = None
        self._last_final_audio_ms = None
        if self.speculative_turns:
            self.speculative_turns.observe(self._current_turn_id, self._current_turn_revision)
        return self._current_turn_id, self._current_turn_revision

    def _speech_buffer_duration_ms(self) -> float:
        if not hasattr(self.iterator, "speech_buffer"):
            return 0.0
        buffer_samples = sum(len(t) for t in self.iterator.speech_buffer())
        return buffer_samples / self.sample_rate * 1000

    def _current_active_speech_duration_ms(self) -> float:
        active_speech_samples = getattr(self.iterator, "active_speech_samples", 0)
        return active_speech_samples / self.sample_rate * 1000

    def _last_utterance_active_speech_duration_ms(self) -> float:
        active_speech_samples = getattr(self.iterator, "last_utterance_active_speech_samples", 0)
        return active_speech_samples / self.sample_rate * 1000

    @staticmethod
    def _resolve_min_speech_continuation_ms(min_speech_ms: int, min_speech_continuation_ms: int) -> int:
        if min_speech_continuation_ms <= 0:
            return min_speech_ms
        return min(
            min_speech_ms,
            max(_SHORT_SEGMENT_MIN_FRAGMENT_MS, min_speech_continuation_ms),
        )

    def _uses_realtime_turn_handling(self) -> bool:
        return self.enable_realtime_transcription or self.speculative_turns is not None

    def _active_speech_min_ms(self, start_ms: int) -> float:
        """Duration hysteresis for speech that continues a reopenable turn."""
        if self._pending_reopen_candidate is not None or self._should_reopen_current_turn(start_ms):
            return self.min_speech_continuation_ms
        return self.min_speech_ms

    def _should_reopen_current_turn(self, audio_start_ms: int) -> bool:
        if not self._uses_realtime_turn_handling():
            return False
        if self._current_turn_id is None or self._current_turn_revision is None or self._last_final_audio_ms is None:
            return False

        is_committed = self.speculative_turns is not None and self.speculative_turns.is_committed(
            self._current_turn_id,
            self._current_turn_revision,
        )
        if is_committed:
            return False

        # Elapsed is measured on the audio clock, so the window only advances
        # while the client streams audio (continuous capture behaves like wall
        # time; push-to-talk style gaps freeze it).
        elapsed_ms = max(0, audio_start_ms - self._last_final_audio_ms)

        # Within the short grace window, any uncommitted turn may reopen.
        # Beyond it, an unanswered turn (no assistant output committed yet)
        # remains reopenable up to the unanswered_reopen_ms sanity cap, so a
        # user pause longer than speculative_reopen_ms does not orphan a turn
        # the assistant has not replied to. The cap also bounds the
        # empty-transcript case, where no request is queued and the turn would
        # otherwise never commit.
        reopen_limit_ms = self.speculative_reopen_ms
        if self.speculative_turns is not None:
            reopen_limit_ms = self.unanswered_reopen_ms
        return elapsed_ms <= reopen_limit_ms

    def _begin_pending_reopen_if_needed(self, audio_start_ms: int) -> None:
        if self._pending_reopen_candidate is not None or not self._should_reopen_current_turn(audio_start_ms):
            return
        if self.speculative_turns is None:
            return
        candidate_revision = self.speculative_turns.begin_reopen_candidate(
            self._current_turn_id,
            self._current_turn_revision,
        )
        if candidate_revision is None or self._current_turn_id is None or self._current_turn_revision is None:
            return
        self._pending_reopen_candidate = (
            self._current_turn_id,
            self._current_turn_revision,
            candidate_revision,
        )
        logger.info(
            "VAD: pending reopen candidate for speculative turn %s revision %d",
            self._current_turn_id,
            candidate_revision,
        )

    def _cancel_pending_reopen(self) -> None:
        if self._pending_reopen_candidate is None:
            return
        turn_id, _base_revision, candidate_revision = self._pending_reopen_candidate
        if self.speculative_turns:
            self.speculative_turns.cancel_reopen_candidate(turn_id, candidate_revision)
        self._pending_reopen_candidate = None

    def _confirm_pending_reopen(self) -> tuple[str, int, bool] | None:
        if self._pending_reopen_candidate is None:
            return None
        turn_id, base_revision, candidate_revision = self._pending_reopen_candidate
        self._pending_reopen_candidate = None
        if self.speculative_turns and not self.speculative_turns.confirm_reopen_candidate(
            turn_id,
            base_revision,
            candidate_revision,
        ):
            return None
        self._current_turn_id = turn_id
        self._current_turn_revision = candidate_revision
        logger.info("VAD: reopened speculative turn %s revision %d", turn_id, candidate_revision)
        return turn_id, candidate_revision, True

    def _reopen_current_turn(self) -> tuple[str, int, bool] | None:
        if self._current_turn_id is None or self._current_turn_revision is None:
            return None

        turn_id = self._current_turn_id
        base_revision = self._current_turn_revision
        if self.speculative_turns is not None:
            candidate_revision = self.speculative_turns.begin_reopen_candidate(turn_id, base_revision)
            if candidate_revision is None or not self.speculative_turns.confirm_reopen_candidate(
                turn_id,
                base_revision,
                candidate_revision,
            ):
                return None
        else:
            candidate_revision = base_revision + 1

        self._current_turn_id = turn_id
        self._current_turn_revision = candidate_revision
        logger.info("VAD: reopened speculative turn %s revision %d", turn_id, candidate_revision)
        return turn_id, candidate_revision, True

    def _ensure_turn_for_speech_start(self, audio_start_ms: int) -> tuple[str, int, bool]:
        if (
            self._speech_started_emitted
            and self._current_turn_id is not None
            and self._current_turn_revision is not None
        ):
            return self._current_turn_id, self._current_turn_revision, False

        confirmed_reopen = self._confirm_pending_reopen()
        if confirmed_reopen is not None:
            return confirmed_reopen

        reopened = False
        if self._should_reopen_current_turn(audio_start_ms):
            reopened_turn = self._reopen_current_turn()
            if reopened_turn is not None:
                return reopened_turn

        self._start_new_turn()

        if self._current_turn_id is None or self._current_turn_revision is None:
            raise RuntimeError("VAD failed to allocate turn metadata")
        if self.speculative_turns:
            self.speculative_turns.observe(self._current_turn_id, self._current_turn_revision)
        return self._current_turn_id, self._current_turn_revision, reopened

    def _current_turn_metadata(self) -> tuple[str | None, int | None]:
        return self._current_turn_id, self._current_turn_revision

    def _combined_turn_audio(self, current_segment: np.ndarray) -> np.ndarray:
        if self._speculative_audio_prefix is None:
            return current_segment
        return np.concatenate((self._speculative_audio_prefix, current_segment))

    def _short_segment_merge_window_ms(self) -> int:
        return int(getattr(self, "short_segment_merge_ms", 0))

    def _segment_duration_ms(self, segment: np.ndarray) -> float:
        return len(segment) / self.sample_rate * 1000

    def _segment_start_ms(self, segment: np.ndarray, end_ms: int) -> int:
        return max(0, end_ms - int(self._segment_duration_ms(segment)))

    def _short_segment_gap_ms(self, start_ms: int) -> float:
        if self._pending_short_segment is None:
            return float("inf")
        return max(0, start_ms - self._pending_short_segment.end_ms)

    def _can_merge_pending_short_segment(self, start_ms: int) -> bool:
        return (
            self._pending_short_segment is not None
            and self._short_segment_merge_window_ms() > 0
            and self._short_segment_gap_ms(start_ms) <= self._short_segment_merge_window_ms()
        )

    def _effective_active_speech_for_start(self, start_ms: int, active_ms: float) -> tuple[int, float]:
        # A live fragment below the noise floor never counts the held segment
        # toward the speech-start threshold, mirroring the finalization path.
        if active_ms < _SHORT_SEGMENT_MIN_FRAGMENT_MS:
            return start_ms, active_ms
        if not self._can_merge_pending_short_segment(start_ms):
            return start_ms, active_ms
        assert self._pending_short_segment is not None
        return self._pending_short_segment.start_ms, self._pending_short_segment.active_ms + active_ms

    def _merge_pending_short_segment(
        self,
        segment: np.ndarray,
        active_ms: float,
        end_ms: int,
    ) -> tuple[np.ndarray, float, int, bool]:
        start_ms = self._segment_start_ms(segment, end_ms)
        if not self._can_merge_pending_short_segment(start_ms):
            self._discard_expired_pending_short_segment(start_ms)
            return segment, active_ms, start_ms, False

        pending = self._pending_short_segment
        assert pending is not None
        # Reinsert the silence between the two segments so the stitched audio
        # keeps its acoustic gap and its length matches the audio-clock span.
        gap_samples = int(self._short_segment_gap_ms(start_ms) * self.sample_rate / 1000)
        self._pending_short_segment = None
        parts = [pending.audio]
        if gap_samples > 0:
            parts.append(np.zeros(gap_samples, dtype=segment.dtype))
        parts.append(segment)
        merged = np.concatenate(parts)
        return merged, pending.active_ms + active_ms, pending.start_ms, True

    def _hold_short_segment(self, segment: np.ndarray, active_ms: float, start_ms: int, end_ms: int) -> None:
        self._pending_short_segment = _PendingShortSegment(
            audio=segment,
            active_ms=active_ms,
            start_ms=start_ms,
            end_ms=end_ms,
        )
        logger.info(
            "VAD: holding short segment=%.0fms active=%.0fms (active_min=%sms, merge_max=%sms)",
            self._segment_duration_ms(segment),
            active_ms,
            self.min_speech_ms,
            self._short_segment_merge_window_ms(),
        )

    def _discard_pending_short_segment(self, reason: str = "expired") -> None:
        pending = self._pending_short_segment
        if pending is None:
            return
        self._pending_short_segment = None
        logger.info(
            "VAD: discarding held short segment=%.0fms active=%.0fms (%s, active_min=%sms)",
            self._segment_duration_ms(pending.audio),
            pending.active_ms,
            reason,
            self.min_speech_ms,
        )

    def _discard_expired_pending_short_segment(self, next_start_ms: int | None = None) -> None:
        pending = self._pending_short_segment
        if pending is None or self._short_segment_merge_window_ms() <= 0:
            return
        reference_ms = self._audio_ms if next_start_ms is None else next_start_ms
        gap_ms = max(0, reference_ms - pending.end_ms)
        if gap_ms > self._short_segment_merge_window_ms():
            self._discard_pending_short_segment("merge window elapsed")

    def before_emit_output(self, output: VADOut) -> None:
        if isinstance(output, VADAudio):
            self._drop_superseded_vad_audio(output)

    def _drop_superseded_vad_audio(self, latest: VADAudio) -> int:
        if not hasattr(self.queue_out, "mutex") or not hasattr(self.queue_out, "queue"):
            return 0

        dropped = 0
        with self.queue_out.mutex:
            kept: list[Any] = []
            while self.queue_out.queue:
                queued_item = self.queue_out.queue.popleft()
                if isinstance(queued_item, VADAudio) and self._vad_audio_is_superseded(
                    queued_item,
                    latest,
                ):
                    dropped += 1
                else:
                    kept.append(queued_item)
            self.queue_out.queue.extend(kept)
            if dropped:
                self.queue_out.not_full.notify_all()

        if dropped:
            logger.debug(
                "VAD: dropped %d superseded audio chunk(s) before enqueueing turn=%s rev=%s mode=%s",
                dropped,
                latest.turn_id,
                latest.turn_revision,
                latest.mode,
            )
        return dropped

    def _vad_audio_is_superseded(self, queued_item: VADAudio, latest: VADAudio) -> bool:
        if queued_item.turn_id is None or queued_item.turn_revision is None:
            return False
        if self.speculative_turns is not None and not self.speculative_turns.is_latest(
            queued_item.turn_id,
            queued_item.turn_revision,
        ):
            return True
        return (
            queued_item.mode == "progressive"
            and queued_item.turn_id == latest.turn_id
            and queued_item.turn_revision == latest.turn_revision
        )

    def process(self, audio_chunk: VADIn) -> Iterator[VADOut]:
        runtime_config = None
        if isinstance(audio_chunk, tuple):
            audio_chunk, runtime_config = audio_chunk
        self._apply_runtime_turn_detection(runtime_config)

        if not self.should_listen.is_set():
            return

        # Normal listening mode
        self._log_chunks += 1
        audio_int16 = np.frombuffer(audio_chunk, dtype=np.int16)
        self._total_samples += len(audio_int16)
        audio_float32 = int2float(audio_int16)

        vad_output = self.iterator(torch.from_numpy(audio_float32))

        # Deferred speech_started: only emit once active VAD speech reaches the valid speech threshold.
        is_triggered_now = self.iterator.triggered
        if is_triggered_now and not self._speech_started_emitted:
            segment_samples = sum(len(t) for t in self.iterator.buffer)
            segment_duration_ms = segment_samples / self.sample_rate * 1000
            active_speech_duration_ms = self._current_active_speech_duration_ms()
            speech_buffer_duration_ms = self._speech_buffer_duration_ms()
            start_ms = max(0, self._audio_ms - int(speech_buffer_duration_ms))
            effective_start_ms, effective_active_speech_duration_ms = self._effective_active_speech_for_start(
                start_ms,
                active_speech_duration_ms,
            )
            self._begin_pending_reopen_if_needed(effective_start_ms)
            active_speech_min_ms = self._active_speech_min_ms(effective_start_ms)
            if effective_active_speech_duration_ms >= active_speech_min_ms:
                turn_id, turn_revision, reopened = self._ensure_turn_for_speech_start(effective_start_ms)
                self._speech_started_emitted = True
                self._log_speech_starts += 1
                logger.info(
                    "Speech started (confirmed, active=%.0fms, min=%.0fms, segment=%.0fms, turn=%s rev=%s)",
                    effective_active_speech_duration_ms,
                    active_speech_min_ms,
                    segment_duration_ms,
                    turn_id,
                    turn_revision,
                )
                if self.text_output_queue:
                    self.text_output_queue.put(
                        SpeechStartedEvent(
                            audio_start_ms=effective_start_ms,
                            turn_id=turn_id,
                            turn_revision=turn_revision,
                            reopened=reopened,
                        )
                    )
        elif not is_triggered_now and vad_output is None:
            self._discard_expired_pending_short_segment()

        # Log a summary once per second instead of every chunk
        now = time.time()
        if now - self._last_log_time >= 1.0:
            state = "SPEAKING" if is_triggered_now else "silent"
            logger.debug(
                f"VAD: {self._log_chunks} chunks/s | {state} | "
                f"starts={self._log_speech_starts} ends={self._log_speech_ends} progressive={self._log_progressive_yields}"
            )
            self._log_chunks = 0
            self._log_speech_starts = 0
            self._log_speech_ends = 0
            self._log_progressive_yields = 0
            self._last_log_time = now

        if self._uses_realtime_turn_handling():
            # Realtime mode keeps turns reopenable; live transcription additionally
            # emits progressive audio chunks while speaking.
            yield from self._process_realtime(vad_output)
        else:
            # Original mode: yield only when speech ends
            yield from self._process_normal(vad_output)

    def _process_realtime(self, vad_output: list[torch.Tensor] | None) -> Iterator[VADOut]:
        """Process with real-time progressive audio release."""
        # Check if we're currently in a speech segment.
        if self.enable_realtime_transcription and hasattr(self.iterator, "buffer") and len(self.iterator.buffer) > 0:
            current_time = time.time()
            duration_ms = self._speech_buffer_duration_ms()
            progressive_pause = self._progressive_processing_pause(duration_ms)

            # Yield accumulated audio periodically while speaking
            if (current_time - self.last_process_time) >= progressive_pause:
                array = torch.cat(self.iterator.speech_buffer()).cpu().numpy()
                duration_ms = len(array) / self.sample_rate * 1000
                active_speech_duration_ms = self._current_active_speech_duration_ms()
                start_ms = max(0, self._audio_ms - int(duration_ms))

                if active_speech_duration_ms >= self._active_speech_min_ms(start_ms):
                    self._log_progressive_yields += 1
                    logger.debug(
                        "VAD: yielding progressive audio (segment=%.0fms, active=%.0fms, interval=%.2fs)",
                        duration_ms,
                        active_speech_duration_ms,
                        progressive_pause,
                    )
                    turn_id, turn_revision = self._current_turn_metadata()
                    yield VADAudio(
                        audio=self._combined_turn_audio(array),
                        mode="progressive",
                        turn_id=turn_id,
                        turn_revision=turn_revision,
                    )
                    self.last_process_time = current_time

        # Handle end of speech
        if vad_output is not None:
            if len(vad_output) == 0:
                logger.info("VAD: phantom trigger (empty buffer), closing speech pair")
                if self._speech_started_emitted and self.text_output_queue:
                    turn_id, turn_revision = self._current_turn_metadata()
                    self.text_output_queue.put(
                        SpeechStoppedEvent(
                            audio_end_ms=self._audio_ms,
                            turn_id=turn_id,
                            turn_revision=turn_revision,
                        )
                    )
                if not self._speech_started_emitted:
                    self._cancel_pending_reopen()
                self._speech_started_emitted = False
                self._discard_expired_pending_short_segment()
                return

            array = torch.cat(vad_output).cpu().numpy()
            end_ms = self._audio_ms
            raw_active_ms = self._last_utterance_active_speech_duration_ms()
            active_speech_duration_ms = raw_active_ms
            stitched_short_segment = False
            # Fragments below the noise floor never merge with or replace a
            # held segment; the pending segment's own expiry handles it.
            if raw_active_ms >= _SHORT_SEGMENT_MIN_FRAGMENT_MS:
                array, active_speech_duration_ms, start_ms, stitched_short_segment = self._merge_pending_short_segment(
                    array,
                    active_speech_duration_ms,
                    end_ms,
                )
            else:
                start_ms = self._segment_start_ms(array, end_ms)
            duration_ms = self._segment_duration_ms(array)
            min_active_ms = 0.0 if self._speech_started_emitted else self._active_speech_min_ms(start_ms)

            if active_speech_duration_ms < min_active_ms or duration_ms > self.max_speech_ms:
                if (
                    self._short_segment_merge_window_ms() > 0
                    and raw_active_ms >= _SHORT_SEGMENT_MIN_FRAGMENT_MS
                    and active_speech_duration_ms < min_active_ms
                    and duration_ms <= self.max_speech_ms
                ):
                    self._hold_short_segment(array, active_speech_duration_ms, start_ms, end_ms)
                else:
                    logger.info(
                        "VAD: discarding segment=%.0fms active=%.0fms (active_min=%sms, segment_max=%sms)",
                        duration_ms,
                        active_speech_duration_ms,
                        min_active_ms,
                        self.max_speech_ms,
                    )
                if self._speech_started_emitted and self.text_output_queue:
                    turn_id, turn_revision = self._current_turn_metadata()
                    self.text_output_queue.put(
                        SpeechStoppedEvent(
                            audio_end_ms=self._audio_ms,
                            turn_id=turn_id,
                            turn_revision=turn_revision,
                        )
                    )
                if not self._speech_started_emitted:
                    self._cancel_pending_reopen()
                self._speech_started_emitted = False
            else:
                if stitched_short_segment:
                    logger.info(
                        "VAD: stitched short segment(s) into segment=%.0fms active=%.0fms",
                        duration_ms,
                        active_speech_duration_ms,
                    )
                if not self._speech_started_emitted:
                    turn_id, turn_revision, reopened = self._ensure_turn_for_speech_start(start_ms)
                    if self.text_output_queue:
                        self.text_output_queue.put(
                            SpeechStartedEvent(
                                audio_start_ms=start_ms,
                                turn_id=turn_id,
                                turn_revision=turn_revision,
                                reopened=reopened,
                                interrupt_response=False,
                            )
                        )
                else:
                    turn_id, turn_revision = self._current_turn_metadata()
                self._log_speech_ends += 1
                logger.info(
                    "Speech soft-ended (segment=%.0fms, active=%.0fms, turn=%s rev=%s)",
                    duration_ms,
                    active_speech_duration_ms,
                    turn_id,
                    turn_revision,
                )
                if self.audio_enhancement:
                    array = self._apply_audio_enhancement(array)
                output_array = self._combined_turn_audio(array)
                combined_duration_s = len(output_array) / self.sample_rate
                if self.text_output_queue:
                    self.text_output_queue.put(
                        SpeechStoppedEvent(
                            duration_s=combined_duration_s,
                            audio_end_ms=end_ms,
                            turn_id=turn_id,
                            turn_revision=turn_revision,
                        )
                    )
                self._speculative_audio_prefix = output_array
                self._last_final_wall_time = time.time()
                self._last_final_audio_ms = end_ms
                if self.speculative_turns:
                    # The grace window only delays response commits; reopen
                    # eligibility for unanswered turns is extended separately
                    # via unanswered_reopen_ms in _should_reopen_current_turn.
                    self.speculative_turns.start_reopen_grace(
                        turn_id,
                        turn_revision,
                        self.speculative_reopen_ms / 1000.0,
                    )
                else:
                    self.should_listen.clear()
                yield VADAudio(audio=output_array, mode="final", turn_id=turn_id, turn_revision=turn_revision)
                self.last_process_time = 0.0
                self._speech_started_emitted = False

    def _progressive_processing_pause(self, duration_ms: float) -> float:
        base_pause = max(0.0, self.realtime_processing_pause)
        duration_s = duration_ms / 1000.0
        if duration_s < 8.0:
            multiplier = 1.0
        elif duration_s < 15.0:
            multiplier = 2.0
        elif duration_s < 30.0:
            multiplier = 4.0
        else:
            multiplier = 6.0
        return min(base_pause * multiplier, 2.0)

    def _process_normal(self, vad_output: list[torch.Tensor] | None) -> Iterator[VADOut]:
        """Original processing: yield only when speech ends."""
        if vad_output is not None:
            if len(vad_output) == 0:
                logger.info("VAD: phantom trigger (empty buffer), closing speech pair")
                if self._speech_started_emitted and self.text_output_queue:
                    self.text_output_queue.put(SpeechStoppedEvent(audio_end_ms=self._audio_ms))
                self._speech_started_emitted = False
                self._discard_expired_pending_short_segment()
                return

            array = torch.cat(vad_output).cpu().numpy()
            end_ms = self._audio_ms
            raw_active_ms = self._last_utterance_active_speech_duration_ms()
            active_speech_duration_ms = raw_active_ms
            stitched_short_segment = False
            # Fragments below the noise floor never merge with or replace a
            # held segment; the pending segment's own expiry handles it.
            if raw_active_ms >= _SHORT_SEGMENT_MIN_FRAGMENT_MS:
                array, active_speech_duration_ms, start_ms, stitched_short_segment = self._merge_pending_short_segment(
                    array,
                    active_speech_duration_ms,
                    end_ms,
                )
            else:
                start_ms = self._segment_start_ms(array, end_ms)
            duration_ms = self._segment_duration_ms(array)
            min_active_ms = 0.0 if self._speech_started_emitted else self._active_speech_min_ms(start_ms)
            if active_speech_duration_ms < min_active_ms or duration_ms > self.max_speech_ms:
                if (
                    self._short_segment_merge_window_ms() > 0
                    and raw_active_ms >= _SHORT_SEGMENT_MIN_FRAGMENT_MS
                    and active_speech_duration_ms < min_active_ms
                    and duration_ms <= self.max_speech_ms
                ):
                    self._hold_short_segment(array, active_speech_duration_ms, start_ms, end_ms)
                else:
                    logger.info(
                        "VAD: discarding segment=%.0fms active=%.0fms (active_min=%sms, segment_max=%sms)",
                        duration_ms,
                        active_speech_duration_ms,
                        min_active_ms,
                        self.max_speech_ms,
                    )
                if self._speech_started_emitted and self.text_output_queue:
                    self.text_output_queue.put(SpeechStoppedEvent(audio_end_ms=self._audio_ms))
                self._speech_started_emitted = False
            else:
                if stitched_short_segment:
                    logger.info(
                        "VAD: stitched short segment(s) into segment=%.0fms active=%.0fms",
                        duration_ms,
                        active_speech_duration_ms,
                    )
                if not self._speech_started_emitted and self.text_output_queue:
                    self.text_output_queue.put(
                        SpeechStartedEvent(
                            audio_start_ms=start_ms,
                            interrupt_response=False,
                        )
                    )
                self._log_speech_ends += 1
                self.should_listen.clear()
                logger.info(f"Speech ended ({duration_ms:.0f}ms), stop listening")
                if self.text_output_queue:
                    self.text_output_queue.put(SpeechStoppedEvent(duration_s=duration_ms / 1000.0, audio_end_ms=end_ms))
                if self.audio_enhancement:
                    array = self._apply_audio_enhancement(array)
                yield VADAudio(audio=array)
                self._speech_started_emitted = False

    def _apply_audio_enhancement(self, array: np.ndarray) -> np.ndarray:
        """Apply audio enhancement if enabled."""
        import torchaudio

        if self.sample_rate != self.df_state.sr():
            audio_float32 = torchaudio.functional.resample(
                torch.from_numpy(array),
                orig_freq=self.sample_rate,
                new_freq=self.df_state.sr(),
            )
            enhanced = enhance(
                self.enhanced_model,
                self.df_state,
                audio_float32.unsqueeze(0),
            )
            enhanced = torchaudio.functional.resample(
                enhanced,
                orig_freq=self.df_state.sr(),
                new_freq=self.sample_rate,
            )
        else:
            enhanced = enhance(self.enhanced_model, self.df_state, torch.from_numpy(array))
        return enhanced.numpy().squeeze()

    def on_session_end(self):
        self.iterator.reset_states()
        self._pending_short_segment = None
        self.iterator.buffer = []
        self.last_process_time = 0.0
        self._total_samples = 0
        self._speech_started_emitted = False
        self._turn_counter = 0
        self._current_turn_id = None
        self._current_turn_revision = None
        self._speculative_audio_prefix = None
        self._last_final_wall_time = None
        self._last_final_audio_ms = None
        self._pending_reopen_candidate = None
        if self.speculative_turns:
            self.speculative_turns.reset()
        self.should_listen.set()
        logger.debug("VAD session state reset")

    @property
    def min_time_to_debug(self) -> float:
        return 0.00001
