From 753d0d77e30bf6f6e8f6b5083cd75e6686a4207f Mon Sep 17 00:00:00 2001 From: Frowtek Date: Sun, 19 Jul 2026 16:19:14 +0300 Subject: [PATCH] fix(gateway): keep the STT echo ledger across pending-media merges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _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 | 7 +- gateway/run.py | 20 ++- .../test_telegram_voice_v0_regressions.py | 132 ++++++++++++++++++ 3 files changed, 154 insertions(+), 5 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 7e5a93a1d155..a7c3ea9fa609 100644 --- a/gateway/platforms/base.py +++ b/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) diff --git a/gateway/run.py b/gateway/run.py index 79795a49fb4e..a8b34c74163e 100644 --- a/gateway/run.py +++ b/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, diff --git a/tests/gateway/test_telegram_voice_v0_regressions.py b/tests/gateway/test_telegram_voice_v0_regressions.py index 43ac41629062..523f6467a5c3 100644 --- a/tests/gateway/test_telegram_voice_v0_regressions.py +++ b/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"