fix(models): enforce Hinglish output contract

Author: Mr-Sunglasses

app/models/whisper_cpp.py

@@ -4,8 +4,9 @@
 import tempfile
 from pathlib import Path
 
+from app import scripts
 from app.catalog import CatalogModel
-from app.errors import EngineUnavailableError, TranscriptionProcessError
+from app.errors import EngineUnavailableError, LanguageUnsupportedError, TranscriptionProcessError
 from app.models.base import EngineHealth, TranscriptionOptions
 from app.models.warmup import prefetch_model_paths
 
@@ -35,16 +36,38 @@ async def transcribe(self, audio_path: Path, options: TranscriptionOptions) -> s
         await self._require_ready()
         with tempfile.TemporaryDirectory(prefix="vocagateway-transcript-") as temporary:
             output_stem = Path(temporary) / "result"
-            language = self.catalog_model.decoder_language_code if self.catalog_model else None
             arguments = _build_arguments(
                 self.binary,
                 self.model,
                 audio_path,
                 output_stem,
-                language or options.language,
+                self._decoder_language(options.language),
             )
             await _execute_whisper_cpp(arguments)
-            return _read_output_text(output_stem.with_suffix(".txt"))
+            transcript = _read_output_text(output_stem.with_suffix(".txt"))
+            self._require_fixed_output_script(transcript)
+            return transcript
+
+    def _decoder_language(self, requested: str) -> str:
+        model = self.catalog_model
+        if model is None or model.decoder_language_code is None:
+            return requested
+        if requested != "auto" and requested not in model.language_codes:
+            supported = ", ".join(model.language_codes)
+            raise LanguageUnsupportedError(
+                f"The selected model supports only {supported}; choose that output mode or Auto."
+            )
+        return model.decoder_language_code
+
+    def _require_fixed_output_script(self, transcript: str) -> None:
+        model = self.catalog_model
+        if model is None or model.decoder_language_code is None or len(model.language_codes) != 1:
+            return
+        output_language = model.language_codes[0]
+        if not scripts.transcript_matches_language(transcript, output_language):
+            raise LanguageUnsupportedError(
+                f"The model did not produce the required {output_language} writing system."
+            )
 
     async def _require_ready(self) -> None:
         health = await self.health()

app/routes/transcriptions.py

@@ -13,7 +13,7 @@
 
 from app import audio, context, errors, schemas
 
-_LANGUAGE_PATTERN = re.compile(r"^[A-Za-z_-]+$|^auto$")
+_LANGUAGE_PATTERN = re.compile(r"^(?:[A-Za-z-]+|hinglish_roman)$")
 _TRUTHY = frozenset(("true", "1", "yes"))
 _MAX_LANGUAGE_LENGTH = 20
 _READ_CHUNK_BYTES = 65_536
@@ -129,7 +129,7 @@ def language(cls, language: str | None) -> str:
             raise errors.APIProblem(
                 status.HTTP_422_UNPROCESSABLE_CONTENT,
                 "invalid_language",
-                "Language must be auto or a language tag.",
+                "Language must be auto, a language tag, or hinglish_roman.",
             )
         return language_value
 

app/schemas.py

@@ -22,7 +22,7 @@ class CreateSessionRequest(BaseModel):
     language: str = Field(
         default=AUTO_ENGINE,
         max_length=MAXIMUM_LANGUAGE_TAG_LENGTH,
-        pattern=r"^[A-Za-z_-]+$|^auto$",
+        pattern=r"^(?:[A-Za-z-]+|hinglish_roman)$",
     )
     style: Literal[
         "raw",

app/scripts.py

@@ -176,6 +176,10 @@
 # code-switching keeps far more of the base script than that — Hinglish rarely
 # drops below a third — so this sits well below any real transcript.
 _MINIMUM_EXPECTED_SHARE = 0.15
+# A fixed-script output mode is a stronger promise than an ordinary spoken
+# language. Code-switching is valid Hindi, but any non-Latin letter means the
+# Roman Hinglish contract was not met.
+_STRICT_SCRIPT_LANGUAGES = frozenset(("hinglish_roman",))
 
 
 def _script_of(character: str) -> str:
@@ -202,6 +206,8 @@ def transcript_matches_language(text: str, language: str) -> bool:
     with no letters at all. Otherwise the expected script has to account for at
     least `_MINIMUM_EXPECTED_SHARE` of the letters, which admits code-switching
     while rejecting text transliterated into another writing system entirely.
+    Fixed-script output contracts require every letter to use their declared
+    script.
     """
     if language == "auto":
         return True
@@ -212,4 +218,5 @@ def transcript_matches_language(text: str, language: str) -> bool:
     if not letters:
         return True
     matching = sum(1 for character in letters if _script_of(character) in expected)
-    return matching / len(letters) >= _MINIMUM_EXPECTED_SHARE
+    minimum_share = 1.0 if language.lower() in _STRICT_SCRIPT_LANGUAGES else _MINIMUM_EXPECTED_SHARE
+    return matching / len(letters) >= minimum_share

tests/test_api.py

@@ -58,6 +58,9 @@ def test_session_schema_accepts_roman_hinglish_output_contract() -> None:
     request = CreateSessionRequest(client_session_id=uuid4(), language="hinglish_roman")
     assert request.language == "hinglish_roman"
 
+    with pytest.raises(ValueError):
+        CreateSessionRequest(client_session_id=uuid4(), language="not_a_language")
+
 
 async def test_unsupported_language_is_reported_a_d68a4(
     settings: Settings, authorization: dict[str, str], audio_bytes: bytes

tests/test_scripts.py

@@ -58,6 +58,7 @@ def test_uncertain_cases_pass_rather_than_fail() -> None:
 def test_roman_hinglish_uses_the_latin_script_contract() -> None:
     assert transcript_matches_language("Aaj mujhe office jaana hai", "hinglish_roman") is True
     assert transcript_matches_language("आज मुझे जाना है", "hinglish_roman") is False
+    assert transcript_matches_language("Aaj office में meeting hai", "hinglish_roman") is False
 
 
 @pytest.mark.parametrize(

tests/test_transcriptions.py

@@ -53,6 +53,7 @@ async def transcribe(self, audio_path: Path, options: TranscriptionOptions) -> s
 
 def test_audio_form_accepts_roman_hinglish_output_contract() -> None:
     assert _AudioForm.language("hinglish_roman") == "hinglish_roman"
+    assert _AudioForm._invalid_language("not_a_language") is True
 
 
 async def test_transcriptions_require_a_bearer_token(

tests/test_whisper_cpp.py

@@ -5,7 +5,7 @@
 import pytest
 
 from app.catalog import CatalogModel
-from app.errors import EngineUnavailableError, TranscriptionProcessError
+from app.errors import EngineUnavailableError, LanguageUnsupportedError, TranscriptionProcessError
 from app.models.base import TranscriptionOptions
 from app.models.whisper_cpp import WhisperCppEngine
 
@@ -124,13 +124,53 @@ async def test_transcribe_uses_catalog_decoder_language_for_output_contract(
         decoder_language_code="hi",
     )
 
-    await WhisperCppEngine(binary, model, catalog_model).transcribe(
-        audio, TranscriptionOptions("hinglish_roman", RAW_STYLE)
-    )
+    engine = WhisperCppEngine(binary, model, catalog_model)
+    await engine.transcribe(audio, TranscriptionOptions("hinglish_roman", RAW_STYLE))
 
     recorded = (tmp_path / "whisper-cli.args").read_text(encoding="utf-8").splitlines()
     assert recorded[recorded.index("-l") + 1] == "hi"
 
+    with pytest.raises(LanguageUnsupportedError, match="only hinglish_roman"):
+        await engine.transcribe(audio, TranscriptionOptions("en", RAW_STYLE))
+
+
+async def test_fixed_output_contract_rejects_devanagari_leakage(tmp_path: Path) -> None:
+    binary = tmp_path / WHISPER_BINARY_NAME
+    _write_binary(
+        binary,
+        r"""#!/bin/sh
+of=""
+while [ "$#" -gt 0 ]; do
+  case "$1" in
+    -of) of="$2"; shift 2 ;;
+    *) shift ;;
+  esac
+done
+printf '%s' "Aaj office में meeting hai" > "$of.txt"
+""",
+    )
+    model = tmp_path / MODEL_FILE_NAME
+    model.write_bytes(MODEL_BYTES)
+    audio = tmp_path / AUDIO_FILE_NAME
+    audio.write_bytes(AUDIO_BYTES)
+    catalog_model = CatalogModel(
+        id="whisper.cpp:hinglish",
+        engine="whisper.cpp",
+        key=MODEL_FILE_NAME,
+        label="Hinglish",
+        size_bytes=1,
+        languages="Hindi + English, Roman script",
+        quality="Experimental",
+        minimum_ram_gb=4,
+        language_codes=("hinglish_roman",),
+        decoder_language_code="hi",
+    )
+
+    with pytest.raises(LanguageUnsupportedError, match="required hinglish_roman"):
+        await WhisperCppEngine(binary, model, catalog_model).transcribe(
+            audio, TranscriptionOptions("auto", RAW_STYLE)
+        )
+
 
 async def test_transcribe_raises_when_the_engine_aaaa(tmp_path: Path) -> None:
     engine = WhisperCppEngine(tmp_path / "missing-cli", tmp_path / "missing-model.bin")