fix(transcription): key the cache by engine, not just by message

A cached telegram transcript was answering groq requests. The lookup ran
before the engine was resolved and matched any source, so an explicit
engine=groq silently returned the native text - and the native engine
drops the recording's last speech segment in roughly 2 of 3 recordings,
a loss that is invisible in the text itself.

The cache is now keyed by (chat_id, message_id, source) and the tool
resolves the engine before looking anything up. Unpinned callers
(listing render, backfill skip check) still accept any source and
prefer the default engine, so display and quota behaviour are unchanged.
Existing caches are rebuilt in place on first use.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: PathKnower

telegram_mcp/tools/messages.py

@@ -1021,9 +1021,11 @@ async def transcribe_voice(
       Requires Telegram Premium on this account; polls briefly (up to ~20s)
       while Telegram finishes a long recording.
 
-    Results are cached by (chat_id, message_id) - a repeat call for an
-    already-transcribed message returns the cached text without hitting
-    either API again.
+    Results are cached per engine, by (chat_id, message_id, engine) - a
+    repeat call with the same engine returns the cached text without
+    hitting either API again. Asking for an engine that has no cached
+    result transcribes with it, even when the other engine's text is
+    already cached.
 
     The returned text is a machine transcript, not a verbatim quote: proper
     names, punctuation and occasional words drift under both engines.
@@ -1045,7 +1047,16 @@ async def transcribe_voice(
         entity = await resolve_entity(chat_id, cl)
         numeric_chat_id = get_marked_id(entity)
 
-        cached = transcription.get_cached_transcript(numeric_chat_id, message_id)
+        chosen_engine = (engine or transcription.default_engine()).strip().lower()
+        if chosen_engine not in transcription.ENGINES:
+            return f"Invalid engine '{engine}'. Use 'telegram' or 'groq'."
+
+        # Pinned to the chosen engine on purpose: a cached telegram transcript
+        # must not answer a groq request. The native engine drops the last
+        # speech segment and the loss cannot be seen in the text.
+        cached = transcription.get_cached_transcript(
+            numeric_chat_id, message_id, source=chosen_engine
+        )
         if cached is not None:
             return json.dumps(
                 {
@@ -1066,9 +1077,6 @@ async def transcribe_voice(
         if not transcription.is_transcribable(msg):
             return f"Message {message_id} has no voice message or video note to transcribe."
 
-        chosen_engine = (engine or transcription.default_engine()).strip().lower()
-        if chosen_engine not in transcription.ENGINES:
-            return f"Invalid engine '{engine}'. Use 'telegram' or 'groq'."
         if chosen_engine == "groq" and not os.getenv("GROQ_API_KEY"):
             return (
                 "GROQ_API_KEY is not configured on this server. "

telegram_mcp/transcription.py

@@ -118,10 +118,11 @@ def _connect() -> sqlite3.Connection:
             duration INTEGER,
             lang TEXT,
             created_at TEXT NOT NULL,
-            PRIMARY KEY (chat_id, message_id)
+            PRIMARY KEY (chat_id, message_id, source)
         )
         """
     )
+    _migrate_source_into_key(conn)
     conn.commit()
     try:
         os.chmod(path, 0o600)
@@ -130,14 +131,65 @@ def _connect() -> sqlite3.Connection:
     return conn
 
 
-def get_cached_transcript(chat_id: int, message_id: int) -> Optional[dict]:
+def _migrate_source_into_key(conn: sqlite3.Connection) -> None:
+    """Older builds keyed the cache on (chat_id, message_id) alone, so a cheap
+    telegram transcript permanently shadowed the groq one - including for
+    callers that asked for groq explicitly. Rebuild such a table in place."""
+    row = conn.execute(
+        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'transcripts'"
+    ).fetchone()
+    if not row or not row[0]:
+        return
+    if "PRIMARY KEY (chat_id, message_id, source)" in row[0]:
+        return
+    conn.executescript(
+        """
+        ALTER TABLE transcripts RENAME TO transcripts_legacy;
+        CREATE TABLE transcripts (
+            chat_id INTEGER NOT NULL,
+            message_id INTEGER NOT NULL,
+            source TEXT NOT NULL,
+            text TEXT NOT NULL,
+            duration INTEGER,
+            lang TEXT,
+            created_at TEXT NOT NULL,
+            PRIMARY KEY (chat_id, message_id, source)
+        );
+        INSERT OR IGNORE INTO transcripts
+            (chat_id, message_id, source, text, duration, lang, created_at)
+        SELECT chat_id, message_id, source, text, duration, lang, created_at
+        FROM transcripts_legacy;
+        DROP TABLE transcripts_legacy;
+        """
+    )
+
+
+def get_cached_transcript(
+    chat_id: int, message_id: int, source: Optional[str] = None
+) -> Optional[dict]:
+    """Cached transcript for a message.
+
+    ``source`` pins the engine: asking for groq must never be answered with a
+    telegram transcript, because the native engine drops the recording's last
+    segment and the loss is invisible in the text. Callers that only want to
+    display whatever exists (listings, backfill skip checks) pass None and get
+    the default engine's row when there is one, any row otherwise.
+    """
     conn = _connect()
     try:
-        row = conn.execute(
-            "SELECT source, text, duration, lang, created_at FROM transcripts "
-            "WHERE chat_id = ? AND message_id = ?",
-            (chat_id, message_id),
-        ).fetchone()
+        if source is not None:
+            row = conn.execute(
+                "SELECT source, text, duration, lang, created_at FROM transcripts "
+                "WHERE chat_id = ? AND message_id = ? AND source = ?",
+                (chat_id, message_id, source),
+            ).fetchone()
+        else:
+            row = conn.execute(
+                "SELECT source, text, duration, lang, created_at FROM transcripts "
+                "WHERE chat_id = ? AND message_id = ? "
+                "ORDER BY source = ? DESC, created_at DESC LIMIT 1",
+                (chat_id, message_id, default_engine()),
+            ).fetchone()
     finally:
         conn.close()
     if row is None:

tests/test_transcription.py

@@ -107,6 +107,57 @@ def test_cache_keyed_by_chat_and_message_independently(transcript_cache_dir):
     assert transcription.get_cached_transcript(1, 999) is None
 
 
+def test_cache_pinned_to_an_engine_never_serves_the_other_engines_text(transcript_cache_dir):
+    """The whole point of choosing groq is that the native engine drops the
+    recording's last segment. A telegram transcript answering a groq request
+    would hand back that truncated text with no way to tell."""
+    transcription.save_transcript(1, 2, "telegram", "truncated tail")
+    assert transcription.get_cached_transcript(1, 2, source="groq") is None
+    assert transcription.get_cached_transcript(1, 2, source="telegram")["text"] == (
+        "truncated tail"
+    )
+
+
+def test_cache_keeps_both_engines_side_by_side(transcript_cache_dir):
+    transcription.save_transcript(1, 2, "telegram", "native text")
+    transcription.save_transcript(1, 2, "groq", "groq text")
+    assert transcription.get_cached_transcript(1, 2, source="telegram")["text"] == "native text"
+    assert transcription.get_cached_transcript(1, 2, source="groq")["text"] == "groq text"
+
+
+def test_legacy_cache_keyed_without_engine_is_migrated_in_place(transcript_cache_dir):
+    """Builds before the fix keyed on (chat_id, message_id) alone. Rows must
+    survive the rebuild, and the pinned lookup must start working on them."""
+    import sqlite3
+
+    path = transcription._cache_db_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(str(path))
+    conn.executescript(
+        """
+        CREATE TABLE transcripts (
+            chat_id INTEGER NOT NULL,
+            message_id INTEGER NOT NULL,
+            source TEXT NOT NULL,
+            text TEXT NOT NULL,
+            duration INTEGER,
+            lang TEXT,
+            created_at TEXT NOT NULL,
+            PRIMARY KEY (chat_id, message_id)
+        );
+        INSERT INTO transcripts VALUES (1, 2, 'telegram', 'old row', 23, 'ru', '2026-08-20');
+        """
+    )
+    conn.commit()
+    conn.close()
+
+    assert transcription.get_cached_transcript(1, 2, source="telegram")["text"] == "old row"
+    assert transcription.get_cached_transcript(1, 2, source="groq") is None
+    transcription.save_transcript(1, 2, "groq", "new row")
+    assert transcription.get_cached_transcript(1, 2, source="groq")["text"] == "new row"
+    assert transcription.get_cached_transcript(1, 2, source="telegram")["text"] == "old row"
+
+
 @pytest.mark.skipif(os.name == "nt", reason="POSIX permissions only")
 def test_cache_file_and_directory_get_restrictive_permissions(transcript_cache_dir):
     transcription.save_transcript(1, 2, "groq", "hi")