refactor: extract shared audio container sniffer to tools/audio_container.py

One sniffer owns magic-byte container detection (Teknium's one-concept-
one-owner rule): the new tools/audio_container.py is used by

- gateway/platforms/base.py _sniff_audio_ext (inbound cache — PR #36166's
  central sniffer, now covering AAC/ADTS, MP4-brand disambiguation, webm)
- gateway/platforms/signal.py _guess_extension (audio/AV branches
  delegated; RIFF/WAVE fix from PR #50690 and M4A-brand fix from
  PR #72490 now live centrally)
- tools/tts_tool.py _sniff_audio_container (outbound repair, PR #73072)

cache_audio_from_url inherits the sniff via cache_audio_from_bytes.
Adds tests/tools/test_audio_container.py covering every magic-byte type,
wrong-extension repair on the inbound cache, unknown passthrough, the
URL path, and Signal's delegation.

tools/audio_container.py

@@ -0,0 +1,97 @@
+"""Shared magic-byte audio/AV container detection.
+
+ONE sniffer owns container detection for the whole codebase:
+
+- **Outbound** (``tools/tts_tool.py``): TTS backends silently ignore the
+  requested opus format (Edge emits MP3, Piper writes WAV, ...), so the
+  synthesized file is sniffed and repaired when the bytes don't match the
+  ``.ogg`` extension (PR #73072).
+- **Inbound** (``gateway/platforms/base.py`` ``cache_audio_from_bytes`` /
+  ``cache_audio_from_url``): platform adapters frequently pass a wrong or
+  guessed extension for voice notes (Telegram ``.oga``, iOS Signal M4A-branded
+  MP4, RIFF/WAVE attachments). The cache sniffs the real container so STT and
+  downstream players get an honest extension — the inbound mirror of the
+  outbound repair.
+- ``gateway/platforms/signal.py`` ``_guess_extension`` delegates its audio/AV
+  branches here instead of duplicating the byte patterns.
+
+Detection notes:
+
+- RIFF needs the form-type at bytes 8-11 to split ``WAVE`` (wav) from ``WEBP``
+  (image — deliberately NOT handled here; this module only claims audio/AV
+  containers, callers check images first).
+- ``ftyp`` needs the brand at bytes 8-11 to split audio brands (``M4A ``,
+  ``M4B ``) from video brands (isom/mp42/avc1/qt).
+- The ``0xFF 0xFx`` sync word is shared by MP3 and ADTS AAC; bits 3-1 of
+  byte 1 disambiguate (ADTS: ``ID=0``, ``layer=00``).
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+# Container id -> canonical file extension.
+CONTAINER_TO_EXT = {
+    "m4a": ".m4a",
+    "mp4": ".mp4",
+    "ogg": ".ogg",
+    "flac": ".flac",
+    "wav": ".wav",
+    "mp3": ".mp3",
+    "aac": ".aac",
+    "webm": ".webm",
+}
+
+# MP4 ftyp brands that mean "this is audio" (iOS voice notes use M4A ).
+_MP4_AUDIO_BRANDS = (b"m4a ", b"m4b ")
+
+
+def sniff_container(data: bytes) -> Optional[str]:
+    """Return a container id from magic bytes, or ``None`` when unknown.
+
+    Possible ids: ``m4a``, ``mp4``, ``ogg``, ``flac``, ``wav``, ``mp3``,
+    ``aac``, ``webm``. Only audio/AV containers are claimed — images
+    (including RIFF/WEBP) return ``None`` so callers can layer their own
+    image detection first.
+    """
+    if len(data) >= 8 and data[4:8] == b"ftyp":
+        # Brand at bytes 8-11: audio brands ("M4A ", "M4B ") are voice
+        # notes / audiobooks; everything else (isom/mp42/avc1/qt) is video.
+        if len(data) >= 12 and data[8:12].lower() in _MP4_AUDIO_BRANDS:
+            return "m4a"
+        return "mp4"
+    if data.startswith(b"OggS"):
+        return "ogg"
+    if data.startswith(b"fLaC"):
+        return "flac"
+    if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WAVE":
+        return "wav"
+    if data.startswith(b"ID3"):
+        return "mp3"
+    if len(data) >= 2 and data[0] == 0xFF and (data[1] & 0xE0) == 0xE0:
+        # ``0xFF 0xFx`` is shared by MP3 and ADTS AAC. Bits 3-1 of byte 1
+        # disambiguate: ADTS has ``ID=0`` and ``layer=00`` (mask 0xF6,
+        # target 0xF0); MP3 has ``ID=1`` and ``layer`` in {01,10,11}.
+        if (data[1] & 0xF6) == 0xF0:
+            return "aac"
+        return "mp3"
+    if data.startswith(b"\x1a\x45\xdf\xa3"):
+        return "webm"
+    return None
+
+
+def sniff_audio_ext(data: bytes, fallback_ext: str = ".ogg") -> str:
+    """Return a container-matching extension, or ``fallback_ext`` when unknown.
+
+    Used on inbound audio paths where the caller *claims* the bytes are audio:
+    generic MP4 containers are mapped to ``.m4a`` (audio-in-MP4) because in an
+    audio context the payload is AAC audio regardless of brand — STT accepts
+    ``.m4a``/``.mp4`` but voice-bubble routing keys off audio extensions.
+    """
+    fallback = fallback_ext if fallback_ext.startswith(".") else f".{fallback_ext}"
+    container = sniff_container(data)
+    if container is None:
+        return fallback
+    if container == "mp4":
+        return ".m4a"
+    return CONTAINER_TO_EXT[container]

tools/tts_tool.py

@@ -995,21 +995,20 @@ def _ffmpeg_transcode_to_opus(input_path: str, ogg_path: str) -> Optional[str]:
 # ---------------------------------------------------------------------------
 
 def _sniff_audio_container(path: str) -> str:
-    """Return 'ogg' | 'wav' | 'mp3' | 'flac' | 'unknown' from magic bytes."""
+    """Return a container id ('ogg', 'wav', 'mp3', 'flac', ...) or 'unknown'.
+
+    Delegates to the shared magic-byte sniffer in ``tools.audio_container``
+    (one module owns container detection for both this outbound repair and
+    the inbound gateway audio cache).
+    """
+    from tools.audio_container import sniff_container
+
     try:
         with open(path, "rb") as fh:
             head = fh.read(12)
     except OSError:
         return "unknown"
-    if head[:4] == b"OggS":
-        return "ogg"
-    if head[:4] == b"RIFF" and head[8:12] == b"WAVE":
-        return "wav"
-    if head[:4] == b"fLaC":
-        return "flac"
-    if head[:3] == b"ID3" or (len(head) >= 2 and head[0] == 0xFF and (head[1] & 0xE0) == 0xE0):
-        return "mp3"
-    return "unknown"
+    return sniff_container(head) or "unknown"
 
 
 def _repair_ogg_container(file_str: str) -> str: