main.py

"""
Hugging Face Space: Whisper ASR HTTP API (same contract as ASR/src/web_realtime ``/audio``).

POST /audio
  Body: raw little-endian float32 PCM (mono).
  Header: X-Sample-Rate: <Hz> (default 16000)

Response JSON: {"text": "......."}
"""

from __future__ import annotations

import copy
import json
import math
import os
import re
import struct
import zlib
import threading
import time
from typing import Optional

import numpy as np
import torch
import torch.nn.functional as F
import torchaudio
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from starlette.requests import ClientDisconnect
from transformers import WhisperForConditionalGeneration, WhisperProcessor

SAMPLE_RATE = 16_000

def _resolve_model_path(env_key: str, default_local: str, default_remote: str) -> str:
    val = os.environ.get(env_key, "").strip()
    if val:
        return val
    local_path = os.path.join(os.path.dirname(__file__), default_local)
    if os.path.isdir(local_path):
        return local_path
    return default_remote

MODEL_ID_EN = _resolve_model_path("MODEL_ID_EN", "whisper_finetuned", "Thedeezat/ASR-Hearing-Impaired")
MODEL_ID_IT = _resolve_model_path(
    "MODEL_ID_IT",
    "whisper_large_v3_turbo",
    "openai/whisper-large-v3-turbo",
)
# Primary single-pass model for low-latency inference (no language-detect pre-pass).
MODEL_ID_PRIMARY = (
    os.environ.get("MODEL_ID_PRIMARY", "").strip()
    or os.environ.get("MODEL_ID", "").strip()
    or MODEL_ID_IT
)
MODEL_ID_FALLBACK = os.environ.get("MODEL_ID_FALLBACK", "").strip() or MODEL_ID_EN
_FALLBACK_ON_EMPTY = os.environ.get("ASR_FALLBACK_ON_EMPTY", "0").strip().lower() in ("1", "true", "yes")
# Defaults favor quality; override for faster (lower quality) inference.
_WHISPER_MAX_NEW = max(32, int(os.environ.get("ASR_WHISPER_MAX_NEW_TOKENS", "224")))
_WHISPER_BEAMS = max(1, int(os.environ.get("ASR_WHISPER_NUM_BEAMS", "1")))
_MIN_RMS = float(os.environ.get("ASR_MIN_RMS", "0.004"))
# On CUDA, fp16 autocast is much faster with minimal WER change for Whisper (default on).
_USE_FP16_CUDA = os.environ.get("ASR_USE_FP16", "1").strip().lower() in ("1", "true", "yes")
_DUAL_DECODE = os.environ.get("ASR_DUAL_DECODE", "0").strip().lower() in ("1", "true", "yes")
_NO_SPEECH_THRESHOLD = float(os.environ.get("ASR_NO_SPEECH_THRESHOLD", "0.6"))
# Forced-language overrides (faster-whisper): higher = less decoding on “silent” frames → fewer invented phrases on HoloLens.
_NO_SPEECH_THRESHOLD_EN = float(os.environ.get("ASR_NO_SPEECH_THRESHOLD_EN", "0.60"))
_NO_SPEECH_THRESHOLD_IT = float(os.environ.get("ASR_NO_SPEECH_THRESHOLD_IT", "0.84"))
# Lower threshold = reject repetitive / gibberish text sooner (helps tail hallucinations).
_COMPRESSION_RATIO_THRESHOLD = float(os.environ.get("ASR_COMPRESSION_RATIO_THRESHOLD", "1.6"))
# Italian-only zlib/text compression gate — keeps EN/global defaults stable when transformers omits nsp/alp.
_COMPRESSION_RATIO_THRESHOLD_IT = float(os.environ.get("ASR_COMPRESSION_RATIO_THRESHOLD_IT", "2.8"))
_ASR_ULTRA_SHORT_TAIL_WINDOW_S = float(os.environ.get("ASR_ULTRA_SHORT_TAIL_WINDOW_S", "3.0"))
_ASR_ULTRA_SHORT_TAIL_MAX_SEGMENT_S = float(os.environ.get("ASR_ULTRA_SHORT_TAIL_MAX_SEGMENT_S", "0.75"))
_LOGPROB_THRESHOLD = float(os.environ.get("ASR_LOGPROB_THRESHOLD", "-1.0"))
# Transformers forced-Italian: reject when average token logprob is below this (more negative = stricter).
_ASR_AVG_LOGPROB_REJECT_IT = float(os.environ.get("ASR_AVG_LOGPROB_REJECT_IT", "-0.92"))
_ASR_AVG_LOGPROB_REJECT_EN = float(os.environ.get("ASR_AVG_LOGPROB_REJECT_EN", "-0.90"))
_ASR_MAX_WORDS_PER_SECOND_EN = float(os.environ.get("ASR_MAX_WORDS_PER_SECOND_EN", "6.0"))
_CPU_THREADS = max(1, int(os.environ.get("ASR_CPU_THREADS", str(os.cpu_count() or 4))))
_CPU_INTEROP_THREADS = max(1, int(os.environ.get("ASR_CPU_INTEROP_THREADS", "1")))
_ASR_BACKEND = (os.environ.get("ASR_BACKEND", "faster-whisper" if os.environ.get("ASR_BACKEND") is None else "").strip().lower() or "faster-whisper")
_FAST_MODEL_IT = (os.environ.get("ASR_FAST_MODEL_IT") or "distil-large-v3").strip()
_FAST_MODEL_EN = (os.environ.get("ASR_FAST_MODEL_EN") or _FAST_MODEL_IT).strip()
_FAST_COMPUTE_TYPE = (os.environ.get("ASR_FAST_COMPUTE_TYPE") or "").strip()
_FAST_REALTIME_MODEL_EN = (os.environ.get("ASR_FAST_REALTIME_MODEL_EN") or "medium.en").strip()
_FAST_REALTIME_MODEL_IT = (os.environ.get("ASR_FAST_REALTIME_MODEL_IT") or "medium").strip()
_FAST_USE_REALTIME_FOR_FORCED_RAW = (os.environ.get("ASR_FAST_USE_REALTIME_FOR_FORCED") or "").strip().lower()
_ASR_DROP_WHEN_BUSY_RAW = (os.environ.get("ASR_DROP_WHEN_BUSY") or "").strip().lower()
_ASR_RMS_SKIP_FORCED = float(os.environ.get("ASR_RMS_SKIP_FORCED", "0.0035"))
_ASR_RMS_SKIP_AUTO = float(os.environ.get("ASR_RMS_SKIP_AUTO", "0.0045"))
_GPU_SAFE_PROFILE = os.environ.get("ASR_GPU_SAFE_PROFILE", "1").strip().lower() in ("1", "true", "yes")
_ASR_ITALIAN_BACKEND_RAW = (os.environ.get("ASR_ITALIAN_BACKEND") or "").strip().lower()
_CPU_RELIABLE_MODE_RAW = (os.environ.get("ASR_CPU_RELIABLE_MODE") or "").strip().lower()
_ASR_CPU_AUTO_USE_REALTIME = (os.environ.get("ASR_CPU_AUTO_USE_REALTIME", "1").strip().lower() in ("1", "true", "yes"))
_ASR_FORCED_NUM_BEAMS_RAW = (os.environ.get("ASR_FORCED_NUM_BEAMS") or "").strip()
_ASR_FORCED_MAX_NEW_RAW = (os.environ.get("ASR_FORCED_MAX_NEW_TOKENS") or "").strip()
_ASR_ENGLISH_FAST_ON_CUDA = (os.environ.get("ASR_ENGLISH_FAST_ON_CUDA", "1").strip().lower() in ("1", "true", "yes"))
_ASR_ENGLISH_RERUN_ON_SUSPECT = (
    os.environ.get("ASR_ENGLISH_RERUN_ON_SUSPECT", "1").strip().lower() in ("1", "true", "yes")
)
_ASR_EN_SHORT_RETRY = (
    os.environ.get("ASR_EN_SHORT_RETRY", "0").strip().lower() in ("1", "true", "yes")
)
_ASR_EN_SHORT_RETRY_NS_THR = float(os.environ.get("ASR_EN_SHORT_RETRY_NO_SPEECH_THRESHOLD", "0.42"))
_ASR_IT_SHORT_RETRY = (
    os.environ.get("ASR_IT_SHORT_RETRY", "0").strip().lower() in ("1", "true", "yes")
)
_ASR_IT_SHORT_RETRY_NS_THR = float(os.environ.get("ASR_IT_SHORT_RETRY_NO_SPEECH_THRESHOLD", "0.52"))
_ASR_MAX_CHUNK_S_FORCED = float(os.environ.get("ASR_MAX_CHUNK_SECONDS_FORCED", "4.0"))
_ASR_MAX_CHUNK_S_AUTO = float(os.environ.get("ASR_MAX_CHUNK_SECONDS_AUTO", "4.8"))
_ASR_CHUNK_OVERLAP_S = float(os.environ.get("ASR_CHUNK_OVERLAP_SECONDS", "0.2"))
# Do not decode forced Italian chunks shorter than this duration.
# Disable explicitly with ASR_MIN_CHUNK_SECONDS=0.
_ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN = float(os.environ.get("ASR_MIN_CHUNK_SECONDS", "1.2"))
# Speech-window RMS thresholds for forced Italian.
_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT = float(os.environ.get("ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT", "0.016"))
_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN = float(os.environ.get("ASR_MIN_SPEECH_RMS_FORCED_ITALIAN", "0.018"))
_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_ACCEPT = float(
    os.environ.get("ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_ACCEPT", "0.018")
)
_ASR_MIN_SPEECH_RMS_FORCED_ENGLISH_ACCEPT = float(
    os.environ.get("ASR_MIN_SPEECH_RMS_FORCED_ENGLISH_ACCEPT", "0.018")
)
# Pair high English words/sec with another bad signal (do not drop fast speech on WPS alone).
_WPS_EN_PAIR_NO_SPEECH = float(os.environ.get("ASR_WPS_EN_PAIR_NO_SPEECH", "0.55"))
_WPS_EN_PAIR_AVG_LOGPROB = float(os.environ.get("ASR_WPS_EN_PAIR_AVG_LOGPROB", "-0.90"))
_WPS_CAPTION_HINT_SUBSTRINGS: frozenset[str] = frozenset(
    {
        "thank you for watching",
        "thanks for watching",
        "please subscribe",
        "like and subscribe",
        "subscribe to my channel",
    }
)
_ASR_SHORT_MIN_CHUNK_S_FORCED_ITALIAN = float(os.environ.get("ASR_SHORT_MIN_CHUNK_SECONDS_FORCED_ITALIAN", "1.4"))
_ASR_SHORT_MAX_CHUNK_S_FORCED_ITALIAN = float(os.environ.get("ASR_SHORT_MAX_CHUNK_SECONDS_FORCED_ITALIAN", "2.4"))
_ASR_MIN_TRIMMED_SEGMENT_SECONDS_EN = float(
    os.environ.get("ASR_MIN_TRIMMED_SEGMENT_SECONDS_EN", "0.45")
)
_ASR_MIN_TRIMMED_SEGMENT_SECONDS_IT = float(
    os.environ.get("ASR_MIN_TRIMMED_SEGMENT_SECONDS_IT", "0.45")
)
_ASR_DROP_MICRO_FRAGMENTS = (
    os.environ.get("ASR_DROP_MICRO_FRAGMENTS", "0").strip().lower() in ("1", "true", "yes")
)
# By default, do not blank near-duplicate phrases in forced-language mode (can hide valid
# repeated short words in realtime conversational Italian/English). Operators can opt in.
_ASR_BLANK_NEAR_DUP_FORCED = (
    os.environ.get("ASR_BLANK_NEAR_DUP_FORCED", "0").strip().lower() in ("1", "true", "yes")
)
# Off by default: never resurrect Italian text that acceptance filters cleared (reject-then-accept bug).
_ASR_IT_RESTORE_ON_FILTER = (
    os.environ.get("ASR_IT_RESTORE_ON_FILTER", "0").strip().lower() in ("1", "true", "yes")
)
# Match longer repeated tails across consecutive POST /audio responses (chunk boundary stitching).
_ASR_CROSS_CHUNK_OVERLAP_WORDS = max(3, int(os.environ.get("ASR_CROSS_CHUNK_OVERLAP_WORDS", "12")))
# One-switch preset for HoloLens-style HTTPS chunking + forced Italian (see ASR/HOLOLENS_TUNING.md).
_HOLOLENS_ITA_PROFILE = (
    os.environ.get("ASR_HOLOLENS_ITA_PROFILE", "0").strip().lower() in ("1", "true", "yes")
)

# Faster-Whisper: average logprob gate (more negative numbers = permissive).
# Omit from transcribe kwargs when ASR_FAST_LOGPROB_THRESHOLD=off; default -0.92 drops weak tails.
_lp_fw_raw = (os.environ.get("ASR_FAST_LOGPROB_THRESHOLD") or "").strip().lower()
if _lp_fw_raw in ("off", "disable", "none"):
    _FAST_LOG_PROB_THRESHOLD: float | None = None
elif _lp_fw_raw:
    _FAST_LOG_PROB_THRESHOLD = float(os.environ["ASR_FAST_LOGPROB_THRESHOLD"])
else:
    _FAST_LOG_PROB_THRESHOLD = -0.92

# Silence trimming before decode; can clip very quiet endings.
# Default demo posture: VAD off (can be enabled explicitly by env).
_ASR_VAD_FILTER_FAST_ENV = (os.environ.get("ASR_VAD_FILTER") or "false").strip().lower()
_ASR_VAD_FILTER_FAST = _ASR_VAD_FILTER_FAST_ENV in ("1", "true", "yes")
_ASR_VAD_FILTER_UNSET = not _ASR_VAD_FILTER_FAST_ENV
# Device clients often append long silence before POST; Whisper hallucinates on those tails.
_ASR_TRIM_TRAILING_SILENCE_IT = (
    os.environ.get("ASR_TRIM_TRAILING_SILENCE_IT", "1").strip().lower() in ("1", "true", "yes")
)
_ASR_TRIM_TRAILING_SILENCE_EN = (
    os.environ.get("ASR_TRIM_TRAILING_SILENCE_EN", "1").strip().lower() in ("1", "true", "yes")
)
_ASR_FORCE_SEGMENTATION_FORCED = (
    os.environ.get("ASR_FORCE_SEGMENTATION_FORCED", "0").strip().lower() in ("1", "true", "yes")
)

_WORD_RE = re.compile(r"[a-zA-ZÀ-ÿ']+")


def _vad_enabled_forced_italian() -> bool:
    # If env is unset, we default VAD ON for forced Italian only.
    if _ASR_VAD_FILTER_UNSET:
        return True
    return _ASR_VAD_FILTER_FAST

_EN_TAIL_FILLER_AFTER_QUESTION: frozenset[str] = frozenset(
    {"once", "yeah", "yep", "yup", "nah", "uh", "um", "umm", "hmm", "hm", "eh", "oh", "ah", "okay", "ok", "wow"}
)
_EN_MARKERS = {
    "the", "and", "you", "hello", "thanks", "thank", "world", "this", "that", "what", "where",
    "please", "good", "morning", "night", "today", "tomorrow", "yes", "no", "is", "are", "was",
    "were", "have", "has", "not", "can", "will", "would", "should", "could", "do", "does", "did",
    "how", "why", "when", "who", "which", "my", "your", "his", "her", "our", "their", "its",
    "with", "from", "for", "but", "very", "here", "there", "now", "then", "also", "about", "like",
    "know", "think", "want", "need", "come", "go", "get", "make", "take", "see", "say", "tell",
}
_IT_MARKERS = {
    "ciao", "grazie", "prego", "buongiorno", "buonasera", "notte", "come", "stai", "sono", "noi",
    "voi", "loro", "oggi", "domani", "per", "con", "questo", "quello", "dove", "quando", "perché",
    "perche", "si", "sì", "non", "che", "una", "uno", "il", "la", "gli", "le", "di", "del", "della",
    "delle", "dei", "degli", "in", "su", "ma", "anche", "più", "molto", "bene", "male", "scusi",
    "scusa", "mi", "ti", "ci", "vi", "lo", "li", "qui", "là", "ora", "poi", "cosa", "chi", "quale",
    "quanto", "tutto", "tutti", "ogni", "altro", "prima", "dopo", "dentro", "fuori", "bello",
}


def _parse_extra_hallucination_phrases_env() -> set[str]:
    raw = os.environ.get("ASR_HALLUCINATION_PHRASES_EXTRA") or ""
    out: set[str] = set()
    for part in re.split(r"[\n|]+", raw):
        p = part.strip().lower()
        if len(p) >= 4:
            out.add(p)
    return out


# Substring strip targets (long “video caption” hallucinations). Bare “thanks” / “thank you” are
# not removed here — those are handled only via weak-RMS word filtering to avoid eating valid speech.
_BASE_HALLUCINATION_SUBSTRING_PHRASES: frozenset[str] = frozenset(
    {
        "thank you for watching",
        "thanks for watching",
        "thank you for me",
        "thank you very much",
        "please subscribe",
        "subscribe to my channel",
        "like and subscribe",
        "see you next time",
        "see you in the next video",
        "don't forget to subscribe",
    }
)
_HALLUCINATION_SUBSTRING_PHRASES: set[str] = set(_BASE_HALLUCINATION_SUBSTRING_PHRASES) | _parse_extra_hallucination_phrases_env()


def _lang_marker_scores(text: str) -> tuple[float, float]:
    words = [w.lower() for w in _WORD_RE.findall(text or "")]
    if not words:
        return 0.0, 0.0
    uniq = set(words)
    denom = float(max(1, len(uniq)))
    en = len(uniq & _EN_MARKERS) / denom
    it = len(uniq & _IT_MARKERS) / denom
    return en, it


def _pick_best_text(it_text: str, en_text: str) -> str:
    it_text = (it_text or "").strip()
    en_text = (en_text or "").strip()
    if it_text and not en_text:
        return it_text
    if en_text and not it_text:
        return en_text
    if not it_text and not en_text:
        return ""

    en_score_it, it_score_it = _lang_marker_scores(it_text)
    en_score_en, it_score_en = _lang_marker_scores(en_text)

    # Prefer transcript that matches its model language profile more strongly.
    it_conf = it_score_it - en_score_it
    en_conf = en_score_en - it_score_en
    if en_conf > it_conf + 0.04:
        return en_text
    if it_conf > en_conf + 0.04:
        return it_text

    # Tie-breakers: avoid mixed punctuation junk by favoring shorter clean phrase.
    return en_text if len(en_text) <= len(it_text) else it_text


def _has_repetition_loop(text: str) -> bool:
    words = [w.lower() for w in _WORD_RE.findall(text or "")]
    if len(words) < 5:
        return False
    run = 1
    max_run = 1
    prev = words[0]
    for w in words[1:]:
        if w == prev:
            run += 1
            max_run = max(max_run, run)
        else:
            prev = w
            run = 1
    return max_run >= 3


def _has_repetitive_ngram_pattern(text: str) -> bool:
    words = [w.lower() for w in _WORD_RE.findall(text or "")]
    if len(words) < 8:
        return False
    for n, threshold in ((2, 0.34), (3, 0.26)):
        grams = [" ".join(words[i : i + n]) for i in range(0, len(words) - n + 1)]
        if not grams:
            continue
        counts: dict[str, int] = {}
        top = 0
        for g in grams:
            counts[g] = counts.get(g, 0) + 1
            if counts[g] > top:
                top = counts[g]
        if (top / max(1, len(grams))) >= threshold:
            return True
    return False


def _should_drop_repetition_hallucination(text: str) -> bool:
    t = (text or "").strip()
    if not t:
        return False
    words = [w.lower() for w in _WORD_RE.findall(t)]
    if len(words) >= 4 and _has_repetition_loop(t):
        return True
    if _has_repetitive_ngram_pattern(t):
        return True
    if len(words) > 40:
        return True
    return False


def _should_drop_whisper_low_energy_hallucination(text: str, rms: float, duration_s: float) -> bool:
    t = (text or "").strip()
    if not t:
        return False
    words = [w for w in _WORD_RE.findall(t)]
    letters_only = "".join(ch for ch in t if ch.isalpha())
    # Very low-energy long chunks producing rich text are often hallucinations.
    if rms < 0.014 and duration_s >= 2.2 and len(words) >= 5:
        return True
    if rms < 0.012 and duration_s >= 1.6 and len(words) >= 3:
        return True
    if rms < 0.008 and duration_s >= 3.0 and len(words) >= 8:
        return True
    if rms < 0.0075 and duration_s >= 3.0 and len(words) >= 7:
        return True
    if rms < 0.008 and len(words) >= 4:
        return True
    # Repetitive output from weak audio is usually unstable decoding.
    if rms < 0.009 and (_has_repetition_loop(t) or _has_repetitive_ngram_pattern(t)):
        return True
    # Tiny lexical variety under weak audio is another hallucination pattern.
    if rms < 0.008 and len(words) >= 8 and len(set(w.lower() for w in words)) <= 3:
        return True
    if rms < 0.008 and duration_s >= 3.5 and len(letters_only) >= 30 and len(words) <= 3:
        return True
    # Catch short "invented phrase" bursts from weak far-field speech.
    if rms < 0.009 and duration_s <= 2.0 and len(words) >= 7:
        return True
    return False


def _should_drop_forced_density_hallucination(text: str, rms: float, duration_s: float) -> bool:
    """Drop outputs that are implausibly dense for the chunk length/energy."""
    t = (text or "").strip()
    if not t:
        return False
    words = [w for w in _WORD_RE.findall(t)]
    if not words or duration_s <= 0.0:
        return False
    wps = len(words) / max(0.2, duration_s)
    # Very low energy + high lexical density is usually made-up text.
    if rms < 0.0075 and wps > 3.0:
        return True
    if rms < 0.0068 and wps > 2.2:
        return True
    # Extremely dense text for short chunks is unlikely.
    if duration_s < 2.0 and len(words) >= 14:
        return True
    return False


def _english_word_density_secondary_signals(
    raw_text: str,
    *,
    speech_rms: float,
    accept_thr: float,
    decode_conf: dict[str, float | None],
) -> list[str]:
    """Secondary bad signals paired with high words/sec (reject hallucinations, not fast speech)."""
    hints: list[str] = []
    nsp = decode_conf.get("no_speech_prob")
    if nsp is not None and float(nsp) > float(_WPS_EN_PAIR_NO_SPEECH):
        hints.append(f"no_speech_prob>{_WPS_EN_PAIR_NO_SPEECH}")
    alp = decode_conf.get("avg_logprob")
    alp_thr = float(_WPS_EN_PAIR_AVG_LOGPROB)
    if alp is not None and float(alp) < alp_thr:
        hints.append(f"avg_logprob<{alp_thr}")
    if speech_rms < accept_thr:
        hints.append("speech_rms_below_accept")
    cr = decode_conf.get("compression_ratio")
    if cr is not None and float(cr) > float(_COMPRESSION_RATIO_THRESHOLD) + 0.2:
        hints.append("high_compression_ratio")
    tl = (raw_text or "").lower()
    if any(s in tl for s in _WPS_CAPTION_HINT_SUBSTRINGS):
        hints.append("caption_like_substring")
    if _has_repetition_loop(raw_text) or _has_repetitive_ngram_pattern(raw_text):
        hints.append("repetition_pattern")
    return hints


def _decode_has_strong_confidence(decode_conf: dict[str, float | None]) -> bool:
    """When metadata exists and indicates confident decoding (avoid rejecting tails aggressively)."""
    nsp = decode_conf.get("no_speech_prob")
    alp = decode_conf.get("avg_logprob")
    if nsp is not None and float(nsp) < 0.35:
        return True
    if alp is not None and float(alp) > -0.55:
        return True
    return False


def _is_utterance_continuation(prev_upper: str, cur_upper: str) -> bool:
    """Overlap-aware continuation for chunked captions (same active utterance / phrase)."""
    pu = (prev_upper or "").strip().upper()
    cu = (cur_upper or "").strip().upper()
    if not pu or not cu:
        return False
    pw = pu.split()
    cw = cu.split()
    if not pw or not cw:
        return False
    max_k = min(len(pw), len(cw), 16)
    for k in range(max_k, 0, -1):
        if pw[-k:] == cw[:k]:
            return True
    pj = " ".join(pw)
    cj = " ".join(cw)
    return pj in cj or cj in pj


def _norm_tail_words(text: str) -> list[str]:
    return [w.upper().strip(".,!?;:\"'") for w in _WORD_RE.findall(text or "")]


def _is_generic_tail_closing_phrase(text: str, forced_language: str) -> bool:
    """Short isolated decoder tails (caller must enforce word_count <= 2 — not used on longer sentences)."""
    w = _norm_tail_words(text)
    if not w or len(w) > 2:
        return False
    if len(w) == 2 and w[0] == "THANK" and w[1] == "YOU":
        return True
    if len(w) == 1 and w[0] in ("THANKS", "THANKYOU"):
        return True
    if len(w) == 1 and w[0] == "YOU":
        return True
    if len(w) == 1 and w[0] == "GRAZIE":
        return True
    # Forced Italian can still decode short English tail hallucinations
    if forced_language == "italian" and (w == ["THANK", "YOU"] or w == ["THANKS"] or w == ["YOU"]):
        return True
    return False


def _intentional_short_tail_evidence(decode_conf: dict[str, float | None], speech_rms: float, duration_s: float) -> bool:
    """High bar: only allow short thanks-like tails when decode + energy look clearly intentional."""
    if not _decode_has_strong_confidence(decode_conf):
        return False
    if float(speech_rms) < 0.032:
        return False
    if float(duration_s) < 0.72:
        return False
    return True


def _strip_italian_conj_echo_suffix(text: str) -> str:
    """Strip Whisper tail hallucination ``... W e W`` (same word echoed after conjunction)."""
    t = (text or "").strip()
    if not t:
        return t
    while True:
        m = re.match(
            r"^(.*)\b([A-Za-zÀ-ÿ']+)\s+[Ee]\s+([A-Za-zÀ-ÿ']+)\s*$",
            t,
            flags=re.UNICODE,
        )
        if not m or m.group(2).lower() != m.group(3).lower():
            break
        t = (m.group(1).strip() + " " + m.group(2)).strip()
    return t


def _strip_repeated_terminal_bigram_suffix(text: str) -> str:
    """``... W1 W2 W1 W2`` at end → ``... W1 W2`` (duplicated closing phrase)."""
    t = (text or "").strip()
    while True:
        m = re.match(
            r"^(.*)\b([A-Za-zÀ-ÿ']+)\s+([A-Za-zÀ-ÿ']+)\s+([A-Za-zÀ-ÿ']+)\s+([A-Za-zÀ-ÿ']+)\s*$",
            t,
            flags=re.UNICODE,
        )
        if (
            not m
            or m.group(2).lower() != m.group(4).lower()
            or m.group(3).lower() != m.group(5).lower()
        ):
            break
        t = (m.group(1).strip() + " " + m.group(2) + " " + m.group(3)).strip()
    return t


def _strip_english_conj_echo_suffix(text: str) -> str:
    """``... W and W`` at line end → ``... W`` (same hallucination shape as Italian ``W e W``)."""
    t = (text or "").strip()
    if not t:
        return t
    while True:
        m = re.match(
            r"^(.*)\b([A-Za-zÀ-ÿ']+)\s+(?:and|&)\s+([A-Za-zÀ-ÿ']+)\s*$",
            t,
            flags=re.IGNORECASE | re.UNICODE,
        )
        if not m or m.group(2).lower() != m.group(3).lower():
            break
        t = (m.group(1).strip() + " " + m.group(2)).strip()
    return t


def _strip_english_terminal_filler_after_question(text: str) -> str:
    """Strip a lone filler token after ``?`` (often a silence tail hallucination, e.g. ``… ? Once``)."""
    t = (text or "").strip()
    while True:
        m = re.match(r"^(.+\?)(\s*)([A-Za-zÀ-ÿ'-]+)(?:[.!?…]|…)*\s*$", t, flags=re.UNICODE)
        if not m:
            break
        w = m.group(3).strip().strip(".,!?;:").lower()
        if w not in _EN_TAIL_FILLER_AFTER_QUESTION:
            break
        t = m.group(1).strip()
    return t


def _strip_dangling_english_tail(text: str) -> str:
    """No lexical stripping: rely on confidence/timing filters."""
    return (text or "").strip()


def _clean_text_artifacts(text: str) -> str:
    t = (text or "").strip()
    if not t:
        return ""
    # Remove stray parenthetical single-letter artifacts like "(E)" often produced by Whisper.
    t = re.sub(r"\(\s*[A-Za-z]\s*\)", "", t)
    # Remove repeated trailing connector patterns.
    t = re.sub(r"(?i)\b(and|e)\b(?:\s+\b(and|e)\b)+\s*$", "", t)
    # Collapse consecutive duplicate tokens (Whisper often doubles words on chunk boundaries).
    t = re.sub(r"\b([A-Za-zÀ-ÿ']+)(?:\s+\1\b)+", r"\1", t, flags=re.IGNORECASE)
    t = re.sub(r"\s{2,}", " ", t).strip(" ,;:-")
    return t


def _strip_dangling_italian_tail(text: str) -> str:
    """No lexical stripping: rely on confidence/timing filters."""
    return (text or "").strip()


def _strip_known_hallucinations(text: str, rms: float) -> str:
    t = (text or "").strip()
    if not t:
        return ""
    # Avoid aggressive deletion on strong-energy chunks.
    if rms > 0.06:
        return t
    out = t
    for p in sorted(_HALLUCINATION_SUBSTRING_PHRASES, key=len, reverse=True):
        out = re.sub(re.escape(p), " ", out, flags=re.IGNORECASE)
    out = re.sub(r"\s+", " ", out).strip()
    out = re.sub(r"\s{2,}", " ", out).strip(" ,;:-")
    return out


def _should_try_english_fallback(primary_text: str) -> bool:
    t = _clean_text_artifacts(primary_text)
    if not t:
        return True
    words = [w.lower() for w in _WORD_RE.findall(t)]
    if len(words) >= 3 and _has_repetition_loop(t):
        return True
    en_score, it_score = _lang_marker_scores(t)
    # If text looks more English than Italian, route to EN model.
    if en_score > it_score + 0.06:
        return True
    # If both scores are weak on non-trivial output, primary likely uncertain.
    if len(words) >= 4 and max(en_score, it_score) < 0.12:
        return True
    return False


def _should_drop_low_information_fragment(
    text: str,
    rms: float,
    duration_s: float,
    forced_language: str = "auto",
) -> bool:
    t = (text or "").strip()
    if not t:
        return False
    words = [w for w in _WORD_RE.findall(t)]
    letters_only = "".join(ch for ch in t if ch.isalpha())
    # Generic guardrail for silence hallucinations: tiny fragments from short/weak chunks.
    if len(words) <= 1 and len(letters_only) <= 1:
        return True
    if len(words) <= 1 and len(letters_only) <= 2 and (rms <= (_MIN_RMS * 1.35) or duration_s < 0.9):
        return True
    # Low-energy long chunks that decode to tiny snippets are usually hallucinations.
    # In forced Italian mode, keep this less aggressive to avoid cutting long-word continuations.
    if forced_language != "italian":
        if duration_s >= 2.2 and rms < 0.022 and len(letters_only) <= 6:
            return True
        if duration_s >= 2.0 and rms < 0.018 and len(words) <= 2:
            return True
    return False


def _should_drop_language_mismatch(text: str, forced_language: str) -> bool:
    t = (text or "").strip()
    if not t:
        return False
    if forced_language not in {"english", "italian"}:
        return False
    words = [w for w in _WORD_RE.findall(t)]
    if not words:
        return False
    en_score, it_score = _lang_marker_scores(t)
    if forced_language == "italian":
        # Italian lock: reject even short chunks if EN markers dominate.
        if len(words) <= 2:
            return en_score > it_score and en_score >= 0.20
        return en_score >= max(0.16, it_score + 0.04)
    # English lock symmetry.
    if len(words) <= 2:
        return it_score > en_score and it_score >= 0.20
    return it_score >= max(0.16, en_score + 0.04)


def _sanitize_forced_italian_output(text: str) -> str:
    t = (text or "").strip()
    if not t:
        return ""
    words = [w for w in t.split() if w]
    if not words:
        return ""
    cleaned: list[str] = []
    for w in words:
        n = _normalize_token(w)
        # Remove clear English marker tokens when in forced Italian mode.
        if n and (n in _EN_MARKERS) and (n not in _IT_MARKERS):
            continue
        cleaned.append(w)
    out = " ".join(cleaned).strip()
    if not out:
        return ""
    out = _strip_italian_conj_echo_suffix(out)
    if not out:
        return ""
    out = _strip_repeated_terminal_bigram_suffix(out)
    if not out:
        return ""
    en_score, it_score = _lang_marker_scores(out)
    # If still clearly English-leaning on non-trivial text, suppress output.
    if len(_WORD_RE.findall(out)) >= 3 and en_score > it_score + 0.08:
        return ""
    return out


def _sanitize_forced_english_output(text: str) -> str:
    t = (text or "").strip()
    if not t:
        return ""
    words = [w for w in t.split() if w]
    if not words:
        return ""
    cleaned: list[str] = []
    for w in words:
        n = _normalize_token(w)
        # Remove clear Italian marker tokens when in forced English mode.
        if n and (n in _IT_MARKERS) and (n not in _EN_MARKERS):
            continue
        cleaned.append(w)
    out = " ".join(cleaned).strip()
    if not out:
        return ""
    out = _strip_english_conj_echo_suffix(out)
    out = _strip_repeated_terminal_bigram_suffix(out)
    out = _strip_english_terminal_filler_after_question(out)
    out = _strip_dangling_english_tail(out)
    if not out:
        return ""
    en_score, it_score = _lang_marker_scores(out)
    # If still clearly Italian-leaning on non-trivial text, suppress output.
    if len(_WORD_RE.findall(out)) >= 3 and it_score > en_score + 0.08:
        return ""
    return out


def _hf_token() -> str | None:
    """Space secret / env: use **Secret** `HF_TOKEN` (or `HUGGING_FACE_HUB_TOKEN`) with Read on the model repo."""
    for key in (
        "HF_TOKEN",
        "HUGGING_FACE_HUB_TOKEN",
        "HUGGINGFACE_TOKEN",
        "HF_ACCESS_TOKEN",
    ):
        raw = os.environ.get(key)
        if raw:
            t = raw.strip().strip('"').strip("'")
            if t:
                return t


def _english_decode_suspect(text: str) -> bool:
    t = (text or "").strip()
    if not t:
        return True
    words = [w for w in _WORD_RE.findall(t)]
    if len(words) >= 3 and _should_drop_repetition_hallucination(t):
        return True
    if _should_drop_language_mismatch(t, "english"):
        return True
    return False


def _english_candidate_score(text: str) -> float:
    """Higher is better for selecting between two English decodes."""
    t = (text or "").strip()
    if not t:
        return -100.0
    words = [w.lower() for w in _WORD_RE.findall(t)]
    if not words:
        return -100.0
    uniq = len(set(words))
    score = 0.0
    score += min(1.0, uniq / max(1.0, len(words)))
    if _has_repetition_loop(t):
        score -= 1.0
    if _has_repetitive_ngram_pattern(t):
        score -= 0.8
    if _should_drop_language_mismatch(t, "english"):
        score -= 0.7
    return score


def _device() -> torch.device:
    if torch.cuda.is_available():
        return torch.device("cuda")
    return torch.device("cpu")


DEVICE = _device()
if _CPU_RELIABLE_MODE_RAW:
    _CPU_RELIABLE_MODE = _CPU_RELIABLE_MODE_RAW in ("1", "true", "yes")
else:
    # CPU fallback should prioritize completeness over dropping chunks.
    _CPU_RELIABLE_MODE = DEVICE.type == "cpu"
_CPU_BALANCED_MODE = (os.environ.get("ASR_CPU_BALANCED_MODE", "1").strip().lower() in ("1", "true", "yes"))
if _ASR_ITALIAN_BACKEND_RAW:
    _ASR_ITALIAN_BACKEND = _ASR_ITALIAN_BACKEND_RAW
else:
    # Keep Italian strict/clean by default on GPU; keep shared fast path on CPU.
    _ASR_ITALIAN_BACKEND = "transformers" if DEVICE.type == "cuda" else "shared"

# Device-aware defaults so HF setup needs fewer env edits.
if not _FAST_COMPUTE_TYPE:
    _FAST_COMPUTE_TYPE = "float16" if DEVICE.type == "cuda" else "int8"

if _FAST_USE_REALTIME_FOR_FORCED_RAW:
    _FAST_USE_REALTIME_FOR_FORCED = _FAST_USE_REALTIME_FOR_FORCED_RAW in ("1", "true", "yes")
else:
    # Quality-first on GPU; speed-first on CPU.
    _FAST_USE_REALTIME_FOR_FORCED = DEVICE.type != "cuda"

if _ASR_DROP_WHEN_BUSY_RAW:
    _ASR_DROP_WHEN_BUSY = _ASR_DROP_WHEN_BUSY_RAW in ("1", "true", "yes")
else:
    # On GPU avoid dropping words; on CPU keep realtime bounded.
    _ASR_DROP_WHEN_BUSY = DEVICE.type != "cuda"

if DEVICE.type == "cpu" and _CPU_RELIABLE_MODE:
    # CPU reliability mode: avoid dropping chunk content due to queue/energy heuristics.
    _ASR_DROP_WHEN_BUSY = False
    _ASR_RMS_SKIP_FORCED = 0.0
    _ASR_RMS_SKIP_AUTO = 0.0

# Keep forced-Italian speech-window RMS thresholds as configured on CPU too.

# CPU reliable: only relax server near-duplicate blanking (false empty text); keep all
# anti-hallucination paths enabled.
_CPU_RELAX_NEAR_DUP = DEVICE.type == "cpu" and _CPU_RELIABLE_MODE

if DEVICE.type == "cpu" and _CPU_BALANCED_MODE:
    # CPU balanced profile: faster decode defaults unless user explicitly set env values.
    if os.environ.get("ASR_WHISPER_NUM_BEAMS") is None:
        _WHISPER_BEAMS = 1
    if os.environ.get("ASR_WHISPER_MAX_NEW_TOKENS") is None:
        _WHISPER_MAX_NEW = 160

if _ASR_FORCED_NUM_BEAMS_RAW:
    _ASR_FORCED_NUM_BEAMS = max(1, int(_ASR_FORCED_NUM_BEAMS_RAW))
else:
    # Default forced decode beams to 1 (less tail wandering on streaming chunks).
    _ASR_FORCED_NUM_BEAMS = 1

if _ASR_FORCED_MAX_NEW_RAW:
    _ASR_FORCED_MAX_NEW = max(32, int(_ASR_FORCED_MAX_NEW_RAW))
else:
    _ASR_FORCED_MAX_NEW = 192 if DEVICE.type == "cuda" else _WHISPER_MAX_NEW

# Hard safety profile on GPU to avoid stale env overrides causing cutoffs/hallucination.
if DEVICE.type == "cuda" and _GPU_SAFE_PROFILE:
    _FAST_COMPUTE_TYPE = "float16"
    _FAST_USE_REALTIME_FOR_FORCED = False
    _ASR_DROP_WHEN_BUSY = False
    # Keep a small RMS guardrail on GPU to reduce silence hallucinations.
    _ASR_RMS_SKIP_FORCED = max(_ASR_RMS_SKIP_FORCED, 0.0035)
    _ASR_RMS_SKIP_AUTO = max(_ASR_RMS_SKIP_AUTO, 0.0045)


def _apply_hololens_ita_profile() -> None:
    """Tune forced-Italian + chunked HTTPS for HoloLens-class clients.

    Applies only when ``ASR_HOLOLENS_ITA_PROFILE=1``. Each knob is overridden **only** if its
    dedicated env key is **missing** (unset), so operators can still override any single value.
    """
    global _NO_SPEECH_THRESHOLD_IT, _ASR_MAX_CHUNK_S_FORCED, _ASR_CHUNK_OVERLAP_S
    global _ASR_FORCED_NUM_BEAMS, _ASR_FORCED_MAX_NEW
    global _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN, _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT
    global _ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN, _ASR_SHORT_MIN_CHUNK_S_FORCED_ITALIAN
    global _ASR_SHORT_MAX_CHUNK_S_FORCED_ITALIAN
    global _ASR_DROP_MICRO_FRAGMENTS, _ASR_BLANK_NEAR_DUP_FORCED
    if not _HOLOLENS_ITA_PROFILE:
        return
    # Use stricter no-speech for forced Italian to suppress weak tail decodes.
    if os.environ.get("ASR_NO_SPEECH_THRESHOLD_IT") is None:
        _NO_SPEECH_THRESHOLD_IT = 0.84
    # Align server-side stitch window with shorter forced chunks.
    if os.environ.get("ASR_MAX_CHUNK_SECONDS_FORCED") is None:
        _ASR_MAX_CHUNK_S_FORCED = 4.0
    if os.environ.get("ASR_CHUNK_OVERLAP_SECONDS") is None:
        _ASR_CHUNK_OVERLAP_S = 0.2
    if os.environ.get("ASR_MIN_CHUNK_SECONDS") is None:
        _ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN = 1.2
    if os.environ.get("ASR_SHORT_MIN_CHUNK_SECONDS_FORCED_ITALIAN") is None:
        _ASR_SHORT_MIN_CHUNK_S_FORCED_ITALIAN = 1.4
    if os.environ.get("ASR_FORCED_NUM_BEAMS") is None and DEVICE.type == "cuda":
        _ASR_FORCED_NUM_BEAMS = 1
    if os.environ.get("ASR_FORCED_MAX_NEW_TOKENS") is None and DEVICE.type == "cuda":
        _ASR_FORCED_MAX_NEW = 192
    # Keep speech-window RMS thresholds unless operator overrides (aligned with HoloLens fragments).
    if os.environ.get("ASR_MIN_SPEECH_RMS_FORCED_ITALIAN") is None:
        _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN = 0.018
    if os.environ.get("ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT") is None:
        _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT = 0.016
    if os.environ.get("ASR_SHORT_MAX_CHUNK_SECONDS_FORCED_ITALIAN") is None:
        _ASR_SHORT_MAX_CHUNK_S_FORCED_ITALIAN = 2.4
    if os.environ.get("ASR_DROP_MICRO_FRAGMENTS") is None:
        _ASR_DROP_MICRO_FRAGMENTS = False
    if os.environ.get("ASR_BLANK_NEAR_DUP_FORCED") is None:
        _ASR_BLANK_NEAR_DUP_FORCED = False


_apply_hololens_ita_profile()

# Startup defaults log (so you can verify live settings without guessing).
try:
    vad_it_effective = True if _ASR_VAD_FILTER_UNSET else _ASR_VAD_FILTER_FAST
    if _ASR_ITALIAN_BACKEND == "transformers":
        active_forced_it_model = MODEL_ID_IT
    elif _ASR_BACKEND in {"faster-whisper", "fast", "ctranslate2"}:
        active_forced_it_model = _FAST_REALTIME_MODEL_IT if _FAST_USE_REALTIME_FOR_FORCED else _FAST_MODEL_IT
    else:
        active_forced_it_model = MODEL_ID_PRIMARY
    print(
        f"[startup] backend={_ASR_BACKEND} italian_backend={_ASR_ITALIAN_BACKEND} "
        f"forced_it_model={active_forced_it_model} primary_model={MODEL_ID_PRIMARY} beams_forced={_ASR_FORCED_NUM_BEAMS} beams_whisper={_WHISPER_BEAMS} "
        f"it_short_retry={_ASR_IT_SHORT_RETRY} fallback_on_empty={_FALLBACK_ON_EMPTY} "
        f"vad_env_unset={_ASR_VAD_FILTER_UNSET} vad_it_effective={vad_it_effective} "
        f"min_chunk_seconds_forced_it={_ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN:.2f}"
    )
except Exception as e:
    print(f"[startup] log failed ({e})")

# Optional: set `ASR_CPU_THREADS` (e.g. 4) on CPU-only Spaces to cap thread contention.
if DEVICE.type == "cpu":
    try:
        torch.set_num_threads(_CPU_THREADS)
    except ValueError:
        pass
    try:
        torch.set_num_interop_threads(_CPU_INTEROP_THREADS)
    except RuntimeError:
        # Can be set only once per process; safe to continue.
        pass


def _zlib_compression_ratio_from_token_ids(token_ids: list[int], vocab_size: int) -> float:
    """Whisper-style zlib compression ratio over raw token bytes (HF generation_whisper)."""
    if not token_ids:
        return 1.0
    vs = max(2, int(vocab_size))
    length = int(math.log2(vs) / 8) + 1
    token_bytes = b"".join([int(t).to_bytes(length, byteorder="little", signed=False) for t in token_ids])
    compressed = zlib.compress(token_bytes)
    if not compressed:
        return 1.0
    return float(len(token_bytes) / len(compressed))


def _avg_logprob_from_generate_scores(
    scores: tuple[torch.Tensor, ...] | None,
    seq_tokens: torch.Tensor,
    *,
    temperature: float = 1.0,
) -> float | None:
    """Mirror HF Whisper `_retrieve_avg_logprobs` for greedy/beam batch 0."""
    if not scores or seq_tokens.numel() == 0:
        return None
    stacked = torch.stack([s[0] for s in scores]).float()
    tokens = seq_tokens.long()
    if stacked.shape[0] > tokens.shape[0]:
        stacked = stacked[: tokens.shape[0]]
    else:
        tokens = tokens[-stacked.shape[0] :]
    if stacked.shape[0] == 0:
        return None
    rescale = temperature if temperature > 0 else 1.0
    logprobs = F.log_softmax(stacked * rescale, dim=-1).to(stacked.dtype)
    sum_lp = sum(logprobs[i][tokens[i]] for i in range(logprobs.shape[0]))
    return float(sum_lp / logprobs.shape[0])


def _whisper_transformers_segment_meta(
    model: WhisperForConditionalGeneration, gen_out: object
) -> dict[str, float | None]:
    """Decode-side confidence from Transformers Whisper `generate` when `return_dict_in_generate=True`."""
    meta: dict[str, float | None] = {"no_speech_prob": None, "avg_logprob": None, "compression_ratio": None}
    if gen_out is None or isinstance(gen_out, torch.Tensor):
        return meta
    sequences = getattr(gen_out, "sequences", None)
    if sequences is None and isinstance(gen_out, dict):
        sequences = gen_out.get("sequences")
    if sequences is None:
        return meta
    try:
        if sequences.numel() == 0:
            return meta
    except AttributeError:
        return meta
    seq_row = sequences[0]
    vocab_size = int(getattr(model.config, "vocab_size", 51866))
    ids_list = [int(x) for x in seq_row.tolist()]
    pad_id = getattr(model.generation_config, "pad_token_id", None)
    eos_id = getattr(model.generation_config, "eos_token_id", None)
    if pad_id is not None:
        while ids_list and ids_list[-1] == pad_id:
            ids_list.pop()
    try:
        meta["compression_ratio"] = float(_zlib_compression_ratio_from_token_ids(ids_list, vocab_size))
    except Exception:
        meta["compression_ratio"] = None

    scores = getattr(gen_out, "scores", None)
    if scores is None and isinstance(gen_out, dict):
        scores = gen_out.get("scores")
    avg_lp: float | None = None
    if scores:
        try:
            n_gen = len(scores)
            tail = seq_row[-n_gen:] if seq_row.shape[-1] >= n_gen else seq_row
            avg_lp = _avg_logprob_from_generate_scores(scores, tail, temperature=1.0)
        except Exception:
            avg_lp = None
    if avg_lp is None:
        seq_scores = getattr(gen_out, "sequences_scores", None)
        if seq_scores is None and isinstance(gen_out, dict):
            seq_scores = gen_out.get("sequences_scores")
        if seq_scores is not None:
            try:
                total_lp = float(seq_scores[0].item())
                denom = len(scores) if scores else max(1, len(ids_list))
                avg_lp = total_lp / float(denom)
            except Exception:
                avg_lp = None
    meta["avg_logprob"] = avg_lp

    nsp = getattr(gen_out, "no_speech_prob", None)
    if nsp is None and isinstance(gen_out, dict):
        nsp = gen_out.get("no_speech_prob")
    if nsp is not None:
        try:
            meta["no_speech_prob"] = float(nsp[0]) if hasattr(nsp, "__getitem__") else float(nsp)
        except Exception:
            meta["no_speech_prob"] = None
    return meta


class WhisperASRSingle:
    def __init__(self, primary_model_name: str, fallback_model_name: str):
        self.primary_model_name = primary_model_name
        self.fallback_model_name = fallback_model_name
        self._primary_model: Optional[WhisperForConditionalGeneration] = None
        self._primary_processor: Optional[WhisperProcessor] = None
        self._fallback_model: Optional[WhisperForConditionalGeneration] = None
        self._fallback_processor: Optional[WhisperProcessor] = None

    def load(self) -> None:
        if self._primary_model is not None:
            return
        # Do not call huggingface_hub.login() here: it hits whoami and can 429 on Spaces.
        # Passing token= into from_pretrained is enough for private/gated models.
        token = _hf_token()
        kwargs: dict = {}
        if token:
            kwargs["token"] = token
            
        print(f"[load] Loading primary model from {self.primary_model_name}")
        self._primary_processor = WhisperProcessor.from_pretrained(self.primary_model_name, **kwargs)
        self._primary_model = WhisperForConditionalGeneration.from_pretrained(
            self.primary_model_name, **kwargs
        ).to(DEVICE)
        self._primary_model.eval()
        self._primary_model.generation_config.forced_decoder_ids = None

        if DEVICE.type == "cuda":
            torch.backends.cudnn.benchmark = True

    def _ensure_fallback_loaded(self) -> bool:
        if self._fallback_model is not None and self._fallback_processor is not None:
            return True
        if not self.fallback_model_name or self.fallback_model_name == self.primary_model_name:
            return False
        token = _hf_token()
        kwargs: dict = {}
        if token:
            kwargs["token"] = token
        print(f"[load] Loading fallback model from {self.fallback_model_name}")
        try:
            self._fallback_processor = WhisperProcessor.from_pretrained(self.fallback_model_name, **kwargs)
        except OSError as e:
            print(
                f"[WARN] Could not load fallback processor from {self.fallback_model_name} ({e}). "
                "Using primary processor."
            )
            self._fallback_processor = self._primary_processor
        try:
            self._fallback_model = WhisperForConditionalGeneration.from_pretrained(
                self.fallback_model_name, **kwargs
            ).to(DEVICE)
            self._fallback_model.eval()
            self._fallback_model.generation_config.forced_decoder_ids = None
            return True
        except Exception as e:
            print(f"[WARN] Could not load fallback model ({e})")
            self._fallback_model = None
            return False

    def _generate_ids(self, model, input_features: torch.Tensor, gen_common: dict):
        """Single generate path; fp16 autocast on CUDA when enabled."""
        if DEVICE.type == "cuda" and _USE_FP16_CUDA:
            with torch.autocast(device_type="cuda", dtype=torch.float16):
                return model.generate(input_features, **gen_common)
        return model.generate(input_features, **gen_common)

    @torch.no_grad()
    def transcribe(
        self,
        waveform: np.ndarray,
        sample_rate: int,
        *,
        forced_language: str | None = None,
        max_new_tokens: int | None = None,
        num_beams: int | None = None,
        no_speech_threshold: float | None = None,
        compression_ratio_threshold: float | None = None,
        logprob_threshold: float | None = None,
    ) -> tuple[str, dict[str, float | None]]:
        meta_empty: dict[str, float | None] = {"no_speech_prob": None, "avg_logprob": None, "compression_ratio": None}
        if max_new_tokens is None:
            max_new_tokens = _WHISPER_MAX_NEW
        if num_beams is None:
            num_beams = _WHISPER_BEAMS
        w = np.asarray(waveform, dtype=np.float32).reshape(-1)
        if w.size == 0 or not math.isfinite(float(np.max(np.abs(w)))):
            return "", meta_empty
        sr = int(sample_rate)
        t = torch.from_numpy(w).unsqueeze(0)
        if sr != SAMPLE_RATE:
            t = torchaudio.functional.resample(t, sr, SAMPLE_RATE)
        audio_16k = t.squeeze(0).numpy()

        self.load()
        assert self._primary_processor is not None and self._primary_model is not None

        try:
            inputs = self._primary_processor(
                audio_16k,
                sampling_rate=SAMPLE_RATE,
                return_tensors="pt",
                return_attention_mask=True,
            )
        except TypeError:
            inputs = self._primary_processor(audio_16k, sampling_rate=SAMPLE_RATE, return_tensors="pt")

        input_features = inputs["input_features"].to(DEVICE)
        attention_mask = inputs.get("attention_mask")
        if attention_mask is not None:
            attention_mask = attention_mask.to(DEVICE)

        lim = int(getattr(self._primary_model.config, "max_target_positions", 448))
        cap = max(1, min(max_new_tokens, lim - 24))

        requested = (forced_language or "").strip().lower()
        if requested not in {"english", "italian"}:
            requested = "auto"

        def _decode_with_meta(model: WhisperForConditionalGeneration, processor: WhisperProcessor, language: str):
            gen_common: dict = {
                "max_new_tokens": cap,
                "num_beams": num_beams,
                "do_sample": False,
                "task": "transcribe",
                "language": language,
                "forced_decoder_ids": None,
                "condition_on_prev_tokens": False,
            }
            if attention_mask is not None:
                gen_common["attention_mask"] = attention_mask
            if no_speech_threshold is not None:
                gen_common["no_speech_threshold"] = no_speech_threshold
            if compression_ratio_threshold is not None:
                gen_common["compression_ratio_threshold"] = compression_ratio_threshold
            if logprob_threshold is not None and no_speech_threshold is not None:
                gen_common["logprob_threshold"] = logprob_threshold

            # Whisper may ignore kwargs unless applied on GenerationConfig; without this,
            # output_scores is dropped and avg_logprob stays None (see transformers warning).
            try:
                gen_cfg = copy.deepcopy(model.generation_config)
                gen_cfg.return_dict_in_generate = True
                gen_cfg.output_scores = True
            except Exception:
                gen_cfg = None

            try:
                if gen_cfg is not None:
                    out = self._generate_ids(model, input_features, {**gen_common, "generation_config": gen_cfg})
                else:
                    out = self._generate_ids(
                        model,
                        input_features,
                        {**gen_common, "return_dict_in_generate": True, "output_scores": True},
                    )
            except TypeError:
                try:
                    out = self._generate_ids(model, input_features, dict(gen_common))
                except TypeError:
                    gen_common.pop("condition_on_prev_tokens", None)
                    gen_common.pop("no_speech_threshold", None)
                    gen_common.pop("compression_ratio_threshold", None)
                    gen_common.pop("logprob_threshold", None)
                    gen_common.pop("forced_decoder_ids", None)
                    out = self._generate_ids(model, input_features, dict(gen_common))

            if isinstance(out, torch.Tensor):
                ids_local = out
                meta = dict(meta_empty)
            else:
                ids_local = getattr(out, "sequences", None)
                if ids_local is None and isinstance(out, dict):
                    ids_local = out.get("sequences")
                if ids_local is None:
                    ids_local = out
                meta = _whisper_transformers_segment_meta(model, out)

            text = _clean_text_artifacts(processor.batch_decode(ids_local, skip_special_tokens=True)[0] or "")
            return text, meta

        # Explicit single-language decode for speed.
        if requested == "italian":
            return _decode_with_meta(self._primary_model, self._primary_processor, "italian")

        if requested == "english":
            if self._ensure_fallback_loaded() and self._fallback_model is not None and self._fallback_processor is not None:
                return _decode_with_meta(self._fallback_model, self._fallback_processor, "english")
            return _decode_with_meta(self._primary_model, self._primary_processor, "english")

        # Auto mode: Italian first, English only when needed.
        text_it, meta_it = _decode_with_meta(self._primary_model, self._primary_processor, "italian")
        should_fallback = _DUAL_DECODE or _should_try_english_fallback(text_it) or (not text_it and _FALLBACK_ON_EMPTY)
        if not should_fallback:
            return text_it, meta_it

        if not self._ensure_fallback_loaded() or self._fallback_model is None or self._fallback_processor is None:
            return text_it, meta_it
        text_en, meta_en = _decode_with_meta(self._fallback_model, self._fallback_processor, "english")
        chosen = _pick_best_text(text_it, text_en)
        en_st = (text_en or "").strip()
        if en_st and chosen == en_st:
            return chosen, meta_en
        return chosen, meta_it


class FasterWhisperASR:
    def __init__(self, model_it: str, model_en: str):
        self.model_it_id = model_it
        self.model_en_id = model_en
        self.model_it_rt_id = _FAST_REALTIME_MODEL_IT
        self.model_en_rt_id = _FAST_REALTIME_MODEL_EN
        self._model_it = None
        self._model_en = None
        self._model_it_rt = None
        self._model_en_rt = None
        self._lib_error: Optional[Exception] = None

    def _load_model(self, model_id: str):
        from faster_whisper import WhisperModel

        cpu_threads = _CPU_THREADS if DEVICE.type == "cpu" else 0
        return WhisperModel(
            model_id,
            device="cuda" if DEVICE.type == "cuda" else "cpu",
            compute_type=_FAST_COMPUTE_TYPE,
            cpu_threads=cpu_threads,
            num_workers=1,
        )

    def _ensure_it(self):
        if self._model_it is not None:
            return self._model_it
        self._model_it = self._load_model(self.model_it_id)
        return self._model_it

    def _ensure_en(self):
        if self._model_en is not None:
            return self._model_en
        if self.model_en_id == self.model_it_id and self._model_it is not None:
            self._model_en = self._model_it
            return self._model_en
        self._model_en = self._load_model(self.model_en_id)
        return self._model_en

    def _ensure_it_rt(self):
        if self._model_it_rt is not None:
            return self._model_it_rt
        if self.model_it_rt_id == self.model_it_id and self._model_it is not None:
            self._model_it_rt = self._model_it
            return self._model_it_rt
        self._model_it_rt = self._load_model(self.model_it_rt_id)
        return self._model_it_rt

    def _ensure_en_rt(self):
        if self._model_en_rt is not None:
            return self._model_en_rt
        if self.model_en_rt_id == self.model_en_id and self._model_en is not None:
            self._model_en_rt = self._model_en
            return self._model_en_rt
        self._model_en_rt = self._load_model(self.model_en_rt_id)
        return self._model_en_rt

    def load(self) -> None:
        # Keep interface parity with WhisperASRSingle for startup warmup.
        self._ensure_it()

    @torch.no_grad()
    def transcribe(
        self,
        waveform: np.ndarray,
        sample_rate: int,
        *,
        forced_language: str | None = None,
        max_new_tokens: int | None = None,
        num_beams: int | None = None,
        no_speech_threshold: float | None = None,
        compression_ratio_threshold: float | None = None,
        logprob_threshold: float | None = None,
    ) -> tuple[str, dict[str, float | None]]:
        del sample_rate
        if max_new_tokens is None:
            max_new_tokens = _WHISPER_MAX_NEW
        if num_beams is None:
            num_beams = _WHISPER_BEAMS
        requested = (forced_language or "").strip().lower()
        if requested not in {"english", "italian"}:
            requested = "auto"

        legacy_lp = float(logprob_threshold if logprob_threshold is not None else _LOGPROB_THRESHOLD)
        if legacy_lp <= -1.0 + 1e-9:
            eff_log_prob = _FAST_LOG_PROB_THRESHOLD
        else:
            eff_log_prob = legacy_lp

        if requested == "italian":
            vad_filter_effective = _vad_enabled_forced_italian()
        else:
            vad_filter_effective = _ASR_VAD_FILTER_FAST

        seg_kwargs = {
            "beam_size": num_beams,
            "task": "transcribe",
            "condition_on_previous_text": False,
            "without_timestamps": True,
            "no_speech_threshold": _NO_SPEECH_THRESHOLD if no_speech_threshold is None else no_speech_threshold,
            "compression_ratio_threshold": _COMPRESSION_RATIO_THRESHOLD if compression_ratio_threshold is None else compression_ratio_threshold,
            "max_new_tokens": max_new_tokens,
            "vad_filter": vad_filter_effective,
        }
        if eff_log_prob is not None:
            seg_kwargs["log_prob_threshold"] = eff_log_prob

        audio = np.asarray(waveform, dtype=np.float32).reshape(-1)
        if requested == "english":
            use_rt_english = _FAST_USE_REALTIME_FOR_FORCED or (DEVICE.type == "cuda" and _ASR_ENGLISH_FAST_ON_CUDA)
            model = self._ensure_en_rt() if use_rt_english else self._ensure_en()
            en_kwargs = dict(seg_kwargs)
            if no_speech_threshold is None:
                en_kwargs["no_speech_threshold"] = _NO_SPEECH_THRESHOLD_EN
            segs, _ = model.transcribe(audio, language="en", **en_kwargs)
            segs = list(segs)
            text = _clean_text_artifacts(" ".join((s.text or "").strip() for s in segs).strip())
            text = _sanitize_forced_english_output(text)
            conf = {
                "no_speech_prob": float(np.mean([float(getattr(s, "no_speech_prob", 0.0)) for s in segs])) if segs else None,
                "avg_logprob": float(np.mean([float(getattr(s, "avg_logprob", 0.0)) for s in segs])) if segs else None,
                "compression_ratio": float(np.max([float(getattr(s, "compression_ratio", 1.0)) for s in segs])) if segs else None,
            }
            # Avoid prompt-based retries; if decode looks unstable, rerun with stronger EN model.
            can_rerun_stronger = _ASR_ENGLISH_RERUN_ON_SUSPECT and use_rt_english and (
                self._model_en is None or self._model_en is not model
            )
            if can_rerun_stronger and _english_decode_suspect(text):
                retry_model = self._ensure_en()
                retry_kwargs = dict(seg_kwargs)
                if no_speech_threshold is None:
                    retry_kwargs["no_speech_threshold"] = _NO_SPEECH_THRESHOLD_EN
                retry_kwargs["beam_size"] = max(2, num_beams)
                segs2, _ = retry_model.transcribe(audio, language="en", **retry_kwargs)
                segs2 = list(segs2)
                text2 = _clean_text_artifacts(" ".join((s.text or "").strip() for s in segs2).strip())
                text2 = _sanitize_forced_english_output(text2)
                if _english_candidate_score(text2) >= _english_candidate_score(text):
                    conf2 = {
                        "no_speech_prob": float(np.mean([float(getattr(s, "no_speech_prob", 0.0)) for s in segs2])) if segs2 else None,
                        "avg_logprob": float(np.mean([float(getattr(s, "avg_logprob", 0.0)) for s in segs2])) if segs2 else None,
                        "compression_ratio": float(np.max([float(getattr(s, "compression_ratio", 1.0)) for s in segs2])) if segs2 else None,
                    }
                    return text2, conf2
            return text, conf
        elif requested == "italian":
            model = self._ensure_it_rt() if _FAST_USE_REALTIME_FOR_FORCED else self._ensure_it()
            it_kwargs = dict(seg_kwargs)
            if no_speech_threshold is None:
                it_kwargs["no_speech_threshold"] = _NO_SPEECH_THRESHOLD_IT
            it_kwargs["initial_prompt"] = "Trascrivi fedelmente in italiano. Non tradurre."
            segs, _ = model.transcribe(audio, language="it", **it_kwargs)
            segs = list(segs)
            text = _clean_text_artifacts(" ".join((s.text or "").strip() for s in segs).strip())
            text = _sanitize_forced_italian_output(text)
            conf = {
                "no_speech_prob": float(np.mean([float(getattr(s, "no_speech_prob", 0.0)) for s in segs])) if segs else None,
                "avg_logprob": float(np.mean([float(getattr(s, "avg_logprob", 0.0)) for s in segs])) if segs else None,
                "compression_ratio": float(np.max([float(getattr(s, "compression_ratio", 1.0)) for s in segs])) if segs else None,
            }
            if _should_drop_language_mismatch(text, "italian"):
                retry_kwargs = dict(seg_kwargs)
                if no_speech_threshold is None:
                    retry_kwargs["no_speech_threshold"] = _NO_SPEECH_THRESHOLD_IT
                retry_kwargs["beam_size"] = max(3, num_beams)
                retry_kwargs["initial_prompt"] = "Trascrivi fedelmente in italiano. Non tradurre."
                segs2, _ = model.transcribe(audio, language="it", **retry_kwargs)
                segs2 = list(segs2)
                text2 = _clean_text_artifacts(" ".join((s.text or "").strip() for s in segs2).strip())
                text2 = _sanitize_forced_italian_output(text2)
                # Avoid blanking the whole chunk: prefer sanitized retry output.
                conf2 = {
                    "no_speech_prob": float(np.mean([float(getattr(s, "no_speech_prob", 0.0)) for s in segs2])) if segs2 else None,
                    "avg_logprob": float(np.mean([float(getattr(s, "avg_logprob", 0.0)) for s in segs2])) if segs2 else None,
                    "compression_ratio": float(np.max([float(getattr(s, "compression_ratio", 1.0)) for s in segs2])) if segs2 else None,
                }
                return text2, conf2
            return text, conf
        else:
            if DEVICE.type == "cpu" and _CPU_RELIABLE_MODE and _ASR_CPU_AUTO_USE_REALTIME:
                model = self._ensure_it_rt()
            else:
                model = self._ensure_it()
            segs, _ = model.transcribe(audio, language=None, **seg_kwargs)
            segs = list(segs)

        text = " ".join((s.text or "").strip() for s in segs).strip()
        conf = {
            "no_speech_prob": float(np.mean([float(getattr(s, "no_speech_prob", 0.0)) for s in segs])) if segs else None,
            "avg_logprob": float(np.mean([float(getattr(s, "avg_logprob", 0.0)) for s in segs])) if segs else None,
            "compression_ratio": float(np.max([float(getattr(s, "compression_ratio", 1.0)) for s in segs])) if segs else None,
        }
        return _clean_text_artifacts(text), conf


_pipe: Optional[object] = None
_pipe_it_transformers: Optional[WhisperASRSingle] = None
_asr_busy = False
_asr_infer_lock = threading.Lock()
_last_text_by_lang: dict[str, str] = {"english": "", "italian": "", "auto": ""}
_last_text_lock = threading.Lock()
_last_signal_by_lang: dict[str, dict[str, float]] = {
    "english": {"speech_rms": 0.0, "ts": 0.0},
    "italian": {"speech_rms": 0.0, "ts": 0.0},
    "auto": {"speech_rms": 0.0, "ts": 0.0},
}
# Last non-empty accepted transcript per forced language (post-dedup), for ultra-short tail rejection.
_last_accepted_phrase_by_lang: dict[str, dict[str, float | str]] = {}


def _text_compression_ratio(text: str) -> float:
    t = (text or "").strip()
    if not t:
        return 1.0
    return len(t) / max(1, len(set(t)))


def _normalize_token(token: str) -> str:
    return re.sub(r"^[^\wÀ-ÿ']+|[^\wÀ-ÿ']+$", "", (token or "").lower())


def _remove_overlap_repetition(previous: str, current: str, max_overlap_words: int = 10) -> str:
    prev = (previous or "").strip()
    curr = (current or "").strip()
    if not curr:
        return ""
    if not prev:
        return curr

    prev_words = [w for w in prev.split() if w]
    curr_words = [w for w in curr.split() if w]
    if not prev_words or not curr_words:
        return curr

    prev_norm = [_normalize_token(w) for w in prev_words]
    curr_norm = [_normalize_token(w) for w in curr_words]
    max_k = min(max_overlap_words, len(prev_norm), len(curr_norm))

    best_k = 0
    for k in range(max_k, 0, -1):
        if prev_norm[-k:] == curr_norm[:k]:
            best_k = k
            break
    if best_k > 0:
        trimmed = " ".join(curr_words[best_k:]).strip()
        return trimmed
    return curr


def _is_near_duplicate_phrase(previous: str, current: str) -> bool:
    prev = " ".join(_normalize_token(w) for w in (previous or "").split() if _normalize_token(w))
    curr = " ".join(_normalize_token(w) for w in (current or "").split() if _normalize_token(w))
    if not prev or not curr:
        return False
    if prev == curr:
        return True
    pw = prev.split()
    cw = curr.split()
    # suppress near-identical short repeats emitted in adjacent chunks
    if len(cw) >= 4 and len(pw) >= 4:
        k = min(7, len(cw), len(pw))
        if pw[-k:] == cw[-k:] or pw[-k:] == cw[:k]:
            return True
    return False


def get_pipe():
    global _pipe
    if _pipe is None:
        use_fast = _ASR_BACKEND in {"faster-whisper", "fast", "ctranslate2"}
        if use_fast:
            try:
                _pipe = FasterWhisperASR(_FAST_MODEL_IT, _FAST_MODEL_EN)
                # Warm load IT model so first request isn't cold.
                _pipe._ensure_it()
                print(
                    f"[startup] backend=faster-whisper compute_type={_FAST_COMPUTE_TYPE} "
                    f"model_it={_FAST_MODEL_IT} model_en={_FAST_MODEL_EN} "
                    f"forced_rt_it={_FAST_REALTIME_MODEL_IT} forced_rt_en={_FAST_REALTIME_MODEL_EN} "
                    f"use_forced_rt={_FAST_USE_REALTIME_FOR_FORCED} italian_backend={_ASR_ITALIAN_BACKEND} "
                    f"no_speech_auto={_NO_SPEECH_THRESHOLD} en={_NO_SPEECH_THRESHOLD_EN} it={_NO_SPEECH_THRESHOLD_IT} "
                    f"compression_ratio={_COMPRESSION_RATIO_THRESHOLD} compression_ratio_it={_COMPRESSION_RATIO_THRESHOLD_IT} "
                    f"log_prob_fw={_FAST_LOG_PROB_THRESHOLD} "
                    f"vad_env_unset={_ASR_VAD_FILTER_UNSET} vad_filter_global={_ASR_VAD_FILTER_FAST} "
                    f"vad_it_effective={_vad_enabled_forced_italian()}"
                )
            except Exception as e:
                print(f"[WARN] faster-whisper init failed ({e}); falling back to transformers backend.")
                _pipe = WhisperASRSingle(MODEL_ID_PRIMARY, MODEL_ID_FALLBACK)
        else:
            _pipe = WhisperASRSingle(MODEL_ID_PRIMARY, MODEL_ID_FALLBACK)
    return _pipe


def get_italian_pipe_transformers() -> WhisperASRSingle:
    global _pipe_it_transformers
    if _pipe_it_transformers is None:
        # Force dedicated Italian checkpoint as primary for strict Italian mode.
        _pipe_it_transformers = WhisperASRSingle(MODEL_ID_IT, MODEL_ID_EN)
    return _pipe_it_transformers


def preprocess_chunk(audio_float32: list[float], sample_rate: int) -> tuple[np.ndarray, float, float] | None:
    arr = np.array(audio_float32, dtype=np.float32)
    if len(arr) < 800:
        return None
    if sample_rate != SAMPLE_RATE:
        t = torch.from_numpy(arr).float().unsqueeze(0).unsqueeze(0)
        t = torchaudio.functional.resample(t, sample_rate, SAMPLE_RATE)
        arr = t.squeeze().numpy()
    arr = (
        torchaudio.functional.highpass_biquad(
            torch.from_numpy(arr).float().unsqueeze(0), SAMPLE_RATE, 80.0
        )
        .squeeze(0)
        .numpy()
    )
    rms = float(np.sqrt(np.mean(arr**2))) if arr.size > 0 else 0.0
    if rms < _MIN_RMS:
        return None
    duration_s = float(arr.size) / float(SAMPLE_RATE)
    return arr, rms, duration_s


def _trim_trailing_silence_italian_chunk(audio_16k: np.ndarray) -> tuple[np.ndarray, float]:
    """Shorten audio by dropping quiet frames at the end (reduces silence-tail hallucinations).

    Returns ``(audio, seconds_removed)``. No-op when trim would remove too much payload or
    would drop below ``preprocess_chunk`` minimum length.
    """
    arr = np.asarray(audio_16k, dtype=np.float32).reshape(-1)
    min_keep = 800
    if arr.size < int(0.35 * SAMPLE_RATE):
        return arr, 0.0
    frame = max(1, int(0.025 * SAMPLE_RATE))
    hop = max(1, int(0.010 * SAMPLE_RATE))
    rms_vals: list[float] = []
    frame_end_idx: list[int] = []
    for i in range(0, arr.size - frame + 1, hop):
        w = arr[i : i + frame]
        rms_vals.append(float(np.sqrt(np.mean(w**2))))
        frame_end_idx.append(i + frame)
    if not rms_vals:
        return arr, 0.0
    peak = max(rms_vals)
    thr = max(0.007, min(0.04, 0.08 * peak))
    last_voiced: int | None = None
    for idx in range(len(rms_vals) - 1, -1, -1):
        if rms_vals[idx] >= thr:
            last_voiced = idx
            break
    if last_voiced is None:
        return arr, 0.0
    pad = int(0.10 * SAMPLE_RATE)
    end = min(arr.size, frame_end_idx[last_voiced] + pad)
    if end >= arr.size - hop:
        return arr, 0.0
    # Avoid nuking real content if threshold mis-fires.
    if end < min_keep:
        return arr, 0.0
    removed = arr.size - end
    if removed / max(1, arr.size) > 0.72:
        return arr, 0.0
    return arr[:end].copy(), float(removed / SAMPLE_RATE)


def _forced_italian_vad_has_speech(audio_16k: np.ndarray) -> bool:
    """Conservative VAD pre-check for forced Italian path."""
    if audio_16k is None or audio_16k.size < int(0.08 * SAMPLE_RATE):
        return False
    try:
        t = torch.from_numpy(audio_16k.astype(np.float32, copy=False)).unsqueeze(0)
        voiced = torchaudio.functional.vad(t, SAMPLE_RATE)
        if voiced.numel() == 0:
            return False
        voiced_np = voiced.squeeze(0).numpy()
        if voiced_np.size < int(0.10 * SAMPLE_RATE):
            return False
        voiced_rms = float(np.sqrt(np.mean(voiced_np**2))) if voiced_np.size > 0 else 0.0
        return voiced_rms >= max(0.004, _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN * 0.35)
    except Exception as e:
        # Fail-open on VAD runtime issues; do not drop valid speech due to VAD errors.
        print(f"[warn] vad_precheck_failed lang=italian err={e}")
        return True


def _forced_italian_speech_rms(audio_16k: np.ndarray) -> tuple[float, float]:
    """Estimate speech energy from active frames, not whole-chunk average."""
    if audio_16k is None or audio_16k.size == 0:
        return 0.0, 0.0
    frame = max(1, int(0.025 * SAMPLE_RATE))  # 25 ms
    hop = max(1, int(0.010 * SAMPLE_RATE))    # 10 ms
    if audio_16k.size < frame:
        r = float(np.sqrt(np.mean(audio_16k**2))) if audio_16k.size > 0 else 0.0
        return r, r
    rms_vals: list[float] = []
    for i in range(0, audio_16k.size - frame + 1, hop):
        w = audio_16k[i : i + frame]
        rms_vals.append(float(np.sqrt(np.mean(w**2))))
    if not rms_vals:
        r = float(np.sqrt(np.mean(audio_16k**2)))
        return r, r
    arr = np.array(rms_vals, dtype=np.float32)
    p90 = float(np.percentile(arr, 90))
    k = max(1, int(np.ceil(0.2 * arr.size)))
    top_mean = float(np.mean(np.sort(arr)[-k:]))
    speech_rms = max(p90, top_mean)
    return speech_rms, p90


def _italian_relaxed_micro_fragment(duration_s: float, speech_rms: float) -> bool:
    """Strong speech but clip shorter than SHORT_MIN (fragmented HoloLens chunks)."""
    return (
        duration_s >= 0.32
        and duration_s < _ASR_SHORT_MIN_CHUNK_S_FORCED_ITALIAN
        and speech_rms >= _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT
    )


def _italian_accept_rms_threshold(duration_s: float, speech_rms: float) -> float:
    """Accept-floor RMS: micro-fragments use SHORT threshold when speech energy already clears it."""
    if _italian_relaxed_micro_fragment(duration_s, speech_rms):
        return float(_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT)
    return float(_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_ACCEPT)


def _extract_strongest_speech_segment(audio_16k: np.ndarray) -> tuple[np.ndarray, float, float]:
    """Return strongest contiguous speech segment and [start_s, end_s] in raw audio."""
    arr = np.asarray(audio_16k, dtype=np.float32).reshape(-1)
    if arr.size < int(0.25 * SAMPLE_RATE):
        return arr, 0.0, float(arr.size / SAMPLE_RATE)
    frame = max(1, int(0.025 * SAMPLE_RATE))
    hop = max(1, int(0.010 * SAMPLE_RATE))
    if arr.size < frame:
        return arr, 0.0, float(arr.size / SAMPLE_RATE)

    rms_vals: list[float] = []
    starts: list[int] = []
    for i in range(0, arr.size - frame + 1, hop):
        w = arr[i : i + frame]
        rms_vals.append(float(np.sqrt(np.mean(w**2))))
        starts.append(i)
    if not rms_vals:
        return arr, 0.0, float(arr.size / SAMPLE_RATE)
    rms_arr = np.asarray(rms_vals, dtype=np.float32)
    peak = float(np.max(rms_arr))
    thr = max(0.008, min(0.05, 0.12 * peak))
    active = rms_arr >= thr
    if not np.any(active):
        return arr, 0.0, float(arr.size / SAMPLE_RATE)

    # Build contiguous runs with a tiny gap tolerance.
    gap_tol = 2
    runs: list[tuple[int, int]] = []
    i = 0
    n = len(active)
    while i < n:
        if not active[i]:
            i += 1
            continue
        s = i
        last_on = i
        i += 1
        gap = 0
        while i < n:
            if active[i]:
                last_on = i
                gap = 0
            else:
                gap += 1
                if gap > gap_tol:
                    break
            i += 1
        runs.append((s, last_on))

    if not runs:
        return arr, 0.0, float(arr.size / SAMPLE_RATE)
    best_run = runs[0]
    best_score = -1.0
    for s_idx, e_idx in runs:
        seg_r = rms_arr[s_idx : e_idx + 1]
        dur_frames = max(1, e_idx - s_idx + 1)
        score = float(np.mean(seg_r) * math.sqrt(float(dur_frames)))
        if score > best_score:
            best_score = score
            best_run = (s_idx, e_idx)

    pad = int(0.08 * SAMPLE_RATE)
    seg_start = max(0, starts[best_run[0]] - pad)
    seg_end = min(arr.size, starts[best_run[1]] + frame + pad)
    if seg_end <= seg_start + 1:
        return arr, 0.0, float(arr.size / SAMPLE_RATE)
    return arr[seg_start:seg_end].copy(), float(seg_start / SAMPLE_RATE), float(seg_end / SAMPLE_RATE)


def _trim_edge_silence_forced_chunk(audio_16k: np.ndarray) -> tuple[np.ndarray, float, float]:
    """Trim only obvious leading/trailing silence; never split internal speech."""
    arr = np.asarray(audio_16k, dtype=np.float32).reshape(-1)
    if arr.size < int(0.35 * SAMPLE_RATE):
        return arr, 0.0, 0.0
    frame = max(1, int(0.025 * SAMPLE_RATE))
    hop = max(1, int(0.010 * SAMPLE_RATE))
    if arr.size < frame:
        return arr, 0.0, 0.0
    rms_vals: list[float] = []
    starts: list[int] = []
    for i in range(0, arr.size - frame + 1, hop):
        w = arr[i : i + frame]
        rms_vals.append(float(np.sqrt(np.mean(w**2))))
        starts.append(i)
    if not rms_vals:
        return arr, 0.0, 0.0
    rms_arr = np.asarray(rms_vals, dtype=np.float32)
    peak = float(np.max(rms_arr))
    thr = max(0.008, min(0.03, 0.08 * peak))
    active = rms_arr >= thr
    if not np.any(active):
        return arr, 0.0, 0.0
    first_idx = int(np.argmax(active))
    last_idx = int(len(active) - 1 - np.argmax(active[::-1]))
    pad = int(0.08 * SAMPLE_RATE)
    start = max(0, starts[first_idx] - pad)
    end = min(arr.size, starts[last_idx] + frame + pad)
    if end <= start + 1:
        return arr, 0.0, 0.0
    if start == 0 and end == arr.size:
        return arr, 0.0, 0.0
    trimmed = arr[start:end].copy()
    return trimmed, float(start / SAMPLE_RATE), float((arr.size - end) / SAMPLE_RATE)


def _tail_low_energy_ratio(audio_16k: np.ndarray) -> float:
    """How much of the final third is low energy (1.0 = mostly weak tail)."""
    arr = np.asarray(audio_16k, dtype=np.float32).reshape(-1)
    if arr.size < int(0.35 * SAMPLE_RATE):
        return 0.0
    frame = max(1, int(0.025 * SAMPLE_RATE))
    hop = max(1, int(0.010 * SAMPLE_RATE))
    if arr.size < frame:
        return 0.0
    rms_vals: list[float] = []
    for i in range(0, arr.size - frame + 1, hop):
        w = arr[i : i + frame]
        rms_vals.append(float(np.sqrt(np.mean(w**2))))
    if len(rms_vals) < 5:
        return 0.0
    r = np.asarray(rms_vals, dtype=np.float32)
    peak = float(np.max(r))
    thr = max(0.008, min(0.03, 0.10 * peak))
    tail_n = max(3, int(math.ceil(0.33 * len(r))))
    tail = r[-tail_n:]
    return float(np.mean(tail <= thr))


def _split_sentences_simple(text: str) -> list[str]:
    t = (text or "").strip()
    if not t:
        return []
    parts = re.split(r"(?<=[.!?])\s+", t)
    return [p.strip() for p in parts if p and p.strip()]


def _trim_suspicious_tail_sentence(
    text: str,
    *,
    audio_16k: np.ndarray,
    duration_s: float,
    decode_conf: dict,
) -> tuple[str, bool]:
    """Trim likely weak late continuation sentence (model tail completion)."""
    sentences = _split_sentences_simple(text)
    if len(sentences) < 2:
        return text, False
    last = sentences[-1]
    main = " ".join(sentences[:-1]).strip()
    if not main:
        return text, False
    last_words = len(_WORD_RE.findall(last))
    if last_words > 4:
        return text, False
    tail_ratio = _tail_low_energy_ratio(audio_16k)
    if tail_ratio < 0.68:
        return text, False
    nsp = decode_conf.get("no_speech_prob")
    alp = decode_conf.get("avg_logprob")
    # Moderate confidence still frequently hallucinates short endings on weak tails.
    moderate_or_worse = (
        (nsp is None or float(nsp) >= 0.15)
        or (alp is None or float(alp) <= -0.25)
    )
    short_chunk = float(duration_s) <= 4.0
    if moderate_or_worse and short_chunk:
        return main, True
    return text, False


def _coerce_transcribe_result(out: tuple[str, dict] | str) -> tuple[str, dict[str, float | None]]:
    if isinstance(out, tuple) and len(out) >= 2 and isinstance(out[1], dict):
        return (out[0] or ""), out[1]
    return (out if isinstance(out, str) else str(out or "")), {}


def _transcribe_chunked(
    arr: np.ndarray,
    forced_language: str,
    max_new_tokens: int,
    num_beams: int,
    no_speech_threshold_override: float | None = None,
) -> tuple[str, dict]:
    """Transcribe long payloads in overlapping windows to reduce latency spikes and hallucinations."""
    max_chunk_s = _ASR_MAX_CHUNK_S_AUTO if forced_language == "auto" else _ASR_MAX_CHUNK_S_FORCED
    chunk_samples = max(1, int(SAMPLE_RATE * max_chunk_s))
    overlap_samples = max(0, int(SAMPLE_RATE * _ASR_CHUNK_OVERLAP_S))
    step = max(1, chunk_samples - overlap_samples)
    n = int(arr.size)
    if n <= chunk_samples:
        segments = [arr]
    else:
        segments = []
        start = 0
        while start < n:
            end = min(n, start + chunk_samples)
            segments.append(arr[start:end])
            if end >= n:
                break
            start += step

    if no_speech_threshold_override is not None:
        ns_thr = float(no_speech_threshold_override)
    elif forced_language == "italian":
        ns_thr = _NO_SPEECH_THRESHOLD_IT
    elif forced_language == "english":
        ns_thr = _NO_SPEECH_THRESHOLD_EN
    else:
        ns_thr = _NO_SPEECH_THRESHOLD
    out_parts: list[str] = []
    prev = ""
    conf_no_speech: list[float] = []
    conf_logprob: list[float] = []
    conf_compression: list[float] = []
    for seg in segments:
        if forced_language == "italian" and _ASR_ITALIAN_BACKEND == "transformers":
            text, seg_meta = _coerce_transcribe_result(
                get_italian_pipe_transformers().transcribe(
                    seg,
                    SAMPLE_RATE,
                    forced_language="italian",
                    max_new_tokens=max_new_tokens,
                    num_beams=num_beams,
                    no_speech_threshold=ns_thr,
                    compression_ratio_threshold=_COMPRESSION_RATIO_THRESHOLD,
                    logprob_threshold=_LOGPROB_THRESHOLD,
                )
            )
        else:
            pipe = get_pipe()
            text, seg_meta = _coerce_transcribe_result(
                pipe.transcribe(
                    seg,
                    SAMPLE_RATE,
                    forced_language=forced_language,
                    max_new_tokens=max_new_tokens,
                    num_beams=num_beams,
                    no_speech_threshold=ns_thr,
                    compression_ratio_threshold=_COMPRESSION_RATIO_THRESHOLD,
                    logprob_threshold=_LOGPROB_THRESHOLD,
                )
            )
        nsp = seg_meta.get("no_speech_prob")
        if nsp is not None:
            conf_no_speech.append(float(nsp))
        alp = seg_meta.get("avg_logprob")
        if alp is not None:
            conf_logprob.append(float(alp))
        cr = seg_meta.get("compression_ratio")
        if cr is not None:
            conf_compression.append(float(cr))
        t = (text or "").strip()
        if not t:
            continue
        ded = (
            _remove_overlap_repetition(prev, t, max_overlap_words=_ASR_CROSS_CHUNK_OVERLAP_WORDS)
            if prev
            else t
        )
        ded = ded.strip()
        if ded:
            out_parts.append(ded)
        prev = t

    text_out = " ".join(out_parts).strip()
    zlib_cr = max(conf_compression) if conf_compression else None
    conf = {
        "no_speech_prob": (sum(conf_no_speech) / len(conf_no_speech)) if conf_no_speech else None,
        "avg_logprob": (sum(conf_logprob) / len(conf_logprob)) if conf_logprob else None,
        "compression_ratio": float(zlib_cr) if zlib_cr is not None else _text_compression_ratio(text_out),
    }
    return text_out, conf


def register_client_log_route(app: FastAPI) -> None:
    """POST /client_log — HoloLens telemetry. Defined here because HF Spaces often ship only ``/app/main.py``."""

    router = APIRouter(tags=["telemetry"])

    @router.post("/client_log")
    async def client_log(request: Request) -> Response:
        try:
            raw = await request.body()
            if not raw:
                print("[client_log] event=<empty_body>")
                return Response(content='{"ok":true}', media_type="application/json")
            payload = json.loads(raw.decode("utf-8"))
            if not isinstance(payload, dict):
                print(f"[client_log] event=INVALID_BODY type={type(payload).__name__}")
                return Response(content='{"ok":false}', media_type="application/json", status_code=400)
            event = str(payload.get("event") or payload.get("type") or "UNKNOWN").strip() or "UNKNOWN"
            rest = {k: v for k, v in payload.items() if k not in ("event", "type")}
            extras_repr = ""
            if rest:
                try:
                    extras_repr = " " + json.dumps(rest, ensure_ascii=False, default=str)[:800]
                except Exception:
                    extras_repr = " " + str(rest)[:800]
            print(f"[client_log] event={event}{extras_repr}")
        except json.JSONDecodeError as e:
            print(f"[client_log] event=PARSE_ERROR err={e!r}")
            return Response(content='{"ok":false}', media_type="application/json", status_code=400)
        except Exception as e:
            print(f"[client_log] event=ERROR err={e!r}")
            return Response(content='{"ok":false}', media_type="application/json", status_code=500)

        return Response(content='{"ok":true}', media_type="application/json")

    app.include_router(router)
    print("[startup] POST /client_log registered (HoloLens telemetry)")


app = FastAPI(title="Whisper ASR")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

register_client_log_route(app)

INDEX_HTML = """
<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>Whisper ASR Space Test</title>
    <style>
      body { font-family: system-ui, sans-serif; margin: 0; background: #0b1220; color: #e2e8f0; }
      .wrap { max-width: 820px; margin: 0 auto; padding: 24px; }
      .title { font-size: 1.5rem; font-weight: 650; margin-bottom: 8px; }
      .sub { color: #94a3b8; margin-bottom: 16px; }
      .row { display: flex; gap: 10px; align-items: center; margin-bottom: 14px; }
      button { border: 0; border-radius: 999px; padding: 10px 18px; cursor: pointer; font-weight: 600; }
      #start { background: #22c55e; color: #052e16; }
      #stop { background: #ef4444; color: white; }
      #clear { background: #334155; color: #e2e8f0; }
      #stop:disabled, #start:disabled { opacity: 0.5; cursor: not-allowed; }
      .status { margin-left: auto; font-size: 0.9rem; color: #93c5fd; }
      .box { background: #111827; border: 1px solid #334155; border-radius: 12px; padding: 14px; min-height: 110px; line-height: 1.6; }
      .label { margin: 10px 0 8px; color: #cbd5e1; font-size: 0.86rem; text-transform: uppercase; letter-spacing: .04em; }
      .err { color: #fca5a5; min-height: 20px; margin-top: 8px; }
    </style>
  </head>
  <body>
    <div class="wrap">
      <div class="title">Whisper ASR Browser Test</div>
      <div class="sub">Speak in English or Italian. This page sends raw float32 chunks to <code>/audio</code> (same API your clients use).</div>
      <div class="row">
        <button id="start" onclick="startMic()">Start</button>
        <button id="stop" onclick="stopMic()" disabled>Stop</button>
        <button id="clear" onclick="clearText()">Clear</button>
        <select id="lang-select" style="border-radius: 10px; padding: 8px 10px; border: 1px solid #475569; background: #0f172a; color: #e2e8f0;">
          <option value="auto">Auto (best quality, slower)</option>
          <option value="italian">Italian (faster)</option>
          <option value="english">English (faster)</option>
        </select>
        <div id="status" class="status">Idle</div>
      </div>
      <div class="label">Transcript</div>
      <div id="out" class="box">Listening...</div>
      <div id="err" class="err"></div>
    </div>
    <script>
      let audioContext = null, stream = null, processor = null, source = null;
      let sampleRate = 16000, buf = [], transcript = "", speaking = false;
      let rmsSamples = [], lastSpeechAt = 0, speechStartAt = 0, noiseRms = 0.01, hasSpeech = false, tick = null;
      let chunkSpeechStartAt = 0;
      let sendInFlight = false;
      const MAX_SECONDS = 3.6, MIN_NOISE_FLOOR = 0.008, SPEECH_MULT = 3.1, SILENCE_MS = 460, MIN_SPEECH_MS = 220;

      function render() {
        document.getElementById("out").textContent = transcript || "Listening...";
      }
      function clearText() { transcript = ""; document.getElementById("err").textContent = ""; render(); }

      function startMic() {
        navigator.mediaDevices.getUserMedia({ audio: true }).then((s) => {
          stream = s;
          audioContext = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: 16000 });
          sampleRate = audioContext.sampleRate;
          source = audioContext.createMediaStreamSource(stream);
          processor = audioContext.createScriptProcessor(4096, 1, 1);
          processor.onaudioprocess = (e) => {
            const input = e.inputBuffer.getChannelData(0);
            for (let i = 0; i < input.length; i++) {
              buf.push(input[i]);
              rmsSamples.push(input[i]);
              if (rmsSamples.length > 3200) rmsSamples.shift();
            }
            const maxSamples = Math.floor(MAX_SECONDS * sampleRate);
            while (buf.length > maxSamples) buf.shift();
          };
          source.connect(processor);
          processor.connect(audioContext.destination);
          document.getElementById("start").disabled = true;
          document.getElementById("stop").disabled = false;
          document.getElementById("status").textContent = "Listening";
          transcript = "";
          document.getElementById("err").textContent = "";
          render();
          speaking = false; lastSpeechAt = 0; speechStartAt = 0; noiseRms = 0.01; hasSpeech = false;
          chunkSpeechStartAt = 0;
          sendInFlight = false;
          if (tick) clearInterval(tick);
          tick = setInterval(vadTick, 200);
        }).catch((e) => {
          document.getElementById("err").textContent = "Mic error: " + (e.message || "permission denied");
        });
      }

      async function stopMic() {
        if (tick) { clearInterval(tick); tick = null; }
        if (buf.length && hasSpeech) await sendAudio();
        if (stream) { stream.getTracks().forEach(t => t.stop()); stream = null; }
        try { if (source) source.disconnect(); if (processor) processor.disconnect(); } catch(e) {}
        buf = []; rmsSamples = []; speaking = false; lastSpeechAt = 0; speechStartAt = 0; noiseRms = 0.01; hasSpeech = false;
        chunkSpeechStartAt = 0;
        sendInFlight = false;
        document.getElementById("start").disabled = false;
        document.getElementById("stop").disabled = true;
        document.getElementById("status").textContent = "Idle";
      }

      function sendAudio() {
        if (sendInFlight) {
          // Keep enough context but avoid long lag tails when the server is still busy.
          const keep = Math.floor(sampleRate * 4.0);
          if (buf.length > keep) buf = buf.slice(buf.length - keep);
          return Promise.resolve();
        }
        if (!buf.length || !hasSpeech) { buf = []; hasSpeech = false; return Promise.resolve(); }
        sendInFlight = true;
        const samples = buf.slice();
        // Keep a small overlap tail so words cut at boundaries can be completed in next chunk.
        const overlap = Math.floor(sampleRate * 0.45);
        const tail = overlap > 0 ? samples.slice(Math.max(0, samples.length - overlap)) : [];
        buf = tail;
        hasSpeech = false;
        chunkSpeechStartAt = 0;
        const selectedLanguage = (document.getElementById("lang-select").value || "auto");
        const ab = new ArrayBuffer(samples.length * 4);
        const view = new DataView(ab);
        for (let i = 0; i < samples.length; i++) view.setFloat32(i * 4, samples[i], true);
        return fetch("/audio", {
          method: "POST",
          headers: {
            "X-Sample-Rate": String(sampleRate),
            "X-Forced-Language": selectedLanguage
          },
          body: ab
        })
          .then(r => r.json())
          .then((data) => {
            console.log("ASR raw response:", data);
            if (data.error) document.getElementById("err").textContent = data.error;
            const t = ((data.text ?? data.transcript ?? data.transcription ?? data.result) || "").trim();
            if (t) transcript = transcript ? (transcript + " " + t) : t;
            render();
          })
          .catch((e) => { document.getElementById("err").textContent = "Network error: " + (e.message || "failed"); })
          .finally(() => { sendInFlight = false; });
      }

      function vadTick() {
        if (!rmsSamples.length) return;
        let sum = 0;
        for (let i = 0; i < rmsSamples.length; i++) sum += rmsSamples[i] * rmsSamples[i];
        const rms = Math.sqrt(sum / rmsSamples.length);
        const now = Date.now();
        const threshold = Math.max(MIN_NOISE_FLOOR, noiseRms * SPEECH_MULT);
        if (rms > threshold) {
          if (!speaking) { speaking = true; speechStartAt = now; }
          lastSpeechAt = now;
          if ((now - speechStartAt) >= MIN_SPEECH_MS) {
            hasSpeech = true;
            if (!chunkSpeechStartAt) chunkSpeechStartAt = now;
          }
          // Do not force periodic flush while speaking: wait for silence boundary
          // so returned text is closer to complete sentences.
        } else {
          noiseRms = 0.98 * noiseRms + 0.02 * rms;
          if (speaking && (now - lastSpeechAt) > SILENCE_MS) {
            speaking = false;
            sendAudio();
          }
        }
      }
    </script>
  </body>
</html>
"""


@app.on_event("startup")
def startup() -> None:
    print(f"[startup] torch.cuda.is_available={torch.cuda.is_available()} torch.cuda.device_count={torch.cuda.device_count()}")
    print(
        f"[startup] DEVICE={DEVICE} MODEL_ID_PRIMARY={MODEL_ID_PRIMARY} "
        f"MODEL_ID_FALLBACK={MODEL_ID_FALLBACK} ASR_FALLBACK_ON_EMPTY={_FALLBACK_ON_EMPTY}"
    )
    if DEVICE.type == "cuda":
        print(f"[startup] ASR_USE_FP16={_USE_FP16_CUDA} (fp16 autocast on GPU)")
    else:
        print(
            "[startup] DEVICE is CPU: the Space has no GPU selected. Whisper+beam search stays slow; "
            "this is not fixable in Python alone. Space Settings → Hardware → pick a GPU tier "
            "(paid on HF). fp16 activates automatically when CUDA is available."
        )
    print(
        f"[startup] ASR_WHISPER_NUM_BEAMS={_WHISPER_BEAMS} ASR_WHISPER_MAX_NEW_TOKENS={_WHISPER_MAX_NEW} "
        f"ASR_FORCED_NUM_BEAMS={_ASR_FORCED_NUM_BEAMS} ASR_FORCED_MAX_NEW_TOKENS={_ASR_FORCED_MAX_NEW} "
        f"ASR_ENGLISH_FAST_ON_CUDA={_ASR_ENGLISH_FAST_ON_CUDA} "
        f"ASR_EN_SHORT_RETRY={_ASR_EN_SHORT_RETRY} "
        f"ASR_EN_SHORT_RETRY_NO_SPEECH_THRESHOLD={_ASR_EN_SHORT_RETRY_NS_THR} "
        f"ASR_IT_SHORT_RETRY={_ASR_IT_SHORT_RETRY} "
        f"ASR_IT_SHORT_RETRY_NO_SPEECH_THRESHOLD={_ASR_IT_SHORT_RETRY_NS_THR} "
        f"ASR_MAX_CHUNK_SECONDS_FORCED={_ASR_MAX_CHUNK_S_FORCED} ASR_MAX_CHUNK_SECONDS_AUTO={_ASR_MAX_CHUNK_S_AUTO} "
        f"ASR_CHUNK_OVERLAP_SECONDS={_ASR_CHUNK_OVERLAP_S} "
        f"ASR_NO_SPEECH_THRESHOLD_EN={_NO_SPEECH_THRESHOLD_EN} "
        f"ASR_NO_SPEECH_THRESHOLD_IT={_NO_SPEECH_THRESHOLD_IT} "
        f"ASR_AVG_LOGPROB_REJECT_EN={_ASR_AVG_LOGPROB_REJECT_EN} "
        f"ASR_MAX_WORDS_PER_SECOND_EN={_ASR_MAX_WORDS_PER_SECOND_EN} "
        f"ASR_AVG_LOGPROB_REJECT_IT={_ASR_AVG_LOGPROB_REJECT_IT} "
        f"ASR_IT_RESTORE_ON_FILTER={_ASR_IT_RESTORE_ON_FILTER} "
        f"ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_ACCEPT={_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_ACCEPT} "
        f"ASR_MIN_SPEECH_RMS_FORCED_ITALIAN={_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN} "
        f"ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT={_ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT} "
        f"ASR_MIN_TRIMMED_SEGMENT_SECONDS_EN={_ASR_MIN_TRIMMED_SEGMENT_SECONDS_EN} "
        f"ASR_MIN_TRIMMED_SEGMENT_SECONDS_IT={_ASR_MIN_TRIMMED_SEGMENT_SECONDS_IT} "
        f"ASR_SHORT_MIN_CHUNK_SECONDS_FORCED_ITALIAN={_ASR_SHORT_MIN_CHUNK_S_FORCED_ITALIAN} "
        f"ASR_SHORT_MAX_CHUNK_SECONDS_FORCED_ITALIAN={_ASR_SHORT_MAX_CHUNK_S_FORCED_ITALIAN} "
        f"ASR_MIN_CHUNK_SECONDS={_ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN} "
        f"ASR_VAD_FILTER={_ASR_VAD_FILTER_FAST} "
        f"VAD_IT_EFFECTIVE={_vad_enabled_forced_italian()} "
        f"ASR_TRIM_TRAILING_SILENCE_IT={_ASR_TRIM_TRAILING_SILENCE_IT} "
        f"ASR_TRIM_TRAILING_SILENCE_EN={_ASR_TRIM_TRAILING_SILENCE_EN} "
        f"ASR_FORCE_SEGMENTATION_FORCED={_ASR_FORCE_SEGMENTATION_FORCED} "
        f"ASR_COMPRESSION_RATIO_THRESHOLD_IT={_COMPRESSION_RATIO_THRESHOLD_IT} "
        f"ASR_ULTRA_SHORT_TAIL_WINDOW_S={_ASR_ULTRA_SHORT_TAIL_WINDOW_S} "
        f"ASR_ULTRA_SHORT_TAIL_MAX_SEGMENT_S={_ASR_ULTRA_SHORT_TAIL_MAX_SEGMENT_S} "
        f"ASR_MIN_RMS={_MIN_RMS}"
    )
    if _HOLOLENS_ITA_PROFILE:
        print(
            "[startup] ASR_HOLOLENS_ITA_PROFILE=1 (bundle for forced Italian + HoloLens chunking; "
            "see ASR/HOLOLENS_TUNING.md)"
        )
    if DEVICE.type == "cpu":
        print(
            f"[startup] ASR_CPU_THREADS={_CPU_THREADS} ASR_CPU_INTEROP_THREADS={_CPU_INTEROP_THREADS} "
            f"ASR_CPU_RELIABLE_MODE={_CPU_RELIABLE_MODE} ASR_CPU_BALANCED_MODE={_CPU_BALANCED_MODE} "
            f"ASR_CPU_AUTO_USE_REALTIME={_ASR_CPU_AUTO_USE_REALTIME} "
            f"ASR_CPU_SERIALIZE={_CPU_RELIABLE_MODE}"
        )
        if _CPU_RELAX_NEAR_DUP:
            print(
                "[startup] CPU reliable: near-duplicate blanking relaxed "
                "(anti-hallucination filters stay on)."
            )
    print(f"[startup] HF token configured: {bool(_hf_token())} (needed for private/gated models)")
    try:
        get_pipe().load()
        # Warm forced-English realtime path and forced-Italian transformers path
        # to avoid first-request latency spikes.
        p = get_pipe()
        if isinstance(p, FasterWhisperASR):
            p._ensure_en_rt()
        # Forced Italian transformers model is large (can be hundreds of MB to GB).
        # Avoid crashing the whole container if warm-load OOMs; fallback to the
        # non-transformers Italian backend (faster-whisper/shared) instead.
        warm_load_it_transformers = os.environ.get("ASR_WARM_LOAD_IT_TRANSFORMERS", "1").strip().lower() in (
            "1",
            "true",
            "yes",
        )
        if _ASR_ITALIAN_BACKEND == "transformers":
            if warm_load_it_transformers:
                try:
                    get_italian_pipe_transformers().load()
                    print("[startup] Warm-loaded forced-Italian transformers (ASR_WARM_LOAD_IT_TRANSFORMERS=1).")
                except Exception as e:
                    print(f"[startup] Forced-Italian transformers warm-load failed (lazy-load on demand): {e}")
            else:
                print(
                    "[startup] Skip warm-load for forced-Italian transformers "
                    "(ASR_WARM_LOAD_IT_TRANSFORMERS!=1). Will lazy-load on first forced-Italian /audio call."
                )
    except OSError as e:
        print(
            "[startup] FAILED to load model. For a **private** repo: Space → Settings → "
            "**Repository secrets** → add secret name **HF_TOKEN** (fine-grained token with Read on the model). "
            "Do not use Variables for secrets."
        )
        # If Italian transformers warm-load fails, fall back to safer non-transformers path.
        # This prevents the Space from failing to start (build error).
        try:
            globals()["_ASR_ITALIAN_BACKEND"] = "shared"
            print("[startup] Fallback: set _ASR_ITALIAN_BACKEND=shared after warm-load failure.")
        except Exception:
            pass
        # Re-raise for other model load failures only if they are critical.
        # In practice, non-transformers paths can still serve /audio.
        # (Avoid hard-failing the container.)
    print("[startup] model ready")


@app.get("/")
def root() -> HTMLResponse:
    return HTMLResponse(content=INDEX_HTML)


@app.get("/info")
def info() -> dict:
    return {
        "service": "whisper-asr-single",
        "model_id_primary": MODEL_ID_PRIMARY,
        "model_id_fallback": MODEL_ID_FALLBACK,
        "audio_post": "/audio",
        "header": "X-Sample-Rate (optional, default 16000)",
        "body": "raw float32 little-endian mono PCM",
    }


@app.get("/health")
def health() -> dict:
    return {"ok": True, "model_id_primary": MODEL_ID_PRIMARY, "model_id_fallback": MODEL_ID_FALLBACK}


@app.post("/audio")
async def audio(request: Request) -> Response:
    global _asr_busy
    t0 = time.perf_counter()
    try:
        raw = await request.body()
    except ClientDisconnect:
        # Client closed the connection before the body finished (abort, navigation, duplicate POST).
        return Response(content=json.dumps({"text": ""}), media_type="application/json")
    if not raw or len(raw) < 4:
        return Response(
            content=json.dumps({"text": "", "error": "empty body"}),
            media_type="application/json",
            status_code=400,
        )
    t_body_ms = (time.perf_counter() - t0) * 1000.0
    try:
        sr = int(request.headers.get("X-Sample-Rate", "16000"))
    except ValueError:
        sr = SAMPLE_RATE
    n = len(raw) // 4
    try:
        floats = list(struct.unpack("<" + "f" * n, raw[: n * 4]))
    except struct.error:
        return Response(
            content=json.dumps({"text": "", "error": "invalid float32 payload"}),
            media_type="application/json",
            status_code=400,
        )

    t_pre_start = time.perf_counter()
    chunk = preprocess_chunk(floats, sr)
    t_pre_ms = (time.perf_counter() - t_pre_start) * 1000.0
    if chunk is None:
        total_ms = (time.perf_counter() - t0) * 1000.0
        print(f"[timing] /audio dropped=silence total_ms={total_ms:.1f} preprocess_ms={t_pre_ms:.1f}")
        return Response(content=json.dumps({"text": ""}), media_type="application/json")
    arr, rms, duration_s = chunk

    forced_language = (request.headers.get("X-Forced-Language", "auto") or "auto").strip().lower()
    if forced_language not in {"auto", "italian", "english"}:
        forced_language = "auto"
    raw_chunk_s = duration_s
    if forced_language in {"italian", "english"}:
        if _ASR_FORCE_SEGMENTATION_FORCED:
            seg_arr, seg_start_s, seg_end_s = _extract_strongest_speech_segment(arr)
            seg_s = float(seg_arr.size) / float(SAMPLE_RATE) if seg_arr.size > 0 else 0.0
            print(
                f"[segment] lang={forced_language} raw_chunk_s={raw_chunk_s:.2f} "
                f"segment_s={seg_s:.2f} start={seg_start_s:.2f} end={seg_end_s:.2f}"
            )
            speech_rms_tmp = float(np.sqrt(np.mean(seg_arr**2))) if seg_arr.size > 0 else 0.0
            hard_min_seg_s = 0.35
            if seg_s < hard_min_seg_s:
                print(
                    f"[reject] lang={forced_language} reason=trimmed_segment_too_short "
                    f"segment_s={seg_s:.2f} speech_rms={speech_rms_tmp:.5f}"
                )
                return Response(content=json.dumps({"text": ""}), media_type="application/json")
            arr = seg_arr
            rms = speech_rms_tmp
            duration_s = seg_s
            print(f"[segment_accept] lang={forced_language} segment_s={seg_s:.2f} speech_rms={speech_rms_tmp:.5f}")
        else:
            # Stable demo mode: decode one phrase-sized chunk, only trim obvious edge silence.
            if (
                (forced_language == "italian" and _ASR_TRIM_TRAILING_SILENCE_IT)
                or (forced_language == "english" and _ASR_TRIM_TRAILING_SILENCE_EN)
            ):
                arr_trim, trim_lead_s, trim_tail_s = _trim_edge_silence_forced_chunk(arr)
                if arr_trim.size > 0 and (trim_lead_s > 0.002 or trim_tail_s > 0.002):
                    arr = arr_trim
                    rms = float(np.sqrt(np.mean(arr**2))) if arr.size > 0 else rms
                    duration_s = float(arr.size) / float(SAMPLE_RATE)
                    print(
                        f"[segment] lang={forced_language} raw_chunk_s={raw_chunk_s:.2f} "
                        f"segment_s={duration_s:.2f} start={trim_lead_s:.2f} end={max(0.0, raw_chunk_s - trim_tail_s):.2f}"
                    )
    short_mode_forced_it = False
    short_decode_forced_it = False
    speech_rms = rms
    speech_rms_p90 = rms
    # Real-time safety: avoid expensive inference on weak chunks.
    # In CPU reliable mode we do not skip low-RMS chunks, to avoid losing words.
    rms_skip = 0.0 if (DEVICE.type == "cpu" and _CPU_RELIABLE_MODE) else (
        _ASR_RMS_SKIP_AUTO if forced_language == "auto" else _ASR_RMS_SKIP_FORCED
    )
    if forced_language != "italian" and rms < rms_skip:
        total_ms = (time.perf_counter() - t0) * 1000.0
        print(
            f"[timing] /audio dropped=low_rms lang={forced_language} rms={rms:.5f} "
            f"threshold={rms_skip:.5f} total_ms={total_ms:.1f} preprocess_ms={t_pre_ms:.1f}"
        )
        return Response(content=json.dumps({"text": ""}), media_type="application/json")
    # Forced-Italian speech-window RMS policy (avoid full-chunk averaging with silence).
    if forced_language == "italian":
        speech_rms, speech_rms_p90 = _forced_italian_speech_rms(arr)
        if (
            duration_s >= _ASR_SHORT_MIN_CHUNK_S_FORCED_ITALIAN
            and duration_s < _ASR_SHORT_MAX_CHUNK_S_FORCED_ITALIAN
        ):
            short_mode_forced_it = True
            print(
                f"[short_ok] lang=italian chunk_s={duration_s:.2f} "
                f"full_rms={rms:.5f} speech_rms={speech_rms:.5f} p90={speech_rms_p90:.5f}"
            )
            short_decode_forced_it = True
            print(
                f"[short_decode] lang=italian chunk_s={duration_s:.2f} "
                f"full_rms={rms:.5f} speech_rms={speech_rms:.5f}"
            )
        elif _italian_relaxed_micro_fragment(duration_s, speech_rms):
            short_decode_forced_it = True
            print(
                f"[micro_ok] lang=italian chunk_s={duration_s:.2f} "
                f"full_rms={rms:.5f} speech_rms={speech_rms:.5f} p90={speech_rms_p90:.5f}"
            )

    # Forced-language micro-fragment guard: reject ultra-short trimmed segments by default.
    if forced_language == "english":
        min_trimmed_s = float(_ASR_MIN_TRIMMED_SEGMENT_SECONDS_EN)
    elif forced_language == "italian":
        min_trimmed_s = float(_ASR_MIN_TRIMMED_SEGMENT_SECONDS_IT)
    else:
        min_trimmed_s = 0.0
    if forced_language in {"english", "italian"} and min_trimmed_s > 0.0 and float(duration_s) < min_trimmed_s:
        total_ms = (time.perf_counter() - t0) * 1000.0
        print(
            f"[timing] /audio dropped=trimmed_segment_too_short_forced lang={forced_language} "
            f"segment_s={duration_s:.2f} min={min_trimmed_s:.2f} total_ms={total_ms:.1f} preprocess_ms={t_pre_ms:.1f}"
        )
        return Response(content=json.dumps({"text": ""}), media_type="application/json")

    # Do not decode forced-language short/unstable windows.
    if (
        forced_language in {"italian", "english"}
        and _ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN > 0.0
        and raw_chunk_s < _ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN
    ):
        total_ms = (time.perf_counter() - t0) * 1000.0
        print(
            f"[timing] /audio dropped=short_chunk_forced lang={forced_language} raw_chunk_s={raw_chunk_s:.2f} "
            f"min={_ASR_MIN_CHUNK_SECONDS_FORCED_ITALIAN:.2f} total_ms={total_ms:.1f} preprocess_ms={t_pre_ms:.1f}"
        )
        return Response(content=json.dumps({"text": ""}), media_type="application/json")

    # Forced Italian VAD pre-check to reject low-energy/no-speech tails before decode.
    if forced_language == "italian" and _vad_enabled_forced_italian() and not short_decode_forced_it:
        if not _forced_italian_vad_has_speech(arr):
            print(f"[reject] lang=italian reason=vad_no_speech rms={rms:.5f} chunk_s={duration_s:.2f}")
            return Response(content=json.dumps({"text": ""}), media_type="application/json")

    # If backend is still transcribing previous chunk, drop this one to prevent queue-lag buildup.
    # Disabled in CPU reliable mode to preserve transcript completeness.
    if _ASR_DROP_WHEN_BUSY and _asr_busy:
        total_ms = (time.perf_counter() - t0) * 1000.0
        print(
            f"[timing] /audio dropped=busy lang={forced_language} chunk_s={duration_s:.2f} "
            f"rms={rms:.5f} total_ms={total_ms:.1f}"
        )
        return Response(content=json.dumps({"text": ""}), media_type="application/json")

    t_asr_start = time.perf_counter()
    # On CPU reliable mode, serialize inference to prevent parallel-request thrashing.
    infer_lock = _asr_infer_lock if (DEVICE.type == "cpu" and _CPU_RELIABLE_MODE) else None
    if infer_lock is not None:
        infer_lock.acquire()
    _asr_busy = True
    decode_conf: dict = {"no_speech_prob": None, "avg_logprob": None, "compression_ratio": None}
    try:
        forced_num_beams = _ASR_FORCED_NUM_BEAMS if forced_language in {"english", "italian"} else _WHISPER_BEAMS
        forced_max_new = _ASR_FORCED_MAX_NEW if forced_language in {"english", "italian"} else _WHISPER_MAX_NEW

        raw_text, decode_conf = _transcribe_chunked(
            arr,
            forced_language=forced_language,
            max_new_tokens=forced_max_new,
            num_beams=forced_num_beams,
        )
        raw_text_original = (raw_text or "").strip()
    finally:
        _asr_busy = False
        if infer_lock is not None:
            infer_lock.release()
    t_asr_ms = (time.perf_counter() - t_asr_start) * 1000.0
    if forced_language == "italian":
        # Enforce Italian-only output at response layer.
        raw_text = _sanitize_forced_italian_output(raw_text or "")
        raw_text = _strip_dangling_italian_tail(raw_text)
        word_count = len(_WORD_RE.findall(raw_text)) if raw_text else 0
        # Conservative forced-Italian acceptance (confidence/audio based, no lexical blocklist).
        it_accept_floor = _italian_accept_rms_threshold(duration_s, speech_rms)
        if raw_text and speech_rms < it_accept_floor:
            print(
                f"[reject] lang=italian reason=below_accept_threshold "
                f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f} "
                f"threshold={it_accept_floor:.5f} chunk_s={duration_s:.2f}"
            )
            raw_text = ""
            word_count = 0
        if raw_text and word_count <= 2 and speech_rms < 0.025:
            print(
                f"[reject] lang=italian reason=short_weak_output "
                f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f} words={word_count}"
            )
            raw_text = ""
            word_count = 0
        # Conservative acceptance filter: reject weak/low-confidence short outputs.
        min_speech_thr = (
            _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN_SHORT
            if short_mode_forced_it or _italian_relaxed_micro_fragment(duration_s, speech_rms)
            else _ASR_MIN_SPEECH_RMS_FORCED_ITALIAN
        )
        if raw_text and speech_rms < min_speech_thr:
            print(
                f"[reject] lang=italian reason=post_low_speech_rms "
                f"speech_rms={speech_rms:.5f} threshold={min_speech_thr:.5f} chunk_s={duration_s:.2f}"
            )
            raw_text = ""
        # Decode-confidence gates: Italian transformers often omits nsp/alp — do not hard-drop on zlib CR alone.
        if raw_text:
            no_speech_prob = decode_conf.get("no_speech_prob")
            avg_logprob = decode_conf.get("avg_logprob")
            comp = decode_conf.get("compression_ratio")
            words_it = len(_WORD_RE.findall(raw_text))
            reasonable_it = len((raw_text or "").strip()) >= 8 or words_it >= 2
            meta_missing = no_speech_prob is None and avg_logprob is None
            cr_val = float(comp) if comp is not None else None

            nsp_bad = no_speech_prob is not None and float(no_speech_prob) > 0.55
            alp_bad = avg_logprob is not None and float(avg_logprob) < float(_ASR_AVG_LOGPROB_REJECT_IT)
            skip_cr_proxy = meta_missing and speech_rms >= 0.035 and reasonable_it
            cr_hard_it = cr_val is not None and cr_val >= float(_COMPRESSION_RATIO_THRESHOLD_IT)
            cr_soft_bad = (
                cr_val is not None
                and cr_val > float(_COMPRESSION_RATIO_THRESHOLD) + 0.2
                and not skip_cr_proxy
            )
            cr_repeat_bad = (
                cr_val is not None
                and cr_val > float(_COMPRESSION_RATIO_THRESHOLD) + 0.2
                and _should_drop_repetition_hallucination(raw_text)
            )
            comp_bad = bool(cr_hard_it or cr_soft_bad or cr_repeat_bad)
            weak_decode = bool(nsp_bad or alp_bad or comp_bad)

            if weak_decode:
                print(
                    f"[it_filter] decision=reject reason=confidence_meta "
                    f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f} compression_ratio={cr_val} "
                    f"nsp_bad={nsp_bad} alp_bad={alp_bad} comp_bad={comp_bad} skip_cr_proxy={skip_cr_proxy}"
                )
                print(
                    f"[reject] lang=italian reason=confidence_meta "
                    f"chunk_s={duration_s:.2f} speech_rms={speech_rms:.5f} "
                    f"no_speech_prob={no_speech_prob} avg_logprob={avg_logprob} "
                    f"compression_ratio={comp}"
                )
                raw_text = ""
            elif cr_val is not None or not meta_missing:
                print(
                    f"[it_filter] decision=accept reason=confidence_ok "
                    f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f} compression_ratio={cr_val}"
                )
        # Short-utterance mode: stricter post-decode confidence checks.
        if short_mode_forced_it and raw_text:
            words_now = [w for w in _WORD_RE.findall(raw_text)]
            en_sc, it_sc = _lang_marker_scores(raw_text)
            if _should_drop_language_mismatch(raw_text, "italian") or _should_drop_repetition_hallucination(raw_text):
                print(
                    f"[reject] lang=italian reason=short_confidence "
                    f"chunk_s={duration_s:.2f} rms={rms:.5f} text={raw_text!r}"
                )
                raw_text = ""
            elif len(words_now) > 1 and it_sc < 0.12 and en_sc > it_sc:
                # Require stronger Italian evidence for multi-word short chunks.
                print(
                    f"[reject] lang=italian reason=short_confidence_markers "
                    f"chunk_s={duration_s:.2f} rms={rms:.5f} it_score={it_sc:.3f} en_score={en_sc:.3f}"
                )
                raw_text = ""
        if raw_text and _should_drop_whisper_low_energy_hallucination(raw_text, rms, duration_s):
            raw_text = ""
    elif forced_language == "english":
        # Enforce English-only output at response layer.
        raw_text = _sanitize_forced_english_output(raw_text or "")
        if raw_text:
            words_en = len(_WORD_RE.findall(raw_text))
            chunk_s_en = max(0.2, float(duration_s))
            wps_en = float(words_en) / chunk_s_en
            nsp_en = decode_conf.get("no_speech_prob")
            alp_en = decode_conf.get("avg_logprob")
            conf_low_en = (
                (nsp_en is not None and float(nsp_en) > _NO_SPEECH_THRESHOLD_EN)
                or (alp_en is not None and float(alp_en) < _ASR_AVG_LOGPROB_REJECT_EN)
            )
            accept_thr_en = float(_ASR_MIN_SPEECH_RMS_FORCED_ENGLISH_ACCEPT)
            # High words/sec alone is not a reject (fast real speech). Pair with another bad signal.
            if wps_en > _ASR_MAX_WORDS_PER_SECOND_EN:
                paired = _english_word_density_secondary_signals(
                    raw_text,
                    speech_rms=float(speech_rms),
                    accept_thr=accept_thr_en,
                    decode_conf=decode_conf,
                )
                if paired:
                    print(
                        f"[reject] lang=english reason=english_word_density "
                        f"text={raw_text[:80]!r} wps={wps_en:.2f} max_wps={_ASR_MAX_WORDS_PER_SECOND_EN:.2f} "
                        f"chunk_s={chunk_s_en:.2f} no_speech_prob={nsp_en} avg_logprob={alp_en} "
                        f"paired={';'.join(paired)}"
                    )
                    raw_text = ""
            elif conf_low_en:
                print(
                    f"[reject] lang=english reason=english_confidence "
                    f"text={raw_text[:80]!r} chunk_s={chunk_s_en:.2f} "
                    f"no_speech_prob={nsp_en} ns_thr={_NO_SPEECH_THRESHOLD_EN:.2f} "
                    f"avg_logprob={alp_en} alp_thr={_ASR_AVG_LOGPROB_REJECT_EN:.2f}"
                )
                raw_text = ""
            elif raw_text:
                # Multi-sentence outputs are rejected only when they are also low-confidence
                # and low-energy; keep fluent short utterances.
                sentence_count = len(re.findall(r"[.!?]+", raw_text))
                if sentence_count >= 2:
                    low_conf_for_multi = (
                        (nsp_en is not None and float(nsp_en) > 0.35)
                        or (alp_en is not None and float(alp_en) < -0.75)
                    )
                    low_energy_for_multi = float(speech_rms) < 0.03
                    if low_conf_for_multi and low_energy_for_multi:
                        print(
                            f"[reject] lang=english reason=english_multi_sentence_short_low_conf "
                            f"text={raw_text[:80]!r} chunk_s={chunk_s_en:.2f} sentences={sentence_count} "
                            f"speech_rms={speech_rms:.5f} no_speech_prob={nsp_en} avg_logprob={alp_en}"
                        )
                        raw_text = ""
        if raw_text and _should_drop_whisper_low_energy_hallucination(raw_text, rms, duration_s):
            raw_text = ""
    elif DEVICE.type == "cpu" and _CPU_RELIABLE_MODE:
        # Keep CPU completeness, but still avoid obvious cross-language mismatches
        # when user explicitly locked the language.
        if forced_language in {"english", "italian"} and raw_text and _should_drop_language_mismatch(raw_text, forced_language):
            raw_text = ""
    else:
        if forced_language == "auto" and raw_text and _should_drop_low_information_fragment(
            raw_text, rms, duration_s, forced_language=forced_language
        ):
            raw_text = ""
        if raw_text and _should_drop_language_mismatch(raw_text, forced_language):
            raw_text = ""
    if raw_text and forced_language in {"english", "italian"}:
        words_n = len(_WORD_RE.findall(raw_text))
        accept_thr = (
            _italian_accept_rms_threshold(duration_s, speech_rms)
            if forced_language == "italian"
            else _ASR_MIN_SPEECH_RMS_FORCED_ENGLISH_ACCEPT
        )
        if speech_rms < accept_thr:
            print(
                f"[reject] lang={forced_language} reason=below_accept_threshold "
                f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f} threshold={accept_thr:.5f}"
            )
            raw_text = ""
        if raw_text and re.match(r"^\s*[\.,!?;:]+", raw_text):
            print(
                f"[reject] lang={forced_language} reason=starts_with_punctuation "
                f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f}"
            )
            raw_text = ""
        if raw_text:
            max_words = max(4, int(math.ceil(duration_s * 3.6)))
            if words_n > max_words:
                if forced_language == "english":
                    paired_tm = _english_word_density_secondary_signals(
                        raw_text,
                        speech_rms=float(speech_rms),
                        accept_thr=0.025,
                        decode_conf=decode_conf,
                    )
                    if paired_tm:
                        print(
                            f"[reject] lang=english reason=too_many_words_for_segment "
                            f"words={words_n} max_words={max_words} chunk_s={duration_s:.2f} "
                            f"no_speech_prob={decode_conf.get('no_speech_prob')} "
                            f"avg_logprob={decode_conf.get('avg_logprob')} speech_rms={speech_rms:.5f} "
                            f"paired={';'.join(paired_tm)}"
                        )
                        raw_text = ""
                else:
                    print(
                        f"[reject] lang={forced_language} reason=too_many_words_for_segment "
                        f"words={words_n} max_words={max_words} chunk_s={duration_s:.2f}"
                    )
                    raw_text = ""
        if raw_text and words_n <= 3:
            nsp = decode_conf.get("no_speech_prob")
            alp = decode_conf.get("avg_logprob")
            comp = decode_conf.get("compression_ratio")
            reasonable_short = len((raw_text or "").strip()) >= 8 or words_n >= 2
            meta_miss_short = nsp is None and alp is None
            skip_cr_short_it = (
                forced_language == "italian"
                and meta_miss_short
                and speech_rms >= 0.035
                and reasonable_short
            )
            comp_weak = False
            if comp is not None:
                cv = float(comp)
                if forced_language == "italian":
                    comp_weak = cv >= float(_COMPRESSION_RATIO_THRESHOLD_IT) or (
                        cv > float(_COMPRESSION_RATIO_THRESHOLD) + 0.2 and not skip_cr_short_it
                    )
                else:
                    comp_weak = cv > float(_COMPRESSION_RATIO_THRESHOLD) + 0.2
            conf_weak = (
                (nsp is not None and float(nsp) > 0.55)
                or (alp is not None and float(alp) < -0.92)
                or comp_weak
            )
            conf_missing = meta_miss_short
            # Italian path can have missing confidence metadata; don't reject strong speech for that alone.
            reject_low_conf_short = conf_weak or (
                forced_language == "english" and conf_missing
            )
            if reject_low_conf_short:
                print(
                    f"[reject] lang={forced_language} reason=low_confidence_short "
                    f"text={raw_text[:80]!r} speech_rms={speech_rms:.5f} "
                    f"no_speech_prob={nsp} avg_logprob={alp} compression_ratio={comp}"
                )
                raw_text = ""
        if raw_text:
            trimmed_text, did_trim_tail = _trim_suspicious_tail_sentence(
                raw_text,
                audio_16k=arr,
                duration_s=float(duration_s),
                decode_conf=decode_conf,
            )
            if did_trim_tail:
                print(
                    f"[reject] lang={forced_language} reason=tail_continuation_trim "
                    f"text={raw_text[:80]!r} tail_low_energy={_tail_low_energy_ratio(arr):.2f}"
                )
                raw_text = trimmed_text
    if raw_text and _should_drop_repetition_hallucination(raw_text):
        raw_text = ""
    if raw_text and forced_language == "auto":
        raw_text = _strip_known_hallucinations(raw_text, rms)
    if raw_text and forced_language in {"english", "italian"} and _should_drop_forced_density_hallucination(
        raw_text, rms, duration_s
    ):
        raw_text = ""
    # Never restore Italian when decode produced text but filters cleared it (below RMS, confidence, tail trim, etc.).
    italian_cleared_after_decode = (
        forced_language == "italian"
        and bool((raw_text_original or "").strip())
        and not (raw_text or "").strip()
    )
    # Optional restore: defaults off — filtered clears must not be undone.
    if (
        _ASR_IT_RESTORE_ON_FILTER
        and forced_language == "italian"
        and not italian_cleared_after_decode
        and not raw_text
        and raw_text_original
        and rms >= 0.009
        and duration_s <= 4.5
    ):
        cand = _sanitize_forced_italian_output(raw_text_original)
        _, cand_trimmed = _trim_suspicious_tail_sentence(
            cand,
            audio_16k=arr,
            duration_s=float(duration_s),
            decode_conf=decode_conf,
        )
        if cand and not _should_drop_whisper_low_energy_hallucination(cand, rms, duration_s) and not cand_trimmed:
            raw_text = cand
    # Italian short-utterance recovery: if speech energy is present but decode is empty,
    # rerun once with a lower no-speech threshold to recover short greetings/commands.
    if (
        forced_language == "italian"
        and not raw_text
        and _ASR_IT_SHORT_RETRY
        and rms >= 0.018
        and duration_s <= 3.5
    ):
        retry_text, _retry_conf = _transcribe_chunked(
            arr,
            forced_language=forced_language,
            max_new_tokens=forced_max_new,
            num_beams=forced_num_beams,
            no_speech_threshold_override=min(_NO_SPEECH_THRESHOLD_IT, _ASR_IT_SHORT_RETRY_NS_THR),
        )
        retry_text = retry_text.strip()
        if retry_text:
            cand = _sanitize_forced_italian_output(retry_text)
            if cand and not _should_drop_repetition_hallucination(cand):
                raw_text = cand
    # English short-utterance recovery: if speech energy is present but decode is empty,
    # rerun once with a lower no-speech threshold to recover short greetings/commands.
    if (
        forced_language == "english"
        and not raw_text
        and _ASR_EN_SHORT_RETRY
        and rms >= 0.018
        and duration_s <= 3.5
    ):
        retry_text, _retry_conf = _transcribe_chunked(
            arr,
            forced_language=forced_language,
            max_new_tokens=forced_max_new,
            num_beams=forced_num_beams,
            no_speech_threshold_override=min(_NO_SPEECH_THRESHOLD_EN, _ASR_EN_SHORT_RETRY_NS_THR),
        )
        retry_text = retry_text.strip()
        if retry_text:
            cand = _sanitize_forced_english_output(retry_text)
            if cand and not _should_drop_repetition_hallucination(cand):
                raw_text = cand
    # Optional micro-fragment guard (off by default to avoid hard-coded word dropping).
    if _ASR_DROP_MICRO_FRAGMENTS and raw_text and forced_language in {"english", "italian"} and rms < 0.0070:
        w = [x for x in _WORD_RE.findall(raw_text)]
        if len(w) <= 2 and len("".join(ch for ch in raw_text if ch.isalpha())) <= 6:
            raw_text = ""
    # Generic forced-language tail suppression: drop weak short follow-up chunks
    # right after a stronger accepted chunk (streaming tail hallucination pattern).
    if raw_text and forced_language in {"english", "italian"}:
        with _last_text_lock:
            prev_sig = _last_signal_by_lang.get(forced_language, {"speech_rms": 0.0, "ts": 0.0})
        prev_speech_rms = float(prev_sig.get("speech_rms", 0.0))
        dt_s = time.time() - float(prev_sig.get("ts", 0.0))
        words_now = len(_WORD_RE.findall(raw_text))
        if (
            prev_speech_rms > 0.0
            and dt_s <= 2.5
            and words_now <= 3
            and float(speech_rms) < (prev_speech_rms * 0.6)
        ):
            print(
                f"[reject] reason=weak_tail_after_strong text={raw_text[:80]!r} "
                f"speech_rms={speech_rms:.5f} prev_speech_rms={prev_speech_rms:.5f} dt={dt_s:.2f}"
            )
            raw_text = ""
    # Ultra-short isolated chunk right after a completed phrase (thank-you / grazie tail pattern).
    if raw_text and forced_language in {"english", "italian"}:
        cand_u = raw_text.strip().upper()
        words_tail = len(_WORD_RE.findall(raw_text))
        with _last_text_lock:
            acc = _last_accepted_phrase_by_lang.get(forced_language)
        if acc and acc.get("text"):
            dt_tail = time.time() - float(acc.get("ts", 0.0))
            prev_txt = str(acc.get("text", ""))
            strong_meta = _decode_has_strong_confidence(decode_conf)
            if (
                float(duration_s) < float(_ASR_ULTRA_SHORT_TAIL_MAX_SEGMENT_S)
                and words_tail <= 2
                and dt_tail <= float(_ASR_ULTRA_SHORT_TAIL_WINDOW_S)
                and not _is_utterance_continuation(prev_txt, cand_u)
                and not strong_meta
            ):
                print(
                    f"[reject] reason=ultra_short_tail_after_phrase text={raw_text[:80]!r} "
                    f"segment_s={duration_s:.2f} dt={dt_tail:.2f}"
                )
                raw_text = ""
    # Short isolated generic closings (THANK YOU. / GRAZIE. / YOU.) after an accepted phrase — pattern, not substring strip.
    if raw_text and forced_language in {"english", "italian"}:
        words_ic = len(_WORD_RE.findall(raw_text))
        if words_ic <= 2 and _is_generic_tail_closing_phrase(raw_text, forced_language):
            cand_ic = raw_text.strip().upper()
            with _last_text_lock:
                acc_ic = _last_accepted_phrase_by_lang.get(forced_language)
            if acc_ic and str(acc_ic.get("text", "")).strip():
                dt_ic = time.time() - float(acc_ic.get("ts", 0.0))
                prev_ic = str(acc_ic.get("text", ""))
                if (
                    dt_ic <= float(_ASR_ULTRA_SHORT_TAIL_WINDOW_S)
                    and not _is_utterance_continuation(prev_ic, cand_ic)
                    and not _intentional_short_tail_evidence(decode_conf, speech_rms, duration_s)
                ):
                    print(
                        f"[reject] reason=isolated_tail_closing text={raw_text[:80]!r} "
                        f"segment_s={duration_s:.2f} dt={dt_ic:.2f} speech_rms={speech_rms:.5f} "
                        f"words={words_ic} lang={forced_language}"
                    )
                    raw_text = ""
    text = raw_text.upper().strip() if raw_text else ""
    lang_key = forced_language if forced_language in {"english", "italian"} else "auto"
    # Optional tiny-fragment suppression (off by default to avoid hard-coded word dropping).
    if _ASR_DROP_MICRO_FRAGMENTS and text:
        toks_now = [w for w in text.split() if w]
        if len(toks_now) <= 1 and len("".join(ch for ch in text if ch.isalpha())) <= 4:
            text = ""
    if text:
        with _last_text_lock:
            prev = _last_text_by_lang.get(lang_key, "")
            deduped = _remove_overlap_repetition(prev, text, max_overlap_words=_ASR_CROSS_CHUNK_OVERLAP_WORDS)
            if not _CPU_RELAX_NEAR_DUP and _is_near_duplicate_phrase(prev, deduped):
                forced_lock = lang_key in {"english", "italian"}
                if (not forced_lock) or _ASR_BLANK_NEAR_DUP_FORCED:
                    deduped = ""
            _last_text_by_lang[lang_key] = text
            _last_signal_by_lang[lang_key] = {"speech_rms": float(speech_rms), "ts": time.time()}
            if deduped and lang_key in {"english", "italian"}:
                _last_accepted_phrase_by_lang[lang_key] = {
                    "text": deduped,
                    "speech_rms": float(speech_rms),
                    "ts": time.time(),
                }
        text = deduped
    if text and forced_language in {"english", "italian"}:
        print(
            f"[accept] lang={forced_language} text={text[:80]!r} "
            f"speech_rms={speech_rms:.5f} chunk_s={duration_s:.2f}"
        )

    total_ms = (time.perf_counter() - t0) * 1000.0
    print(
        f"[timing] /audio lang={forced_language} sr={sr} chunk_s={duration_s:.2f} "
        f"full_rms={rms:.5f} speech_rms={speech_rms:.5f} "
        f"receive_ms={t_body_ms:.1f} preprocess_ms={t_pre_ms:.1f} asr_ms={t_asr_ms:.1f} total_ms={total_ms:.1f} chars={len(text)}"
    )
    print(f"[return] lang={forced_language} chars={len(text)} text={text[:80]!r}")
    print(
        f"[confidence] lang={forced_language} chunk_s={duration_s:.2f} speech_rms={speech_rms:.5f} "
        f"no_speech_prob={decode_conf.get('no_speech_prob')} avg_logprob={decode_conf.get('avg_logprob')} "
        f"compression_ratio={decode_conf.get('compression_ratio')}"
    )
    return Response(
        content=json.dumps(
            {
                "text": text,
                "infer_ms": round(float(t_asr_ms), 2),
                "total_ms": round(float(total_ms), 2),
                "chunk_duration_s": round(float(duration_s), 3),
                "speech_rms": round(float(speech_rms), 6),
                "no_speech_prob": decode_conf.get("no_speech_prob"),
                "avg_logprob": decode_conf.get("avg_logprob"),
                "compression_ratio": decode_conf.get("compression_ratio"),
            }
        ),
        media_type="application/json",
    )