diff --git a/Frontend/client/src/i18n/translations/de/settings.ts b/Frontend/client/src/i18n/translations/de/settings.ts
index 406f7ad0..a7ade41c 100644
--- a/Frontend/client/src/i18n/translations/de/settings.ts
+++ b/Frontend/client/src/i18n/translations/de/settings.ts
@@ -1,6 +1,12 @@
 import type { TranslationCatalog } from "@/i18n/types";
 
 export const settingsTranslations = {
+  "Live partials and completed turns · up to 60 minutes · Meta Model API key":
+    "Live-Zwischenstände und abgeschlossene Äußerungen · bis 60 Minuten · Meta-Model-API-Schlüssel",
+  "Transcribes after stop · files up to 10 minutes · Meta Model API key":
+    "Transkription nach dem Stoppen · Dateien bis 10 Minuten · Meta-Model-API-Schlüssel",
+  "Native speaker labels and approximate turn timestamps. Maximum 10 minutes per recording.":
+    "Native Sprechertrennung und ungefähre Zeitstempel je Äußerung. Maximal 10 Minuten pro Aufnahme.",
   // Languages and shared settings states.
   "Auto-detect": "Automatisch erkennen",
   German: "Deutsch",
diff --git a/Frontend/client/src/lib/meta-transcription-settings.test.ts b/Frontend/client/src/lib/meta-transcription-settings.test.ts
new file mode 100644
index 00000000..ae9ed301
--- /dev/null
+++ b/Frontend/client/src/lib/meta-transcription-settings.test.ts
@@ -0,0 +1,44 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { settingsTranslations } from "@/i18n/translations/de/settings";
+import {
+  META_CREDENTIAL_REQUIREMENT,
+  META_MEETING_FINAL_STT_OPTION,
+  META_TRANSCRIPTION_OPTIONS,
+  metaFrontendModelForService,
+  metaSettingsPatchForModel,
+} from "@/lib/meta-transcription-settings";
+
+test("Meta offers separate realtime and async routes for one published model", () => {
+  assert.deepEqual(
+    META_TRANSCRIPTION_OPTIONS.map((option) => option.group),
+    ["cloud_streaming", "cloud_async"],
+  );
+  for (const option of META_TRANSCRIPTION_OPTIONS) {
+    assert.equal(option.model, "muse-voice-transcribe-1.0");
+    assert.equal(option.usdPerHour, 0.18);
+    assert.equal(metaFrontendModelForService(option.service), option.value);
+    assert.deepEqual(metaSettingsPatchForModel(option.value), { defaultSttService: option.service });
+    assert.ok(settingsTranslations[option.routeNote]);
+    assert.equal("wordErrorRatePercent" in option, false);
+  }
+  assert.equal(metaSettingsPatchForModel("meta-unknown"), null);
+  assert.equal(metaFrontendModelForService("meta"), null);
+});
+
+test("both Meta STT routes reuse the existing Meta Model API credential", () => {
+  assert.deepEqual(META_CREDENTIAL_REQUIREMENT, {
+    provider: "Meta Model API",
+    label: "Meta Model API key",
+    helpKey: "meta",
+  });
+});
+
+test("Meta Meeting option does not promise five-hour or word-level support", () => {
+  assert.equal(META_MEETING_FINAL_STT_OPTION.value, "meta_stt_async");
+  assert.equal(META_MEETING_FINAL_STT_OPTION.nativeDiarization, true);
+  assert.equal(META_MEETING_FINAL_STT_OPTION.fiveHourSupported, false);
+  assert.match(META_MEETING_FINAL_STT_OPTION.detail, /10 minutes/);
+  assert.ok(settingsTranslations[META_MEETING_FINAL_STT_OPTION.detail]);
+});
diff --git a/Frontend/client/src/lib/meta-transcription-settings.ts b/Frontend/client/src/lib/meta-transcription-settings.ts
new file mode 100644
index 00000000..c9091762
--- /dev/null
+++ b/Frontend/client/src/lib/meta-transcription-settings.ts
@@ -0,0 +1,49 @@
+// Both transports use the same published model; Async is not a second model.
+export const META_TRANSCRIPTION_OPTIONS = [
+  {
+    value: "meta-realtime",
+    service: "meta_stt",
+    label: "Meta Muse Voice Transcribe Realtime",
+    model: "muse-voice-transcribe-1.0",
+    group: "cloud_streaming",
+    icon: "meta",
+    usdPerHour: 0.18,
+    routeNote: "Live partials and completed turns · up to 60 minutes · Meta Model API key",
+  },
+  {
+    value: "meta-async",
+    service: "meta_stt_async",
+    label: "Meta Muse Voice Transcribe Async",
+    model: "muse-voice-transcribe-1.0",
+    group: "cloud_async",
+    icon: "meta",
+    usdPerHour: 0.18,
+    routeNote: "Transcribes after stop · files up to 10 minutes · Meta Model API key",
+  },
+] as const;
+
+export const META_CREDENTIAL_REQUIREMENT = {
+  provider: "Meta Model API",
+  label: "Meta Model API key",
+  helpKey: "meta",
+} as const;
+
+export const META_MEETING_FINAL_STT_OPTION = {
+  value: "meta_stt_async",
+  label: "Meta Muse Voice Transcribe",
+  model: "muse-voice-transcribe-1.0",
+  credentialModel: "meta-async",
+  recommended: false,
+  nativeDiarization: true,
+  fiveHourSupported: false,
+  detail: "Native speaker labels and approximate turn timestamps. Maximum 10 minutes per recording.",
+} as const;
+
+export function metaFrontendModelForService(service: string) {
+  return META_TRANSCRIPTION_OPTIONS.find((option) => option.service === service)?.value ?? null;
+}
+
+export function metaSettingsPatchForModel(value: string) {
+  const option = META_TRANSCRIPTION_OPTIONS.find((candidate) => candidate.value === value);
+  return option ? { defaultSttService: option.service } : null;
+}
diff --git a/Frontend/client/src/pages/Settings.tsx b/Frontend/client/src/pages/Settings.tsx
index 111376d3..a08aebcf 100644
--- a/Frontend/client/src/pages/Settings.tsx
+++ b/Frontend/client/src/pages/Settings.tsx
@@ -150,6 +150,13 @@ import {
   geminiFrontendModelForService,
   geminiSettingsPatchForModel,
 } from "@/lib/gemini-transcription-settings";
+import {
+  META_TRANSCRIPTION_OPTIONS,
+  META_CREDENTIAL_REQUIREMENT,
+  META_MEETING_FINAL_STT_OPTION,
+  metaFrontendModelForService,
+  metaSettingsPatchForModel,
+} from "@/lib/meta-transcription-settings";
 
 type Translate = ReturnType<typeof useI18n>["t"];
 type FormatDate = ReturnType<typeof useI18n>["formatDate"];
@@ -255,6 +262,7 @@ const TRANSCRIPTION_MODEL_OPTIONS = [
   { value: "soniox-async", label: "Soniox Async" },
   { value: "modulate-realtime", label: "Modulate.AI Multilingual Realtime" },
   { value: "modulate-async", label: "Modulate.AI Multilingual Batch" },
+  ...META_TRANSCRIPTION_OPTIONS.map(({ value, label }) => ({ value, label })),
   {
     value: GEMINI_REALTIME_TRANSCRIPTION_OPTION.value,
     label: GEMINI_REALTIME_TRANSCRIPTION_OPTION.label,
@@ -892,6 +900,13 @@ function createProviderModelOptions(
 
   return [
     benchmarkOption("elevenlabs", "ElevenLabs Live", 6.5, 3.6, "cloud_streaming", "elevenlabs"),
+    ...META_TRANSCRIPTION_OPTIONS.map((option) => ({
+      ...option,
+      model: providerModels[option.value] || option.model,
+      detail: `${formatEstimatedEuroFromUsd(option.usdPerHour, localeTag)}/h`,
+      routeNote: t(option.routeNote),
+      hourlyCostEur: option.usdPerHour * USD_TO_EUR_FOR_ESTIMATES,
+    })),
     benchmarkOption(
       GEMINI_REALTIME_TRANSCRIPTION_OPTION.value,
       GEMINI_REALTIME_TRANSCRIPTION_OPTION.label,
@@ -974,6 +989,7 @@ function createProviderModelOptions(
 }
 
 const MEETING_FINAL_STT_OPTIONS = [
+  META_MEETING_FINAL_STT_OPTION,
   {
     value: "soniox_async",
     label: "Soniox Async",
@@ -2556,6 +2572,9 @@ export default function Settings() {
 
   const requiredCredentialForTranscriptionModel = (model: string): CredentialRequirement | null => {
     switch (model) {
+      case "meta-realtime":
+      case "meta-async":
+        return META_CREDENTIAL_REQUIREMENT;
       case "soniox-realtime":
       case "soniox-async":
         return { provider: "Soniox", label: "Soniox API key", helpKey: "soniox" };
@@ -2760,6 +2779,10 @@ export default function Settings() {
       if (service === "modulate" || service === "modulate_async") {
         return service === "modulate_async" ? "modulate-async" : "modulate-realtime";
       }
+      const metaModel = metaFrontendModelForService(service);
+      if (metaModel) {
+        return metaModel;
+      }
       if (service === "mistral" || service === "mistral_async") {
         return service === "mistral_async" ? "mistral-async" : "mistral-realtime";
       }
@@ -3320,6 +3343,7 @@ export default function Settings() {
     }
     const previousValue = transcriptionModel;
     const geminiSettingsPatch = geminiSettingsPatchForModel(value);
+    const metaSettingsPatch = metaSettingsPatchForModel(value);
     setTranscriptionModel(value);
     try {
       if (value === "soniox-async") {
@@ -3330,6 +3354,8 @@ export default function Settings() {
         await updateSettings({ defaultSttService: "modulate_async" });
       } else if (value === "modulate-realtime") {
         await updateSettings({ defaultSttService: "modulate" });
+      } else if (metaSettingsPatch) {
+        await updateSettings(metaSettingsPatch);
       } else if (geminiSettingsPatch) {
         await updateSettings(geminiSettingsPatch);
       } else if (value === "mistral-async") {
diff --git a/README.md b/README.md
index e70c81df..ea2ca9bf 100644
--- a/README.md
+++ b/README.md
@@ -213,6 +213,48 @@ fee, not a markup on the model's inference price.
 [See the OpenRouter model page.](https://openrouter.ai/microsoft/mai-transcribe-1.5)
 [See the OpenRouter fee details.](https://openrouter.ai/docs/faq)
 
+### Meta Muse Voice Transcribe
+
+Choose **Meta Muse Voice Transcribe Realtime** for live microphone transcription,
+or **Meta Muse Voice Transcribe Async** to transcribe after stopping. Both use
+`muse-voice-transcribe-1.0`, released by Meta on September 1, 2026, and the
+existing **Meta Model API** credential in Settings (`MODEL_API_KEY`). Model API
+access must be enabled for your Meta project; a Muse Spark entitlement alone
+does not prove Voice access.
+
+Realtime displays replaceable partials and injects only completed speech turns.
+It does not run local VAD or speaker detection. Async buffers one recording and
+submits one HTTP upload; Meta documents no separate async model or job-polling
+endpoint. File, YouTube and Meeting final transcription use that same upload
+route, with native speaker labels when requested.
+
+Limits: 60 minutes per realtime session; 10 minutes per upload. Uploads must fit
+32 MB including multipart overhead. Scriber prepares mono PCM16 WAV at 16 kHz
+when necessary and also accepts verified 24 kHz mono PCM16 WAV unchanged.
+There are 25 documented languages including German; use automatic language
+detection or a supported language hint. Custom vocabulary becomes Meta's
+`keywords`. Turn timestamps are approximate processed-audio boundaries, not
+word timestamps. The published rate is $0.18 per audio hour for both endpoints.
+
+Optional environment configuration (the UI stores credentials through the
+existing secret configuration path):
+
+```dotenv
+MODEL_API_KEY=your-meta-model-api-key
+SCRIBER_DEFAULT_STT=meta_stt
+# Use meta_stt_async above for transcription after stop.
+SCRIBER_META_STT_MODEL=muse-voice-transcribe-1.0
+```
+
+Unknown model IDs fail closed until their audio contract is verified. Cancellation
+never submits buffered audio. Failed uploads and interrupted streams are not
+automatically replayed. No local model download or new ML dependency is needed.
+
+[Meta announcement](https://research.meta.ai/blog/introducing-muse-voice-transcribe),
+[Voice guide and limits](https://dev.meta.ai/docs/speech-to-text),
+[HTTP API](https://dev.meta.ai/docs/api-reference/voice/transcribe),
+[WebSocket protocol](https://dev.meta.ai/docs/api-reference/voice/realtime).
+
 ### Modulate.AI multilingual transcription
 
 Modulate.AI is available for multilingual batch and realtime transcription.
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index 51593778..4a9df46b 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -7,6 +7,44 @@ architecture notes and should be updated when ownership boundaries change.
 
 ## Runtime Overview
 
+### Meta Muse Voice Transcribe (verified 2026-09-02)
+
+`src/meta_stt.py` owns the two Voice API contracts for
+`muse-voice-transcribe-1.0`. `meta_stt` authenticates the WebSocket in its first
+JSON frame, sends paced raw mono PCM16, and uses provider-native ENDPOINTING.
+Interim hypotheses are replaceable `InterimTranscriptionFrame` previews, never
+durable text. Completed turns are keyed by `turnId`, emitted in speech-start
+order, and deduplicated. Neither `transcript.final` nor `speechEnd` commits a
+turn. Stop drains admitted audio, sends `endStream`, and waits for close 1000;
+errors or unfinished turns cannot become successful completion. The receiver
+belongs to an `AsyncTaskSupervisor`, and the application retains ownership of
+the shared HTTP session.
+Connection setup, audio sends and socket closure share one lock. Audio waits
+for handshake acknowledgement and rechecks shutdown state after waiting for
+startup or pacing; cancellation cannot leak a socket or send queued audio.
+
+`meta_stt_async` buffers one recording in the existing disk-backed PCM spool
+and submits one multipart HTTP request on normal stop. Cancel discards it.
+The same HTTP path serves File, YouTube and Meeting finals. It uses buffered
+JSON (`Accept: application/json`), not SSE or a speculative job-polling API.
+Both routes use `MODEL_API_KEY`; no model weights or additional dependencies
+are required. The immutable route binds the model, vocabulary, language,
+endpoint identity and exact audio-format capability. WAV pass-through also
+checks sample rate and channel count. `DIARIZATION` is requested only for
+speaker-enabled file/Meeting results; normal live dictation has no speakers.
+Native turns remain `provider_segment` evidence and never become fabricated
+word timestamps. Uploads are limited to ten minutes and 32 MB including
+multipart overhead; realtime sessions are limited to one hour. There is no
+automatic replay after failure, no session resumption and no automatic
+long-recording splitting.
+
+Protocol sources: [guide](https://dev.meta.ai/docs/speech-to-text),
+[upload](https://dev.meta.ai/docs/api-reference/voice/transcribe),
+[realtime](https://dev.meta.ai/docs/api-reference/voice/realtime),
+[schemas](https://dev.meta.ai/docs/api-reference/voice/schemas).
+
+### Application components
+
 Scriber is a hybrid desktop app:
 
 - Tauri 2 shell for installed Windows desktop runtime.
diff --git a/docs/ROADMAP_AND_KNOWN_ISSUES.md b/docs/ROADMAP_AND_KNOWN_ISSUES.md
index ffbdb3c0..5af7744d 100644
--- a/docs/ROADMAP_AND_KNOWN_ISSUES.md
+++ b/docs/ROADMAP_AND_KNOWN_ISSUES.md
@@ -7,6 +7,15 @@ It tracks current status only.
 
 ## Recently Completed
 
+Meta Muse Voice Transcribe integration (2026-09-02): separate Realtime and Async
+choices use the same `muse-voice-transcribe-1.0` model and Meta Model API key.
+Async means one non-blocking HTTP upload, not a separate Meta model or hosted
+background job. Current constraints: ten minutes per file/Meeting final,
+60 minutes per realtime session, no automatic splitting/resume, and no word
+timestamps. HTTP SSE is available from Meta but not used by Scriber; realtime
+previews use the WebSocket endpoint. Protocol tests are local; production
+account access and installed microphone smoke still require verification.
+
 Desktop runtime:
 
 - Tauri is the primary Windows desktop runtime.
diff --git a/docs/TESTING_AND_RELEASE.md b/docs/TESTING_AND_RELEASE.md
index 5c6c4630..29b437be 100644
--- a/docs/TESTING_AND_RELEASE.md
+++ b/docs/TESTING_AND_RELEASE.md
@@ -7,6 +7,19 @@ notes.
 
 ## Core Test Commands
 
+Muse Voice Transcribe regression coverage is in `tests/test_meta_stt.py`.
+It exercises real local aiohttp HTTP/WebSocket servers: multipart settings and
+auth, PCM framing, cumulative previews, overlapping turn completion, endStream
+drain, authentication rejection, disconnects, timeout, privacy-safe errors,
+no retry, cancel-without-upload, model/language validation and upload limits.
+Startup-race tests initialize the real Pipecat processor and use server-side
+barriers at both WebSocket upgrade and handshake acknowledgement. They check
+waiting audio and cancellation before client-session cleanup can hide leaks.
+Run it together with audio-preparation, route-artifact, config and pipeline-stop
+tests. Live Meta validation requires an explicitly configured `MODEL_API_KEY`
+with Voice access; local protocol tests do not prove account entitlement or
+recognition quality. No installer was built as part of this provider change.
+
 Run from repository root unless specified.
 
 Python (always through Scriber's project environment, never bare `python`):
diff --git a/src/api/upload_policy.py b/src/api/upload_policy.py
index 0a5509ce..59cb8a01 100644
--- a/src/api/upload_policy.py
+++ b/src/api/upload_policy.py
@@ -62,6 +62,8 @@
     "openrouter_stt": (300 * 1024 * 1024, "300MB"),
     "modulate": (100 * 1024 * 1024, "100MB"),
     "modulate_async": (100 * 1024 * 1024, "100MB"),
+    "meta_stt": (32_000_000 - 65_536, "32MB including multipart overhead"),
+    "meta_stt_async": (32_000_000 - 65_536, "32MB including multipart overhead"),
 }
 
 
diff --git a/src/audio_prepare.py b/src/audio_prepare.py
index 3802932a..80c10c2e 100644
--- a/src/audio_prepare.py
+++ b/src/audio_prepare.py
@@ -301,6 +301,11 @@ def resolve_provider_audio_selection(
         else None
     )
     original = probe.audio_format if effective_limit is None or probe.byte_length <= effective_limit else None
+    if provider in {"meta_stt", "meta_stt_async"}:
+        if probe.duration_ms is not None and probe.duration_ms > 600_000:
+            raise ProviderAudioPreparationError("Meta STT recordings must not exceed 10 minutes.")
+        if probe.channels != 1 or probe.sample_rate not in (16_000, 24_000):
+            original = None
     selection = select_audio_input_format(
         capability,
         route_kind=ProviderAudioRouteKind.BATCH,
@@ -386,6 +391,12 @@ async def prepare_provider_audio_file(
             and frozen_selection.audio_format != probe.audio_format
         ):
             raise ProviderAudioPreparationError("Frozen pass-through format does not match the probed source.")
+        if (
+            provider in {"meta_stt", "meta_stt_async"}
+            and frozen_selection.mode == AudioSelectionMode.ORIGINAL_PASSTHROUGH
+            and (probe.channels != 1 or probe.sample_rate not in (16_000, 24_000))
+        ):
+            raise ProviderAudioPreparationError("Frozen Meta WAV must be mono PCM16 at 16 or 24 kHz.")
         selected = frozen_selection
 
     generated = selected.mode != AudioSelectionMode.ORIGINAL_PASSTHROUGH
diff --git a/src/config.py b/src/config.py
index 7c4bf035..85ef8c81 100644
--- a/src/config.py
+++ b/src/config.py
@@ -295,6 +295,8 @@ class Config:
     DEFAULT_OPENROUTER_STT_MODEL = "microsoft/mai-transcribe-1.5"
     DEFAULT_GEMINI_STT_MODEL = "gemini-3.5-transcribe"
     DEFAULT_GEMINI_REALTIME_STT_MODEL = "gemini-3.5-transcribe-live"
+    DEFAULT_META_STT_MODEL = "muse-voice-transcribe-1.0"
+    META_STT_MODEL = os.getenv("SCRIBER_META_STT_MODEL", DEFAULT_META_STT_MODEL)
     _LEGACY_DEFAULT_GEMINI_STT_MODELS: ClassVar[set[str]] = {"gemini-2.5-flash"}
 
     # API Keys
@@ -409,6 +411,8 @@ class Config:
         "speechmatics_async": "SPEECHMATICS_API_KEY",
         "modulate": "MODULATE_API_KEY",
         "modulate_async": "MODULATE_API_KEY",
+        "meta_stt": "MODEL_API_KEY",
+        "meta_stt_async": "MODEL_API_KEY",
         "onnx_local": None,  # No API key needed for local models
     }
 
@@ -438,6 +442,8 @@ class Config:
         "speechmatics_async": "Speechmatics (Batch)",
         "modulate": "Modulate (Realtime Multilingual)",
         "modulate_async": "Modulate (Batch Multilingual)",
+        "meta_stt": "Meta Muse Voice Transcribe (Realtime)",
+        "meta_stt_async": "Meta Muse Voice Transcribe (Async)",
         "onnx_local": "Local (ONNX)",
     }
 
@@ -845,6 +851,8 @@ def configured(value: object, fallback: str) -> str:
             "soniox-async": configured(cls.SONIOX_ASYNC_MODEL, cls.DEFAULT_SONIOX_ASYNC_MODEL),
             "modulate-realtime": "velma-2-stt-streaming",
             "modulate-async": "velma-2-stt-batch",
+            "meta-realtime": cls.META_STT_MODEL,
+            "meta-async": cls.META_STT_MODEL,
             "gemini-stt": configured(cls.GEMINI_STT_MODEL, cls.DEFAULT_GEMINI_STT_MODEL),
             "gemini-realtime": configured(
                 cls.GEMINI_REALTIME_STT_MODEL,
@@ -1042,6 +1050,7 @@ def add(k, v):
         add("GROQ_API_KEY", cls.GROQ_API_KEY or "")
         add("SPEECHMATICS_API_KEY", cls.SPEECHMATICS_API_KEY or "")
         add("MODULATE_API_KEY", cls.MODULATE_API_KEY or "")
+        add("SCRIBER_META_STT_MODEL", cls.META_STT_MODEL)
 
         add("SCRIBER_HOTKEY", cls.HOTKEY)
         add("SCRIBER_POST_PROCESSING_HOTKEY", cls.POST_PROCESSING_HOTKEY)
diff --git a/src/core/provider_audio_formats.py b/src/core/provider_audio_formats.py
index dcc0b899..df6bcf54 100644
--- a/src/core/provider_audio_formats.py
+++ b/src/core/provider_audio_formats.py
@@ -235,6 +235,7 @@ def _capability(
     max_upload_bytes: int | None = None,
     evidence_kind: CapabilityEvidenceKind = (CapabilityEvidenceKind.OFFICIAL_ENDPOINT_DOCS),
     evidence_reference: str,
+    verified_at: date = CAPABILITY_VERIFIED_AT,
     active: bool = True,
 ) -> ProviderAudioInputCapabilities:
     return ProviderAudioInputCapabilities(
@@ -252,7 +253,7 @@ def _capability(
         max_upload_bytes=max_upload_bytes,
         evidence_kind=evidence_kind,
         evidence_reference=evidence_reference,
-        verified_at=CAPABILITY_VERIFIED_AT,
+        verified_at=verified_at,
         capability_id=f"{provider}:{route}:{model_family}",
         revision=CAPABILITY_REVISION,
         active=active,
@@ -655,6 +656,31 @@ def _capability(
         )
         for provider in ("modulate", "modulate_async")
     ),
+    *(
+        _capability(
+            provider,
+            "asr_transcribe",
+            "muse-voice-transcribe-1.0",
+            ProviderAudioRouteKind.BATCH,
+            batch_formats=(AudioInputFormat.WAV_PCM16,),
+            direct_passthrough_formats=(AudioInputFormat.WAV_PCM16,),
+            preferred_lossless_format=AudioInputFormat.WAV_PCM16,
+            max_upload_bytes=32_000_000 - 65_536,
+            evidence_reference="https://dev.meta.ai/docs/api-reference/voice/transcribe",
+            verified_at=date(2026, 9, 2),
+        )
+        for provider in ("meta_stt", "meta_stt_async")
+    ),
+    _capability(
+        "meta_stt",
+        "asr_realtime",
+        "muse-voice-transcribe-1.0",
+        ProviderAudioRouteKind.REALTIME,
+        realtime_formats=(AudioInputFormat.RAW_PCM16,),
+        preferred_lossless_format=AudioInputFormat.RAW_PCM16,
+        evidence_reference="https://dev.meta.ai/docs/api-reference/voice/realtime",
+        verified_at=date(2026, 9, 2),
+    ),
     _capability(
         "modulate_async",
         "velma_2_batch_english_vfast",
@@ -757,6 +783,8 @@ def _capability(
     "speechmatics_async": "batch_v2",
     "modulate": "velma_2_batch",
     "modulate_async": "velma_2_batch",
+    "meta_stt": "asr_transcribe",
+    "meta_stt_async": "asr_transcribe",
     "onnx_local": "decoded_pcm_local",
 }
 
@@ -770,6 +798,7 @@ def _capability(
     "gladia": "v2_live",
     "speechmatics": "realtime_v2",
     "modulate": "velma_2_streaming",
+    "meta_stt": "asr_realtime",
     "elevenlabs": "scribe_v2_realtime",
     "gemini_realtime": "live_transcription",
 }
@@ -1062,7 +1091,7 @@ def _validate_registry() -> None:
         ):
             if preferred is not None and preferred not in accepted:
                 raise RuntimeError(f"{capability.capability_id} has an unsupported preferred format.")
-        if capability.verified_at != CAPABILITY_VERIFIED_AT:
+        if not isinstance(capability.verified_at, date) or capability.verified_at < CAPABILITY_VERIFIED_AT:
             raise RuntimeError(f"{capability.capability_id} has an unexpected verification date.")
 
 
diff --git a/src/core/provider_capabilities.py b/src/core/provider_capabilities.py
index 3661901c..366f37c8 100644
--- a/src/core/provider_capabilities.py
+++ b/src/core/provider_capabilities.py
@@ -173,6 +173,24 @@ class ProviderCapabilities:
         supports_batch_diarization=True,
         supports_word_timestamps=True,
     ),
+    "meta_stt": ProviderCapabilities(
+        supports_live_streaming=True,
+        supports_direct_file_upload=True,
+        injects_immediately_in_live_mode=True,
+        supports_batch_diarization=True,
+        supports_word_timestamps=False,
+        supports_five_hour_meeting=False,
+        meeting_max_duration_seconds=600,
+    ),
+    "meta_stt_async": ProviderCapabilities(
+        supports_live_streaming=False,
+        supports_direct_file_upload=True,
+        injects_immediately_in_live_mode=False,
+        supports_batch_diarization=True,
+        supports_word_timestamps=False,
+        supports_five_hour_meeting=False,
+        meeting_max_duration_seconds=600,
+    ),
     "modulate": ProviderCapabilities(
         supports_live_streaming=True,
         supports_direct_file_upload=True,
diff --git a/src/meta_stt.py b/src/meta_stt.py
new file mode 100644
index 00000000..c61af6c4
--- /dev/null
+++ b/src/meta_stt.py
@@ -0,0 +1,504 @@
+"""Muse Voice Transcribe, using Meta's documented Voice API (2026-09-02).
+
+HTTP uploads and WebSocket sessions are distinct contracts. Live dictation uses
+ENDPOINTING without diarization; cumulative hypotheses are previews only. The
+session ends with endStream, followed by draining results through close 1000.
+No request is automatically replayed after audio may have reached Meta.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import io
+import json
+import time
+import wave
+from collections.abc import Callable
+from typing import Any, BinaryIO
+
+import aiohttp
+from pipecat.frames.frames import (
+    AudioRawFrame,
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    StopFrame,
+    TranscriptionFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.time import time_now_iso8601
+
+from src.core.provider_errors import provider_transport_error
+from src.runtime.audio_spool import append_pcm_frame, close_pcm_spool, create_pcm_spool, pcm_stream_to_wav
+from src.runtime.http_response import read_response_json_limited
+from src.runtime.task_supervisor import AsyncTaskSupervisor
+
+META_STT_MODEL = "muse-voice-transcribe-1.0"
+META_STT_BATCH_URL = "https://api.meta.ai/v1/asr/transcribe"
+META_STT_REALTIME_URL = "wss://api.meta.ai/v1/asr/realtime"
+# The 32 MB cap covers multipart overhead as well as the audio part.
+META_STT_MAX_AUDIO_BYTES = 32_000_000 - 65_536
+META_STT_MAX_SECONDS = 600
+META_STT_REALTIME_MAX_SECONDS = 3600
+_LANGUAGES = {
+    "ar": "Arabic",
+    "bn": "Bengali",
+    "nl": "Dutch",
+    "en": "English",
+    "fr": "French",
+    "de": "German",
+    "he": "Hebrew",
+    "hi": "Hindi",
+    "id": "Indonesian",
+    "it": "Italian",
+    "ja": "Japanese",
+    "kn": "Kannada",
+    "ko": "Korean",
+    "ms": "Malay",
+    "zh": "Mandarin Chinese",
+    "mr": "Marathi",
+    "pl": "Polish",
+    "pt": "Portuguese",
+    "es": "Spanish",
+    "tl": "Tagalog",
+    "fil": "Tagalog",
+    "ta": "Tamil",
+    "te": "Telugu",
+    "th": "Thai",
+    "tr": "Turkish",
+    "vi": "Vietnamese",
+}
+
+
+def meta_request_options(*, model: str, language: Any = None, custom_vocab: str = "") -> dict[str, Any]:
+    if model != META_STT_MODEL:
+        raise ValueError("Meta STT model is not verified. Select muse-voice-transcribe-1.0.")
+    options: dict[str, Any] = {"model": model}
+    raw = str(getattr(language, "value", language) or "").strip().replace("_", "-")
+    if raw and raw.lower() != "auto":
+        name = _LANGUAGES.get(raw.lower().split("-", 1)[0])
+        if name is None:
+            raise ValueError("Meta STT language is not supported. Select automatic language detection.")
+        options["languageBias"] = [name]
+    keywords = list(dict.fromkeys(term.strip() for term in custom_vocab.split(",") if term.strip()))
+    if keywords:
+        options["keywords"] = keywords
+    if len(json.dumps(options).encode("utf-8")) > 32_768:
+        raise ValueError("Meta STT vocabulary exceeds Scriber's 32 KB request-settings limit.")
+    return options
+
+
+def validate_meta_wav(source: BinaryIO) -> None:
+    """Validate without consuming or closing the caller-owned upload stream."""
+    position = source.tell()
+    try:
+        source.seek(0, 2)
+        if source.tell() > META_STT_MAX_AUDIO_BYTES:
+            raise ValueError("Meta STT audio exceeds the 32 MB multipart limit.")
+        source.seek(0)
+        with wave.open(source, "rb") as audio:
+            if audio.getnchannels() != 1 or audio.getsampwidth() != 2 or audio.getframerate() not in (16000, 24000):
+                raise ValueError("Meta STT requires mono PCM16 WAV at 16 or 24 kHz.")
+            if audio.getnframes() > META_STT_MAX_SECONDS * audio.getframerate():
+                raise ValueError("Meta STT recordings must not exceed 10 minutes.")
+            if audio.getnframes() == 0:
+                raise ValueError("Meta STT recording is empty.")
+            audio.setpos(audio.getnframes() - 1)
+            if len(audio.readframes(1)) != 2:
+                raise ValueError("Meta STT WAV data is truncated.")
+    except wave.Error, EOFError:
+        raise ValueError("Meta STT requires a valid PCM16 WAV recording.") from None
+    finally:
+        source.seek(position)
+
+
+def _final_payload(payload: Any) -> dict[str, Any]:
+    if not isinstance(payload, dict) or not isinstance(payload.get("transcript"), str):
+        raise RuntimeError("Meta STT returned an invalid final transcript.")
+    duration = payload.get("audioDurationMs")
+    turns = payload.get("turns")
+    if not isinstance(duration, int) or isinstance(duration, bool) or duration < 0 or not isinstance(turns, list):
+        raise RuntimeError("Meta STT returned invalid recording metadata.")
+    clean_turns = []
+    for turn in turns:
+        if not isinstance(turn, dict) or not isinstance(turn.get("transcript"), str):
+            raise RuntimeError("Meta STT returned an invalid turn.")
+        if any(type(turn.get(key)) is not int for key in ("turnId", "startMs", "endMs")):
+            raise RuntimeError("Meta STT returned invalid turn timing.")
+        if not 0 <= turn["startMs"] <= turn["endMs"] <= duration:
+            raise RuntimeError("Meta STT returned invalid turn timing.")
+        clean = {key: turn[key] for key in ("turnId", "startMs", "endMs", "transcript")}
+        if isinstance(turn.get("speaker"), str):
+            clean["speaker"] = turn["speaker"]
+        clean_turns.append(clean)
+    return {"transcript": payload["transcript"], "audioDurationMs": duration, "turns": clean_turns}
+
+
+async def transcribe_with_meta(
+    *,
+    session: aiohttp.ClientSession,
+    api_key: str,
+    audio_source: bytes | BinaryIO,
+    model: str = META_STT_MODEL,
+    language: Any = None,
+    custom_vocab: str = "",
+    mode: str = "ENDPOINTING",
+    timeout_secs: float = 600,
+    endpoint: str = META_STT_BATCH_URL,
+) -> dict[str, Any]:
+    if not api_key:
+        raise ValueError("Meta API key is missing. Add the Meta Model API key in Settings.")
+    if mode not in {"PUSH_TO_TALK", "ENDPOINTING", "DIARIZATION"}:
+        raise ValueError("Invalid Meta STT mode.")
+    options = meta_request_options(model=model, language=language, custom_vocab=custom_vocab)
+    options.update(audioEncoding="WAV", mode=mode)
+    source = io.BytesIO(audio_source) if isinstance(audio_source, bytes) else audio_source
+    validate_meta_wav(source)
+    source.seek(0)
+    data = aiohttp.FormData()
+    data.add_field("request", json.dumps(options), content_type="application/json")
+    data.add_field("audio", source, filename="audio.wav", content_type="audio/wav")
+    try:
+        async with session.post(
+            endpoint,
+            data=data,
+            headers={"Authorization": f"Bearer {api_key}", "Accept": "application/json"},
+            timeout=aiohttp.ClientTimeout(total=max(1, timeout_secs)),
+        ) as response:
+            if response.status != 200:
+                # Never include response bodies, credentials, or transcript echoes in errors.
+                raise provider_transport_error("meta_stt_async", "transcribe", status=response.status)
+            payload = await read_response_json_limited(response, 4 * 1024 * 1024)
+    except (aiohttp.ClientError, TimeoutError) as exc:
+        status = getattr(exc, "status", None)
+        raise provider_transport_error("meta_stt_async", "transcribe", status=status or None) from None
+    return _final_payload(payload)
+
+
+class MetaAsyncProcessor(FrameProcessor):
+    """One bounded recording, one asynchronous HTTP upload on normal stop."""
+
+    def __init__(
+        self,
+        *,
+        session: aiohttp.ClientSession,
+        api_key: str,
+        model: str = META_STT_MODEL,
+        language: Any = None,
+        custom_vocab: str = "",
+        on_progress: Callable[[str], None] | None = None,
+    ):
+        super().__init__()
+        self._session = session
+        self._api_key = api_key
+        self._options = dict(model=model, language=language, custom_vocab=custom_vocab)
+        meta_request_options(**self._options)
+        self._on_progress = on_progress
+        self._buffer = create_pcm_spool(reserve_wav_header=True)
+        self._size = 0
+        self._rate: int | None = None
+        self._failed = False
+        self._finished = False
+
+    async def cleanup(self):
+        close_pcm_spool(self._buffer)
+        await super().cleanup()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        try:
+            if isinstance(frame, AudioRawFrame) and not self._failed and not self._finished:
+                if frame.sample_rate not in (16000, 24000) or frame.num_channels != 1 or len(frame.audio) % 2:
+                    raise ValueError("Meta STT requires mono PCM16 at 16 or 24 kHz.")
+                if self._rate is not None and self._rate != frame.sample_rate:
+                    raise ValueError("Meta STT audio format changed within a recording.")
+                self._rate = frame.sample_rate
+                if self._size + len(frame.audio) > self._rate * 2 * META_STT_MAX_SECONDS:
+                    raise ValueError("Meta STT recordings must not exceed 10 minutes.")
+                self._size = await append_pcm_frame(self._buffer, self._size, frame.audio)
+            elif isinstance(frame, (EndFrame, StopFrame, CancelFrame)) and not self._finished:
+                self._finished = True
+                try:
+                    if (
+                        not isinstance(frame, CancelFrame)
+                        and not self._failed
+                        and self._size
+                        and self._rate is not None
+                        and not getattr(self, "_skip_terminal_transcription", False)
+                    ):
+                        if self._on_progress:
+                            self._on_progress("Transcribing...")
+                        wav = await asyncio.to_thread(
+                            pcm_stream_to_wav,
+                            self._buffer,
+                            self._rate,
+                            1,
+                            reserved_wav_header=True,
+                            pcm_size=self._size,
+                        )
+                        try:
+                            result = await transcribe_with_meta(
+                                session=self._session,
+                                api_key=self._api_key,
+                                audio_source=wav,
+                                mode="PUSH_TO_TALK",
+                                **self._options,
+                            )
+                        finally:
+                            wav.close()
+                        if result["transcript"].strip():
+                            await self.push_frame(
+                                TranscriptionFrame(
+                                    text=result["transcript"].strip(),
+                                    user_id="user",
+                                    timestamp=time_now_iso8601(),
+                                    finalized=True,
+                                    result=None,
+                                ),
+                                direction,
+                            )
+                finally:
+                    close_pcm_spool(self._buffer)
+        except asyncio.CancelledError:
+            close_pcm_spool(self._buffer)
+            raise
+        except Exception as exc:
+            self._failed = True
+            close_pcm_spool(self._buffer)
+            # Our own validation errors and transport errors are already content-free.
+            from src.core.provider_errors import ProviderTransportError
+
+            message = (
+                str(exc) if isinstance(exc, (ValueError, ProviderTransportError)) else "Meta STT transcription failed."
+            )
+            await self.push_frame(ErrorFrame(error=message), direction)
+        await self.push_frame(frame, direction)
+
+
+class MetaRealtimeSTTService(FrameProcessor):
+    """One authenticated socket per recording; only completed turns are final."""
+
+    def __init__(
+        self,
+        *,
+        session: aiohttp.ClientSession,
+        api_key: str,
+        model: str = META_STT_MODEL,
+        language: Any = None,
+        custom_vocab: str = "",
+        sample_rate: int = 16000,
+        channels: int = 1,
+        endpoint: str = META_STT_REALTIME_URL,
+    ):
+        super().__init__()
+        if sample_rate not in (16000, 24000) or channels != 1:
+            raise ValueError("Meta STT requires mono PCM16 at 16 or 24 kHz.")
+        self._options = meta_request_options(model=model, language=language, custom_vocab=custom_vocab)
+        self._session, self._api_key, self._endpoint = session, api_key, endpoint
+        self._rate = sample_rate
+        self._ws: aiohttp.ClientWebSocketResponse | None = None
+        self._tasks = AsyncTaskSupervisor(owner="meta_stt")
+        self._terminal = asyncio.Event()
+        # Start and audio processing may overlap. Both wait on the same
+        # authenticated connection; shutdown cannot orphan a pending handshake.
+        self._connect_lock = asyncio.Lock()
+        self._failed = False
+        self._ending = False
+        self._closed = False
+        self._started = False
+        self._audio_bytes = 0
+        self._audio_started: float | None = None
+        self._turn_order: list[int] = []
+        self._completed: dict[int, str] = {}
+        self._seen: set[int] = set()
+        self._final_timeout = 30.0
+
+    async def _error(self, message: str, direction: FrameDirection) -> None:
+        if not self._failed:
+            self._failed = True
+            await self.push_frame(ErrorFrame(error=f"Meta STT realtime: {message}"), direction)
+
+    async def _connect(self, direction: FrameDirection) -> None:
+        # _started means startup was attempted, not that the socket is ready.
+        # Do not read it as a fast path before acquiring the readiness barrier.
+        async with self._connect_lock:
+            if self._started or self._closed or self._ending:
+                return
+            await self._connect_locked(direction)
+
+    async def _connect_locked(self, direction: FrameDirection) -> None:
+        self._started = True
+        if not self._api_key:
+            await self._error("API key is missing.", direction)
+            return
+        try:
+            async with asyncio.timeout(30):
+                self._ws = await self._session.ws_connect(self._endpoint, heartbeat=None, max_msg_size=4 * 1024 * 1024)
+            async with asyncio.timeout(10):
+                await self._ws.send_json(
+                    {
+                        **self._options,
+                        "authorization": {"accessToken": f"Bearer {self._api_key}"},
+                        "audioEncoding": f"PCM_{self._rate // 1000}KHZ",
+                        "mode": "ENDPOINTING",
+                        "partialMode": "CUMULATIVE",
+                        "emitAudioProgress": False,
+                    }
+                )
+                ack = await self._ws.receive_json()
+            if (
+                not isinstance(ack, dict)
+                or "type" in ack
+                or not isinstance(ack.get("sessionId"), str)
+                or not ack["sessionId"]
+            ):
+                await self._error("handshake rejected; check Meta Model API access.", direction)
+                await self._ws.close()
+                return
+            self._tasks.spawn(self._receive(direction), name="meta_stt_receive")
+        except (aiohttp.ClientError, TimeoutError, ValueError) as exc:
+            await self._error(
+                str(provider_transport_error("meta_stt", "connect", status=getattr(exc, "status", None))), direction
+            )
+            if self._ws is not None:
+                await self._ws.close()
+
+    async def _handle_event(self, payload: Any, direction: FrameDirection) -> None:
+        if not isinstance(payload, dict):
+            raise ValueError("invalid event")
+        kind = payload.get("type")
+        if kind == "error":
+            await self._error("provider rejected the stream; check access, limits and audio format.", direction)
+        elif kind == "speechStart":
+            turn_id = payload.get("turnId")
+            if type(turn_id) is not int:
+                raise ValueError("invalid turn")
+            if turn_id not in self._seen and turn_id not in self._turn_order:
+                self._turn_order.append(turn_id)
+        elif kind == "transcript":
+            text = payload.get("transcript")
+            if not isinstance(text, str):
+                raise ValueError("invalid transcript")
+            # final:true is NOT turn completion in ENDPOINTING mode.
+            if text and not self._ending:
+                await self.push_frame(
+                    InterimTranscriptionFrame(text=text, user_id="user", timestamp=time_now_iso8601()), direction
+                )
+        elif kind == "speechComplete":
+            turn_id, text = payload.get("turnId"), payload.get("transcript")
+            if type(turn_id) is not int or not isinstance(text, str):
+                raise ValueError("invalid completion")
+            if turn_id in self._seen:
+                return
+            if turn_id not in self._turn_order:
+                raise ValueError("completion without speech start")
+            self._completed[turn_id] = text.strip()
+            while self._turn_order and self._turn_order[0] in self._completed:
+                completed_id = self._turn_order.pop(0)
+                final = self._completed.pop(completed_id)
+                self._seen.add(completed_id)
+                if final:
+                    await self.push_frame(
+                        TranscriptionFrame(
+                            text=final, user_id="user", timestamp=time_now_iso8601(), finalized=True, result=None
+                        ),
+                        direction,
+                    )
+        # speechEnd is a boundary, not final text. Unknown events are additive.
+
+    async def _receive(self, direction: FrameDirection) -> None:
+        try:
+            assert self._ws is not None
+            async for message in self._ws:
+                if message.type == aiohttp.WSMsgType.TEXT:
+                    await self._handle_event(json.loads(message.data), direction)
+                    if self._failed:
+                        break
+                elif message.type == aiohttp.WSMsgType.ERROR:
+                    break
+            if (
+                not self._closed
+                and not self._failed
+                and (self._ws.close_code != 1000 or not self._ending or self._turn_order)
+            ):
+                status = {1008: 400, 1011: 500, 1013: 429}.get(self._ws.close_code)
+                message = (
+                    str(provider_transport_error("meta_stt", "realtime", status=status))
+                    if status
+                    else "stream closed before all turns completed."
+                )
+                await self._error(message, direction)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            await self._error("invalid response or interrupted connection.", direction)
+        finally:
+            self._terminal.set()
+
+    async def _close(self, *, finalize: bool, direction: FrameDirection) -> None:
+        if self._closed:
+            return
+        self._ending = True
+        async with self._connect_lock:
+            await self._close_locked(finalize=finalize, direction=direction)
+
+    async def _close_locked(self, *, finalize: bool, direction: FrameDirection) -> None:
+        if self._closed:
+            return
+        try:
+            if finalize and not self._failed and self._ws is not None and not self._ws.closed:
+                await self._ws.send_json({"type": "endStream"})
+                try:
+                    await asyncio.wait_for(self._terminal.wait(), self._final_timeout)
+                except TimeoutError:
+                    await self._error("timed out waiting for final transcription.", direction)
+        finally:
+            self._closed = True
+            await self._tasks.close(timeout_seconds=2, cancel=True)
+            if self._ws is not None:
+                await self._ws.close()
+
+    async def cleanup(self):
+        await self._close(finalize=False, direction=FrameDirection.DOWNSTREAM)
+        await super().cleanup()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        try:
+            if isinstance(frame, StartFrame):
+                await self._connect(direction)
+            elif isinstance(frame, AudioRawFrame) and not self._ending and not self._failed:
+                await self._connect(direction)
+                # Cancel/cleanup can begin while this frame waits for startup.
+                if self._ending or self._closed or self._failed:
+                    await self.push_frame(frame, direction)
+                    return
+                if frame.sample_rate != self._rate or frame.num_channels != 1 or len(frame.audio) % 2:
+                    raise ValueError("invalid audio format")
+                if self._audio_bytes + len(frame.audio) > self._rate * 2 * META_STT_REALTIME_MAX_SECONDS:
+                    await self._error("60-minute session limit reached; start a new recording.", direction)
+                elif not self._failed and self._ws is not None:
+                    # Pace the prebuffer as well as normal frames; Meta rejects burst uploads.
+                    if self._audio_started is None:
+                        self._audio_started = time.monotonic()
+                    delay = self._audio_started + self._audio_bytes / (self._rate * 2) - time.monotonic()
+                    if delay > 0:
+                        await asyncio.sleep(delay)
+                    # Pacing is outside the lock so it cannot delay shutdown.
+                    # Recheck admission under the same lock that closes the socket.
+                    async with self._connect_lock:
+                        if not self._ending and not self._closed and not self._failed:
+                            await self._ws.send_bytes(frame.audio)
+                            self._audio_bytes += len(frame.audio)
+            elif isinstance(frame, (EndFrame, StopFrame, CancelFrame)):
+                await self._close(finalize=not isinstance(frame, CancelFrame), direction=direction)
+        except asyncio.CancelledError:
+            await self._close(finalize=False, direction=direction)
+            raise
+        except Exception:
+            await self._error("audio send or shutdown failed.", direction)
+            await self._close(finalize=False, direction=direction)
+        await self.push_frame(frame, direction)
diff --git a/src/pipeline.py b/src/pipeline.py
index 52627404..aaa191ba 100644
--- a/src/pipeline.py
+++ b/src/pipeline.py
@@ -78,7 +78,9 @@
 
 _SONIOX_MANUAL_FINALIZE_MESSAGE = '{"type": "finalize"}'
 _GROQ_OPENAI_V1_BASE_URL = "https://api.groq.com/openai/v1"
-_PROVIDER_INGRESS_DRAIN_SERVICES = frozenset({"azure_mai", "gemini_realtime", "speechmatics_async"})
+_PROVIDER_INGRESS_DRAIN_SERVICES = frozenset(
+    {"azure_mai", "gemini_realtime", "speechmatics_async", "meta_stt", "meta_stt_async"}
+)
 _PROVIDER_INGRESS_DRAIN_TIMEOUT_SECONDS = 2.0
 _PROVIDER_INGRESS_ABORT_TIMEOUT_SECONDS = 2.0
 
@@ -432,6 +434,7 @@ def _live_service_uses_async_finalization(service_name: str) -> bool:
         "openrouter_stt",
         "speechmatics_async",
         "modulate_async",
+        "meta_stt_async",
         "azure_mai",
         "assemblyai",
     } or (normalized == "soniox" and Config.SONIOX_MODE == "async")
@@ -2217,6 +2220,8 @@ def stt_runtime_configuration(self) -> dict[str, Any]:
             "speechmatics_async": configured_models["speechmatics-async"],
             "modulate": configured_models["modulate-realtime"],
             "modulate_async": configured_models["modulate-async"],
+            "meta_stt": configured_models["meta-realtime"],
+            "meta_stt_async": configured_models["meta-async"],
             "onnx_local": configured_models["onnx_local"],
         }
         modes: dict[str, str] = {
@@ -2245,6 +2250,8 @@ def stt_runtime_configuration(self) -> dict[str, Any]:
             "speechmatics_async": "batch",
             "modulate": "realtime",
             "modulate_async": "batch",
+            "meta_stt": "realtime",
+            "meta_stt_async": "batch",
             "onnx_local": "local",
         }
         fallback_model = models.get(service, "provider-default")
@@ -3678,6 +3685,23 @@ async def stop(self, frame: EndFrame):
                 on_response_complete=self.on_provider_response_complete,
             )
 
+        elif self.service_name in {"meta_stt", "meta_stt_async"}:
+            from src.meta_stt import MetaAsyncProcessor, MetaRealtimeSTTService
+
+            api_key = _get_api_key("meta_stt")
+            if not api_key:
+                raise ValueError("Meta API key is missing.")
+            options = dict(
+                api_key=api_key,
+                model=self._execution_model(Config.META_STT_MODEL),
+                language=self._execution_language(),
+                custom_vocab=self._execution_custom_vocab(),
+                session=session,
+            )
+            if self.service_name == "meta_stt_async" or for_file:
+                return MetaAsyncProcessor(**options, on_progress=self.on_progress)
+            return MetaRealtimeSTTService(**options, sample_rate=Config.SAMPLE_RATE, channels=Config.CHANNELS)
+
         elif self.service_name in {"modulate", "modulate_async"}:
             from src.modulate_stt import (
                 ModulateAsyncProcessor,
@@ -4765,6 +4789,33 @@ async def _transcribe_file_direct_prepared(
                     self.on_progress("Completed")
                 return
 
+            if self.service_name in {"meta_stt", "meta_stt_async"}:
+                from src.meta_stt import META_STT_BATCH_URL, transcribe_with_meta
+
+                endpoint = self._bind_execution_provider_endpoint(META_STT_BATCH_URL)
+                if self.on_progress:
+                    self.on_progress("Transcribing...")
+                async with self._provider_session() as session:
+                    with open(path, "rb") as audio:
+                        payload = await transcribe_with_meta(
+                            session=session,
+                            api_key=Config.get_api_key("meta_stt"),
+                            audio_source=audio,
+                            model=self._execution_model(Config.META_STT_MODEL),
+                            language=self._execution_language(),
+                            custom_vocab=self._execution_custom_vocab(),
+                            mode="DIARIZATION" if self.direct_file_speaker_diarization else "ENDPOINTING",
+                            timeout_secs=batch_timeout_seconds,
+                            endpoint=endpoint,
+                        )
+                self.last_structured_transcript_payload = payload
+                text = payload["transcript"].strip()
+                if text and self.on_transcription:
+                    self.on_transcription(text, True)
+                if self.on_progress:
+                    self.on_progress("Completed")
+                return
+
             if self.service_name in {"modulate", "modulate_async"}:
                 from src.modulate_stt import (
                     modulate_transcript_payload_to_text,
@@ -5198,7 +5249,7 @@ async def stop(self, timeout_secs: float | None = None):
             # existing 30-second budget.
             if is_async_finalization:
                 wait_timeout = 600.0
-            elif self.service_name == "modulate":
+            elif self.service_name in {"modulate", "meta_stt"}:
                 wait_timeout = 40.0
             else:
                 wait_timeout = 30.0
diff --git a/src/provider_transcript.py b/src/provider_transcript.py
index 4dd91757..c0c7c962 100644
--- a/src/provider_transcript.py
+++ b/src/provider_transcript.py
@@ -243,6 +243,36 @@ def normalize_provider_segments(provider: str, payload: Any, source: str, origin
         return []
     provider = str(provider).lower()
 
+    if provider in {"meta_stt", "meta_stt_async"}:
+        turns = payload.get("turns", [])
+        if not isinstance(turns, list):
+            return []
+        segments = []
+        for turn in turns:
+            if not isinstance(turn, dict) or not isinstance(turn.get("transcript"), str):
+                continue
+            if any(type(turn.get(key)) is not int for key in ("turnId", "startMs", "endMs")):
+                continue
+            if not 0 <= turn["startMs"] <= turn["endMs"] or not turn["transcript"].strip():
+                continue
+            speaker = _speaker_key(turn.get("speaker"))
+            segments.append(
+                {
+                    "revision": "canonical",
+                    "source": source,
+                    "providerSegmentId": f"meta-turn-{turn['turnId']}",
+                    "speakerKey": speaker,
+                    "speakerLabel": f"Speaker {speaker}" if speaker else "Meeting audio",
+                    "startMs": origin_ms + turn["startMs"],
+                    "endMs": origin_ms + turn["endMs"],
+                    "text": turn["transcript"].strip(),
+                    "confidence": None,
+                    "isFinal": True,
+                    "alignmentQuality": "provider_segment",
+                }
+            )
+        return segments
+
     if provider in {"soniox", "soniox_async"}:
         return group_provider_words(
             _timed_items(payload.get("tokens", []), start_key="start_ms", end_key="end_ms", scale=1),
diff --git a/src/runtime/provider_dependencies.py b/src/runtime/provider_dependencies.py
index 5818251a..6d81af09 100644
--- a/src/runtime/provider_dependencies.py
+++ b/src/runtime/provider_dependencies.py
@@ -45,6 +45,7 @@ def __init__(
         "Deepgram, Gladia, OpenAI, OpenRouter, Speechmatics, and Gemini async STT adapters",
     ),
     ("src.gemini_realtime_stt", "Gemini 3.5 Transcribe Live Pipecat adapter"),
+    ("src.meta_stt", "Meta Muse Voice Transcribe realtime and asynchronous HTTP adapters"),
     ("src.azure_mai_stt", "Microsoft MAI Transcribe adapter"),
     ("pipecat.services.google.stt", "Google Cloud STT provider"),
     ("pipecat.services.elevenlabs.stt", "ElevenLabs STT provider"),
@@ -57,6 +58,12 @@ def __init__(
 
 
 _PROVIDER_DEPENDENCIES: dict[str, tuple[ProviderRuntimeDependency, ...]] = {
+    "meta_stt": (
+        ProviderRuntimeDependency("meta_stt", "src.meta_stt", "requirements-base.txt", "Meta Voice realtime adapter"),
+    ),
+    "meta_stt_async": (
+        ProviderRuntimeDependency("meta_stt_async", "src.meta_stt", "requirements-base.txt", "Meta Voice HTTP adapter"),
+    ),
     "soniox": (
         ProviderRuntimeDependency(
             "soniox",
diff --git a/src/transcript_artifacts.py b/src/transcript_artifacts.py
index 7c03356d..9b0ee4d2 100644
--- a/src/transcript_artifacts.py
+++ b/src/transcript_artifacts.py
@@ -154,6 +154,8 @@ def provider_batch_model(provider: str) -> str:
         "speechmatics_async": "batch-v2",
         "modulate": "velma-2-stt-batch",
         "modulate_async": "velma-2-stt-batch",
+        "meta_stt": Config.META_STT_MODEL,
+        "meta_stt_async": Config.META_STT_MODEL,
         "smallest": "pulse",
         "smallest_async": "pulse",
         "azure_mai": getattr(Config, "AZURE_MAI_MODEL", "mai-transcribe-1.5"),
@@ -205,6 +207,8 @@ def freeze_provider_route(
         "speechmatics_async",
         "modulate",
         "modulate_async",
+        "meta_stt",
+        "meta_stt_async",
     }
     final_text_only = key in {"modulate", "modulate_async", "openrouter_stt"}
     resolved_model = str(model or provider_batch_model(key)).strip()
diff --git a/src/web_api.py b/src/web_api.py
index 0183a676..92ae7042 100644
--- a/src/web_api.py
+++ b/src/web_api.py
@@ -643,6 +643,7 @@ def _validate_settings_text_lengths(payload: dict[str, Any]) -> None:
     "gladia_async": "Gladia pre-recorded transcription is limited to 135 minutes per request.",
     "modulate_async": "Scriber's 64-kbit/s meeting derivative targets up to three hours within Modulate's 100-MB batch limit; five hours are not supported by this route.",
     "gemini_stt": "Gemini 3.5 Transcribe accepts up to 30 minutes when Scriber requests native diarization and word timestamps.",
+    "meta_stt_async": "Meta Muse Voice Transcribe accepts recordings up to 10 minutes and 32 MB including multipart overhead; turn timestamps are approximate, not word-level.",
 }
 _MEETING_FIVE_HOUR_UNSUPPORTED_REASON = (
     "The current whole-track final transcription route is not yet verified for a five-hour source."
@@ -663,6 +664,7 @@ def _validate_settings_text_lengths(payload: dict[str, Any]) -> None:
         "onnx_local",
         "groq",
         "modulate_async",
+        "meta_stt_async",
     }
 )
 _MEETING_TRANSCRIPTION_MODES = frozenset({"live_final", "final_only"})
@@ -729,6 +731,12 @@ def _validate_settings_text_lengths(payload: dict[str, Any]) -> None:
         "pricingUrl": "https://www.modulate.ai/api/speech-to-text",
         "estimateKind": "published_hourly",
     },
+    "meta_stt_async": {
+        "perTrackHourUsd": 0.18,
+        "systemDiarizationHourUsd": 0.0,
+        "pricingUrl": "https://dev.meta.ai/docs/speech-to-text#availability-pricing",
+        "estimateKind": "published_hourly",
+    },
     "gemini_stt": {
         "perTrackHourUsd": 0.30,
         "systemDiarizationHourUsd": 0.0,
@@ -1629,6 +1637,7 @@ def _live_pipeline_uses_async_finalization(pipeline: Any | None) -> bool:
         "soniox_async",
         "smallest_async",
         "modulate_async",
+        "meta_stt_async",
         "azure_mai",
         "assemblyai",
     } or (service_name == "soniox" and Config.SONIOX_MODE == "async")
@@ -5486,6 +5495,8 @@ def _freeze_background_provider_route(
             endpoint_identity = "https://api.groq.com/openai/v1"
         elif provider_key == "openrouter_stt":
             endpoint_identity = "https://openrouter.ai/api/v1/audio/transcriptions"
+        elif provider_key in {"meta_stt", "meta_stt_async"}:
+            endpoint_identity = "https://api.meta.ai/v1/asr/transcribe"
         resolved_endpoint_sha256 = (
             hashlib.sha256(endpoint_identity.encode("utf-8")).hexdigest() if endpoint_identity else ""
         )
@@ -17609,6 +17620,7 @@ async def _update_settings_unlocked(self, payload: dict[str, Any]) -> dict[str,
                 "onnx_local",
                 "groq",
                 "modulate_async",
+                "meta_stt_async",
             }
             if candidate not in allowed_meeting_final_providers:
                 raise ValueError("Unsupported final meeting transcription provider.")
@@ -19158,6 +19170,12 @@ def final_option_payload(provider: str, metadata: dict[str, Any]) -> dict[str, A
                 "diarization": False,
                 "recommendation": "Final transcript only; uses the optional local Sherpa-ONNX speaker fallback.",
             },
+            "meta_stt_async": {
+                "label": "Meta Muse Voice Transcribe",
+                "model": Config.META_STT_MODEL,
+                "diarization": True,
+                "recommendation": "Native speaker labels and approximate turn timestamps; maximum 10 minutes per recording.",
+            },
             "openai_async": {
                 "label": "OpenAI Batch",
                 "model": Config.OPENAI_STT_MODEL,
@@ -20240,6 +20258,8 @@ def _prewarm_stt_service(service_name: str) -> None:
             from src.smallest_stt import SmallestAsyncProcessor, SmallestRealtimeSTTService  # noqa: F401
         elif service_name in {"modulate", "modulate_async"}:
             from src.modulate_stt import ModulateAsyncProcessor, ModulateRealtimeSTTService  # noqa: F401
+        elif service_name in {"meta_stt", "meta_stt_async"}:
+            from src.meta_stt import MetaAsyncProcessor, MetaRealtimeSTTService  # noqa: F401
         elif service_name == "azure_mai":
             from src.azure_mai_stt import AzureMaiTranscribeSTTService  # noqa: F401
     except ImportError as e:
diff --git a/tests/core/test_provider_audio_formats.py b/tests/core/test_provider_audio_formats.py
index 1169aebe..97d763fa 100644
--- a/tests/core/test_provider_audio_formats.py
+++ b/tests/core/test_provider_audio_formats.py
@@ -213,7 +213,8 @@ def test_matrix_entries_carry_evidence_date_and_revision():
     for capability in PROVIDER_AUDIO_CAPABILITY_MATRIX:
         assert capability.capability_id
         assert capability.revision == CAPABILITY_REVISION
-        assert capability.verified_at == date(2026, 8, 26)
+        expected_date = date(2026, 9, 2) if capability.provider in {"meta_stt", "meta_stt_async"} else date(2026, 8, 26)
+        assert capability.verified_at == expected_date
         assert capability.evidence_reference
 
 
diff --git a/tests/test_meta_stt.py b/tests/test_meta_stt.py
new file mode 100644
index 00000000..81b178ea
--- /dev/null
+++ b/tests/test_meta_stt.py
@@ -0,0 +1,475 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import wave
+from contextlib import asynccontextmanager
+from unittest.mock import AsyncMock
+
+import aiohttp
+import pytest
+from aiohttp import web
+from pipecat.clocks.system_clock import SystemClock
+from pipecat.frames.frames import (
+    AudioRawFrame,
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    TranscriptionFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
+from pipecat.utils.asyncio.task_manager import TaskManager
+
+from src.audio_prepare import ProbedAudioInput, ProviderAudioPreparationError, resolve_provider_audio_selection
+from src.config import Config
+from src.core.provider_audio_formats import AudioInputFormat, AudioSelectionMode, UnsupportedProviderAudioRoute
+from src.core.provider_capabilities import get_capabilities
+from src.core.provider_errors import ProviderTransportError, provider_user_error
+from src.meta_stt import (
+    META_STT_MODEL,
+    MetaAsyncProcessor,
+    MetaRealtimeSTTService,
+    meta_request_options,
+    transcribe_with_meta,
+    validate_meta_wav,
+)
+from src.pipeline import ScriberPipeline, _live_analyzer_requirements, _live_service_uses_async_finalization
+from src.provider_transcript import normalize_provider_segments, normalize_provider_words
+from src.transcript_artifacts import freeze_provider_route
+
+DOWN = FrameDirection.DOWNSTREAM
+
+
+def wav_bytes(*, rate=16000, channels=1, seconds=0.02):
+    stream = io.BytesIO()
+    with wave.open(stream, "wb") as output:
+        output.setnchannels(channels)
+        output.setsampwidth(2)
+        output.setframerate(rate)
+        output.writeframes(b"\0\0" * int(rate * seconds) * channels)
+    return stream.getvalue()
+
+
+def final_payload():
+    return {
+        "sessionId": "session",
+        "transcript": "Hallo Welt.",
+        "audioDurationMs": 2000,
+        "turns": [{"turnId": 1, "startMs": 0, "endMs": 2000, "transcript": "Hallo Welt.", "speaker": "A"}],
+    }
+
+
+@asynccontextmanager
+async def server(handler, *, websocket=False):
+    app = web.Application()
+    app.router.add_route("GET" if websocket else "POST", "/asr", handler)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    port = site._server.sockets[0].getsockname()[1]
+    try:
+        yield f"{'ws' if websocket else 'http'}://127.0.0.1:{port}/asr"
+    finally:
+        await runner.cleanup()
+
+
+def test_model_language_and_vocabulary_are_exact():
+    assert meta_request_options(model=META_STT_MODEL, language="de-DE", custom_vocab=" Scriber, Muse, Scriber ") == {
+        "model": META_STT_MODEL,
+        "languageBias": ["German"],
+        "keywords": ["Scriber", "Muse"],
+    }
+    assert meta_request_options(model=META_STT_MODEL, language="auto") == {"model": META_STT_MODEL}
+    with pytest.raises(ValueError, match="not verified"):
+        meta_request_options(model="muse-voice-transcribe-async")
+    with pytest.raises(ValueError, match="not supported"):
+        meta_request_options(model=META_STT_MODEL, language="xx")
+
+
+@pytest.mark.asyncio
+async def test_direct_file_uses_frozen_model_language_and_vocabulary(monkeypatch, tmp_path):
+    from src import audio_prepare
+
+    source = tmp_path / "input.wav"
+    source.write_bytes(wav_bytes())
+    monkeypatch.setattr(Config, "MODEL_API_KEY", "test-key")
+    monkeypatch.setattr(
+        audio_prepare,
+        "probe_audio_input_file",
+        lambda path: ProbedAudioInput(
+            AudioInputFormat.WAV_PCM16, "wav", "pcm_s16le", 16000, 1, 20, path.stat().st_size
+        ),
+    )
+    route = freeze_provider_route(workload="file", provider="meta_stt_async", language="de", custom_vocab="Scriber")
+    monkeypatch.setattr(Config, "META_STT_MODEL", "not-the-frozen-model")
+    monkeypatch.setattr(Config, "LANGUAGE", "fr")
+    transcribe = AsyncMock(return_value=final_payload())
+    monkeypatch.setattr("src.meta_stt.transcribe_with_meta", transcribe)
+
+    class Transport:
+        async def session_view(self, **kwargs):
+            return object()
+
+    pipeline = ScriberPipeline(
+        service_name="meta_stt_async", execution_route=route.execution_route(), provider_http_transport=Transport()
+    )
+    results = []
+    pipeline.on_transcription = lambda text, final: results.append((text, final))
+    await pipeline.transcribe_file_direct(str(source))
+    assert transcribe.call_args.kwargs["model"] == META_STT_MODEL
+    assert transcribe.call_args.kwargs["language"] == "de"
+    assert transcribe.call_args.kwargs["custom_vocab"] == "Scriber"
+    assert transcribe.call_args.kwargs["endpoint"] == "https://api.meta.ai/v1/asr/transcribe"
+    assert results == [("Hallo Welt.", True)]
+    assert pipeline.last_structured_transcript_payload == final_payload()
+    assert source.read_bytes() == wav_bytes()
+
+
+@pytest.mark.parametrize("rate,channels", [(44100, 1), (16000, 2)])
+def test_invalid_wav_is_rejected(rate, channels):
+    with pytest.raises(ValueError, match="mono PCM16"):
+        validate_meta_wav(io.BytesIO(wav_bytes(rate=rate, channels=channels)))
+
+
+def test_wav_size_duration_and_truncation_are_checked():
+    stream = io.BytesIO(wav_bytes())
+    stream.seek(3)
+    validate_meta_wav(stream)
+    assert stream.tell() == 3 and not stream.closed
+    with pytest.raises(ValueError, match="truncated"):
+        validate_meta_wav(io.BytesIO(wav_bytes()[:-2]))
+    with pytest.raises(ValueError, match="10 minutes"):
+        validate_meta_wav(io.BytesIO(wav_bytes(seconds=601)))
+    with pytest.raises(ValueError, match="32 MB"):
+        validate_meta_wav(io.BytesIO(b"0" * 32_000_000))
+
+
+@pytest.mark.asyncio
+async def test_http_contract_real_multipart_and_normalized_output():
+    seen = {}
+
+    async def handler(request):
+        assert request.headers["Authorization"] == "Bearer test-secret"
+        assert request.headers["Accept"] == "application/json"
+        reader = await request.multipart()
+        settings = await reader.next()
+        assert settings.name == "request" and settings.headers["Content-Type"] == "application/json"
+        seen.update(await settings.json())
+        audio = await reader.next()
+        assert audio.name == "audio"
+        validate_meta_wav(io.BytesIO(await audio.read()))
+        return web.json_response({**final_payload(), "unexpected": "private"})
+
+    async with server(handler) as endpoint, aiohttp.ClientSession() as session:
+        payload = await transcribe_with_meta(
+            session=session,
+            api_key="test-secret",
+            audio_source=wav_bytes(),
+            language="de",
+            custom_vocab="Scriber",
+            mode="DIARIZATION",
+            endpoint=endpoint,
+        )
+    assert seen == {
+        "model": META_STT_MODEL,
+        "audioEncoding": "WAV",
+        "mode": "DIARIZATION",
+        "languageBias": ["German"],
+        "keywords": ["Scriber"],
+    }
+    assert set(payload) == {"transcript", "turns", "audioDurationMs"}
+    segments = normalize_provider_segments("meta_stt_async", payload, "mix")
+    assert segments[0]["speakerKey"] == "A" and segments[0]["alignmentQuality"] == "provider_segment"
+    assert normalize_provider_words("meta_stt_async", payload) == []
+
+
+@pytest.mark.parametrize("status", [400, 401, 403, 413, 429, 500])
+@pytest.mark.asyncio
+async def test_http_errors_are_safe_and_never_retried(status):
+    calls = []
+
+    async def handler(request):
+        calls.append(1)
+        await request.read()
+        return web.Response(status=status, text="test-secret transcript-private")
+
+    async with server(handler) as endpoint, aiohttp.ClientSession() as session:
+        with pytest.raises(ProviderTransportError) as raised:
+            await transcribe_with_meta(
+                session=session, api_key="test-secret", audio_source=wav_bytes(), endpoint=endpoint
+            )
+    assert calls == [1]
+    public = provider_user_error("meta_stt_async", raised.value)
+    assert "test-secret" not in str(raised.value) and "transcript-private" not in str(public)
+
+
+@pytest.mark.asyncio
+async def test_http_rejects_invalid_success():
+    async def handler(request):
+        await request.read()
+        return web.json_response({"text": "wrong schema"})
+
+    async with server(handler) as endpoint, aiohttp.ClientSession() as session:
+        with pytest.raises(RuntimeError, match="invalid final"):
+            await transcribe_with_meta(session=session, api_key="test", audio_source=wav_bytes(), endpoint=endpoint)
+
+
+@pytest.mark.asyncio
+async def test_realtime_handshake_pcm_drain_and_final_turns():
+    seen = {}
+
+    async def handler(request):
+        assert "Authorization" not in request.headers
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        seen.update(await ws.receive_json())
+        await ws.send_json({"sessionId": "test"})
+        audio = await ws.receive()
+        assert audio.type == aiohttp.WSMsgType.BINARY and audio.data == b"\0\0" * 160
+        await ws.send_json({"type": "speechStart", "turnId": 4, "audioProcessedMs": 0})
+        await ws.send_json({"type": "transcript", "transcript": "H", "final": False, "audioProcessedMs": 5})
+        await ws.send_json({"type": "transcript", "transcript": "Hallo", "final": True, "audioProcessedMs": 10})
+        assert await ws.receive_json() == {"type": "endStream"}
+        await ws.send_json({"type": "speechComplete", "turnId": 4, "transcript": "Hallo!", "audioProcessedMs": 10})
+        await ws.close(code=1000)
+        return ws
+
+    async with server(handler, websocket=True) as endpoint, aiohttp.ClientSession() as session:
+        service = MetaRealtimeSTTService(session=session, api_key="test-secret", language="de", endpoint=endpoint)
+        service.push_frame = AsyncMock()
+        await service._connect(DOWN)
+        await service.process_frame(AudioRawFrame(audio=b"\0\0" * 160, sample_rate=16000, num_channels=1), DOWN)
+        await asyncio.sleep(0.02)
+        await service._close(finalize=True, direction=DOWN)
+        frames = [call.args[0] for call in service.push_frame.call_args_list]
+        assert not any(isinstance(frame, ErrorFrame) for frame in frames)
+        assert [frame.text for frame in frames if type(frame) is TranscriptionFrame] == ["Hallo!"]
+        assert [frame.text for frame in frames if isinstance(frame, InterimTranscriptionFrame)] == ["H", "Hallo"]
+        assert service._tasks.pending_count == 0
+    assert seen["authorization"] == {"accessToken": "Bearer test-secret"}
+    assert seen["audioEncoding"] == "PCM_16KHZ" and seen["mode"] == "ENDPOINTING"
+    assert seen["partialMode"] == "CUMULATIVE" and seen["model"] == META_STT_MODEL
+
+
+@asynccontextmanager
+async def delayed_realtime_server(stage):
+    paused = asyncio.Event()
+    resume = asyncio.Event()
+    received = asyncio.Queue()
+
+    async def pause_at(boundary):
+        if stage == boundary:
+            paused.set()
+            await resume.wait()
+
+    async def handler(request):
+        await pause_at("upgrade")
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        await ws.receive_json()
+        await pause_at("acknowledgement")
+        await ws.send_json({"sessionId": "test"})
+        async for message in ws:
+            if message.type == aiohttp.WSMsgType.BINARY:
+                received.put_nowait(message.data)
+        return ws
+
+    async with server(handler, websocket=True) as endpoint:
+        yield endpoint, paused, resume, received
+
+
+@pytest.mark.parametrize("stage", ["upgrade", "acknowledgement"])
+@pytest.mark.asyncio
+async def test_audio_waits_for_authenticated_startup(stage):
+    async with (
+        delayed_realtime_server(stage) as (endpoint, paused, resume, received),
+        aiohttp.ClientSession() as session,
+    ):
+        service = MetaRealtimeSTTService(session=session, api_key="test-secret", endpoint=endpoint)
+        service.push_frame = AsyncMock()
+        await service.setup(FrameProcessorSetup(clock=SystemClock(), task_manager=TaskManager(), pipeline_worker=None))
+        start = asyncio.create_task(service.process_frame(StartFrame(), DOWN))
+        audio = None
+        try:
+            await asyncio.wait_for(paused.wait(), 2)
+            audio = asyncio.create_task(
+                service.process_frame(AudioRawFrame(audio=b"\0\0" * 160, sample_rate=16000, num_channels=1), DOWN)
+            )
+            await asyncio.sleep(0)  # let audio enter while startup is held at the barrier
+            assert not audio.done(), "Audio must wait for the authenticated socket, not return during startup"
+            assert service._audio_bytes == 0
+            resume.set()
+            await asyncio.wait_for(asyncio.gather(start, audio), 2)
+            assert await asyncio.wait_for(received.get(), 2) == b"\0\0" * 160
+            assert service._audio_bytes == 320 and received.empty()
+        finally:
+            resume.set()
+            await asyncio.gather(start, *([audio] if audio is not None else []), return_exceptions=True)
+            await service._close(finalize=False, direction=DOWN)
+            await service.cleanup()
+        assert not any(isinstance(call.args[0], ErrorFrame) for call in service.push_frame.call_args_list)
+
+
+@pytest.mark.parametrize("stage", ["upgrade", "acknowledgement"])
+@pytest.mark.asyncio
+async def test_cancel_during_startup_leaves_no_socket_or_receiver(stage):
+    async with (
+        delayed_realtime_server(stage) as (endpoint, paused, resume, received),
+        aiohttp.ClientSession() as session,
+    ):
+        service = MetaRealtimeSTTService(session=session, api_key="test-secret", endpoint=endpoint)
+        service.push_frame = AsyncMock()
+        await service.setup(FrameProcessorSetup(clock=SystemClock(), task_manager=TaskManager(), pipeline_worker=None))
+        start = asyncio.create_task(service.process_frame(StartFrame(), DOWN))
+        cancel = None
+        try:
+            await asyncio.wait_for(paused.wait(), 2)
+            cancel = asyncio.create_task(service.process_frame(CancelFrame(), DOWN))
+            await asyncio.sleep(0)
+            resume.set()
+            await asyncio.wait_for(asyncio.gather(start, cancel), 2)
+            # Assert before ClientSession.__aexit__ can conceal a leaked socket.
+            assert not session.closed
+            assert service._closed and (service._ws is None or service._ws.closed)
+            assert service._tasks.pending_count == 0 and received.empty()
+        finally:
+            resume.set()
+            await asyncio.gather(start, *([cancel] if cancel is not None else []), return_exceptions=True)
+            await service._tasks.close(timeout_seconds=2, cancel=True)
+            if service._ws is not None:
+                await service._ws.close()
+            await service.cleanup()
+
+
+@pytest.mark.asyncio
+async def test_overlapping_turns_are_ordered_and_deduplicated():
+    service = MetaRealtimeSTTService(session=AsyncMock(), api_key="test")
+    service.push_frame = AsyncMock()
+    for turn_id in [8, 3]:
+        await service._handle_event({"type": "speechStart", "turnId": turn_id}, DOWN)
+    await service._handle_event({"type": "speechComplete", "turnId": 3, "transcript": "Second."}, DOWN)
+    service.push_frame.assert_not_called()
+    await service._handle_event({"type": "speechComplete", "turnId": 8, "transcript": "First."}, DOWN)
+    await service._handle_event({"type": "speechComplete", "turnId": 8, "transcript": "Duplicate."}, DOWN)
+    assert [call.args[0].text for call in service.push_frame.call_args_list] == ["First.", "Second."]
+
+
+@pytest.mark.parametrize("stage", ["upgrade", "acknowledgement"])
+@pytest.mark.asyncio
+async def test_cancel_discards_audio_already_waiting_for_startup(stage):
+    async with (
+        delayed_realtime_server(stage) as (endpoint, paused, resume, received),
+        aiohttp.ClientSession() as session,
+    ):
+        service = MetaRealtimeSTTService(session=session, api_key="test-secret", endpoint=endpoint)
+        service.push_frame = AsyncMock()
+        await service.setup(FrameProcessorSetup(clock=SystemClock(), task_manager=TaskManager(), pipeline_worker=None))
+        start = asyncio.create_task(service.process_frame(StartFrame(), DOWN))
+        pending = [start]
+        try:
+            await asyncio.wait_for(paused.wait(), 2)
+            pending.append(
+                asyncio.create_task(
+                    service.process_frame(AudioRawFrame(audio=b"\0\0" * 160, sample_rate=16000, num_channels=1), DOWN)
+                )
+            )
+            await asyncio.sleep(0)
+            pending.append(asyncio.create_task(service.process_frame(CancelFrame(), DOWN)))
+            async with asyncio.timeout(2):
+                while not service._ending:
+                    await asyncio.sleep(0)
+            resume.set()
+            await asyncio.wait_for(asyncio.gather(*pending), 2)
+            assert service._audio_bytes == 0 and received.empty(), "Canceled queued audio must never be sent"
+            assert service._ws.closed and service._tasks.pending_count == 0
+            assert not any(isinstance(call.args[0], ErrorFrame) for call in service.push_frame.call_args_list)
+        finally:
+            resume.set()
+            await asyncio.gather(*pending, return_exceptions=True)
+            await service.cleanup()
+
+
+@pytest.mark.parametrize("mode", ["error", "disconnect", "timeout", "bad_handshake"])
+@pytest.mark.asyncio
+async def test_realtime_failure_never_becomes_success(mode):
+    async def handler(request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        await ws.receive_json()
+        if mode == "bad_handshake":
+            await ws.send_json({"type": "error", "message": "test-secret"})
+        else:
+            await ws.send_json({"sessionId": "test"})
+            await ws.receive_json()
+            if mode == "timeout":
+                await asyncio.sleep(0.1)
+            elif mode == "error":
+                await ws.send_json({"type": "error", "message": "test-secret transcript-private"})
+        await ws.close(code=1008)
+        return ws
+
+    async with server(handler, websocket=True) as endpoint, aiohttp.ClientSession() as session:
+        service = MetaRealtimeSTTService(session=session, api_key="test-secret", endpoint=endpoint)
+        service.push_frame = AsyncMock()
+        service._final_timeout = 0.03
+        await service._connect(DOWN)
+        await service._close(finalize=True, direction=DOWN)
+        errors = [
+            call.args[0].error for call in service.push_frame.call_args_list if isinstance(call.args[0], ErrorFrame)
+        ]
+        assert len(errors) == 1 and "test-secret" not in errors[0] and "transcript-private" not in errors[0]
+        assert service._tasks.pending_count == 0 and not session.closed
+
+
+@pytest.mark.parametrize("cancel", [False, True])
+@pytest.mark.asyncio
+async def test_async_uploads_once_on_stop_never_on_cancel(monkeypatch, cancel):
+    transcribe = AsyncMock(return_value=final_payload())
+    monkeypatch.setattr("src.meta_stt.transcribe_with_meta", transcribe)
+    processor = MetaAsyncProcessor(session=AsyncMock(), api_key="test")
+    processor.push_frame = AsyncMock()
+    await processor.process_frame(AudioRawFrame(audio=b"\0\0" * 160, sample_rate=16000, num_channels=1), DOWN)
+    await processor.process_frame(CancelFrame() if cancel else EndFrame(), DOWN)
+    await processor.process_frame(EndFrame(), DOWN)
+    assert transcribe.await_count == (0 if cancel else 1)
+    if not cancel:
+        assert transcribe.call_args.kwargs["mode"] == "PUSH_TO_TALK"
+    assert processor._buffer.closed
+
+
+def test_settings_capabilities_routes_and_native_analyzers(monkeypatch):
+    monkeypatch.setattr(Config, "MODEL_API_KEY", "test-meta")
+    monkeypatch.setattr(Config, "SEGMENT_SPEECH_WITH_VAD", True)
+    for provider in ("meta_stt", "meta_stt_async"):
+        assert Config.get_api_key(provider) == "test-meta"
+        assert get_capabilities(provider).supports_batch_diarization
+        assert not get_capabilities(provider).supports_word_timestamps
+        assert get_capabilities(provider).meeting_max_duration_seconds == 600
+        route = freeze_provider_route(workload="file", provider=provider)
+        assert route.model == META_STT_MODEL and route.provider_route == "asr_transcribe"
+        assert route.transport == "direct_upload"
+        service = ScriberPipeline(service_name=provider)._create_stt_service(AsyncMock())
+        assert isinstance(service, MetaAsyncProcessor if provider.endswith("async") else MetaRealtimeSTTService)
+    assert _live_analyzer_requirements("meta_stt") == (False, False)
+    assert _live_service_uses_async_finalization("meta_stt_async")
+
+
+@pytest.mark.parametrize(
+    "rate,channels,expected",
+    [(16000, 1, "original_passthrough"), (24000, 1, "original_passthrough"), (44100, 2, "generated")],
+)
+def test_audio_selection_checks_rate_channels_and_model(rate, channels, expected):
+    probe = ProbedAudioInput(AudioInputFormat.WAV_PCM16, "wav", "pcm_s16le", rate, channels, 1000, 32044)
+    _, selection = resolve_provider_audio_selection(provider="meta_stt_async", model=META_STT_MODEL, probe=probe)
+    assert selection.mode == AudioSelectionMode(expected)
+    with pytest.raises(UnsupportedProviderAudioRoute):
+        resolve_provider_audio_selection(provider="meta_stt_async", model="unknown", probe=probe)
+    with pytest.raises(ProviderAudioPreparationError, match="10 minutes"):
+        resolve_provider_audio_selection(
+            provider="meta_stt_async",
+            model=META_STT_MODEL,
+            probe=ProbedAudioInput(AudioInputFormat.WAV_PCM16, "wav", "pcm_s16le", rate, channels, 600001, 32044),
+        )