fix(gateway): keep the STT echo ledger across pending-media merges

_invalidate_pending_stt_cache() clears the gateway-side transcription
cache when merge_pending_message_event() folds a follow-up message into a
still-pending event, so the next transcription picks up the merged text
and attachments.  It also cleared _gateway_pending_stt_echo_sent, but that
flag is not derived state — it records that the transcript was already
delivered to the user.

Dropping it makes the re-run transcription echo the earlier notes a second
time.  Both merge branches are affected, including the text-only follow-up
case where no new audio arrived at all: there the cache is invalidated,
the same voice note is transcribed again (a second paid STT call) and the
same line is echoed again.

Sequence:

  1. voice note arrives, interrupt monitor transcribes it and echoes
     '🎙️ "hello"'
  2. user sends a follow-up while the turn is still pending, so it merges
  3. drain path re-transcribes and echoes '🎙️ "hello"' a second time

Keep the ledger out of the invalidation set and track it as a count of
already-echoed transcripts instead of a single boolean.  A count is what
the merge case actually needs: re-running transcription over the extended
media list returns the earlier transcripts as a prefix of the new one, so
echoing only the unsent tail suppresses the repeat while still surfacing a
newly merged voice note.  A count rather than a set of seen values, so two
separate notes that transcribe identically stay two distinct deliveries —
covered by test_pending_stt_merge_echoes_two_identical_transcripts.

The guard stays within the 12-line window that
test_all_gateway_transcript_echo_sends_are_gated enforces over run.py.

gateway/platforms/base.py

@@ -2316,11 +2316,16 @@ def _invalidate_pending_stt_cache(event: MessageEvent) -> None:
     ``_transcribe_pending_audio_event_once``); if the cached event gains new
     media after the cache was populated, the stale transcript must be
     discarded so the next transcription call picks up the merged attachments.
+
+    Only the *derived* transcription cache is dropped.  The echo ledger
+    (``_gateway_pending_stt_echoed``) records which transcripts were already
+    delivered to the user and must survive the merge: the re-run transcription
+    returns the earlier notes again, so clearing the ledger would echo them a
+    second time.
     """
     for attr in (
         "_gateway_pending_stt_text",
         "_gateway_pending_stt_transcripts",
-        "_gateway_pending_stt_echo_sent",
     ):
         if hasattr(event, attr):
             delattr(event, attr)

gateway/run.py

@@ -18200,16 +18200,28 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         metadata=None,
         log_context: str = "Transcript",
     ) -> None:
-        """Echo pending-event STT transcripts to the chat at most once."""
+        """Echo pending-event STT transcripts to the chat at most once.
+
+        The already-echoed transcripts are tracked as a COUNT rather than a
+        single boolean.  ``merge_pending_message_event`` can append a second
+        voice note to an event whose first transcript was already echoed and
+        invalidates the transcription cache; the re-run transcription then
+        returns the earlier transcripts as a prefix of the new list, so
+        echoing only the unsent tail suppresses the repeat while still
+        surfacing the newly merged note.  A count rather than a set of seen
+        values because two separate notes that transcribe identically are two
+        distinct deliveries and both must be echoed.
+        """
         if (
             not transcripts
             or not self._should_echo_stt_transcripts()
             or adapter is None
-            or getattr(event, "_gateway_pending_stt_echo_sent", False)
         ):
             return
-        setattr(event, "_gateway_pending_stt_echo_sent", True)
-        for tx in transcripts:
+        already_echoed = int(getattr(event, "_gateway_pending_stt_echoed", 0) or 0)
+        unsent = transcripts[already_echoed:]
+        setattr(event, "_gateway_pending_stt_echoed", already_echoed + len(unsent))
+        for tx in unsent:
             try:
                 await adapter.send(
                     source.chat_id,

tests/gateway/test_telegram_voice_v0_regressions.py

@@ -357,3 +357,135 @@ async def test_voice_tts_is_explicit_audio_reply_opt_in():
     assert runner._voice_mode["telegram:12345"] == "all"
     assert "12345" in adapter._auto_tts_enabled_chats
     assert result
+
+
+def _voice_event(source, urls):
+    return MessageEvent(
+        text="",
+        message_type=MessageType.VOICE,
+        source=source,
+        media_urls=list(urls),
+        media_types=["audio/ogg"] * len(urls),
+    )
+
+
+@pytest.mark.asyncio
+async def test_pending_stt_merge_does_not_re_echo_delivered_transcript():
+    """A follow-up message must not replay an already-echoed transcript.
+
+    ``merge_pending_message_event`` invalidates the transcription cache so the
+    merged text/media is picked up, which makes the drain path transcribe
+    again.  The echo ledger has to survive that invalidation, otherwise the
+    user sees the same 🎙️ line twice.
+    """
+    from gateway.platforms.base import merge_pending_message_event
+
+    adapter = SimpleNamespace(send=AsyncMock())
+    runner = _runner(adapter)
+    source = _source()
+    event = _voice_event(source, ["/tmp/voice-1.ogg"])
+
+    with patch(
+        "tools.transcription_tools.transcribe_audio",
+        return_value={"success": True, "transcript": "hello", "provider": "mock"},
+    ):
+        _, transcripts = await runner._transcribe_pending_audio_event_once(event, event.text)
+        await runner._echo_pending_stt_transcripts_once(event, adapter, source, transcripts)
+
+        # A plain text follow-up merges into the still-pending voice event.
+        merge_pending_message_event(
+            {"telegram:dm:12345": event},
+            "telegram:dm:12345",
+            MessageEvent(text="and also this", message_type=MessageType.TEXT, source=source),
+        )
+
+        drain_text, drain_transcripts = await runner._transcribe_pending_audio_event_once(
+            event, event.text
+        )
+        await runner._echo_pending_stt_transcripts_once(
+            event, adapter, source, drain_transcripts
+        )
+
+    assert event.text == "and also this"
+    assert "and also this" in drain_text
+    adapter.send.assert_awaited_once_with("12345", '🎙️ "hello"', metadata=None)
+
+
+@pytest.mark.asyncio
+async def test_pending_stt_merge_echoes_only_the_newly_merged_transcript():
+    """A second voice note still gets echoed, without repeating the first."""
+    from gateway.platforms.base import merge_pending_message_event
+
+    adapter = SimpleNamespace(send=AsyncMock())
+    runner = _runner(adapter)
+    source = _source()
+    event = _voice_event(source, ["/tmp/voice-1.ogg"])
+
+    def _fake_transcribe(path):
+        name = "hello" if path.endswith("voice-1.ogg") else "world"
+        return {"success": True, "transcript": name, "provider": "mock"}
+
+    with patch("tools.transcription_tools.transcribe_audio", side_effect=_fake_transcribe):
+        _, transcripts = await runner._transcribe_pending_audio_event_once(event, event.text)
+        await runner._echo_pending_stt_transcripts_once(event, adapter, source, transcripts)
+
+        merge_pending_message_event(
+            {"telegram:dm:12345": event},
+            "telegram:dm:12345",
+            _voice_event(source, ["/tmp/voice-2.ogg"]),
+        )
+
+        _, drain_transcripts = await runner._transcribe_pending_audio_event_once(
+            event, event.text
+        )
+        await runner._echo_pending_stt_transcripts_once(
+            event, adapter, source, drain_transcripts
+        )
+
+    assert drain_transcripts == ["hello", "world"]
+    assert [c.args[1] for c in adapter.send.await_args_list] == [
+        '🎙️ "hello"',
+        '🎙️ "world"',
+    ]
+
+
+@pytest.mark.asyncio
+async def test_pending_stt_merge_echoes_two_identical_transcripts():
+    """Two separate notes that transcribe identically are two deliveries.
+
+    The ledger counts what was already echoed rather than remembering the
+    transcript strings: a value-based dedup would silently collapse a repeated
+    phrase into one echo, dropping a note the user actually sent.
+    """
+    from gateway.platforms.base import merge_pending_message_event
+
+    adapter = SimpleNamespace(send=AsyncMock())
+    runner = _runner(adapter)
+    source = _source()
+    event = _voice_event(source, ["/tmp/voice-1.ogg"])
+
+    with patch(
+        "tools.transcription_tools.transcribe_audio",
+        return_value={"success": True, "transcript": "on my way", "provider": "mock"},
+    ):
+        _, transcripts = await runner._transcribe_pending_audio_event_once(event, event.text)
+        await runner._echo_pending_stt_transcripts_once(event, adapter, source, transcripts)
+
+        merge_pending_message_event(
+            {"telegram:dm:12345": event},
+            "telegram:dm:12345",
+            _voice_event(source, ["/tmp/voice-2.ogg"]),
+        )
+
+        _, drain_transcripts = await runner._transcribe_pending_audio_event_once(
+            event, event.text
+        )
+        await runner._echo_pending_stt_transcripts_once(
+            event, adapter, source, drain_transcripts
+        )
+
+    assert drain_transcripts == ["on my way", "on my way"]
+    assert [c.args[1] for c in adapter.send.await_args_list] == [
+        '🎙️ "on my way"',
+        '🎙️ "on my way"',
+    ], "the second note must still be echoed even though it transcribes the same"