mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-29 18:46:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(whatsapp): preserve voice notes when STT fails

Browse source
This commit is contained in:
Vivaan Dhawan 2026-07-18 06:33:15 +05:30 • committed by Teknium
parent afab7ed46e
commit 1c30c57f11
6 changed files with 174 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

32
gateway/run.py
View file

@ -18087,7 +18087,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
return prefix, []
try:
from tools.transcription_tools import transcribe_audio
from tools.transcription_tools import (
transcribe_audio,
transcribe_audio_local_fallback,
)
except ModuleNotFoundError as e:
logger.error("Transcription module unavailable: %s", e)
unavailable_note = "[voice message could not be transcribed]"
@ -18104,6 +18107,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
try:
logger.debug("Transcribing user voice: %s", path)
result = await asyncio.to_thread(transcribe_audio, path)
if not result.get("success"):
fallback = await asyncio.to_thread(
transcribe_audio_local_fallback,
path,
)
if fallback.get("success"):
logger.info(
"Configured STT failed for %s; recovered with local STT",
path,
)
result = fallback
if result["success"]:
transcript = result["transcript"]
# Speech-to-text can return success=True with an empty or
@ -18138,10 +18152,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# logged for operator diagnosis but kept out of the
# LLM-visible prompt.
logger.info("Voice transcription failed for %s: %s", path, error)
enriched_parts.append("[voice message could not be transcribed]")
from tools.credential_files import to_agent_visible_cache_path
agent_path = to_agent_visible_cache_path(os.path.abspath(path))
enriched_parts.append(
"[voice message could not be transcribed automatically; "
f"the audio is available at: {agent_path}]"
)
except Exception as e:
logger.error("Transcription error: %s", e)
enriched_parts.append("[voice message could not be transcribed]")
from tools.credential_files import to_agent_visible_cache_path
agent_path = to_agent_visible_cache_path(os.path.abspath(path))
enriched_parts.append(
"[voice message could not be transcribed automatically; "
f"the audio is available at: {agent_path}]"
)
if enriched_parts:
prefix = "\n\n".join(enriched_parts)

9
plugins/platforms/whatsapp/adapter.py
View file

@ -1484,6 +1484,15 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
body = data.get("body", "")
if data.get("isGroup"):
body = self._clean_bot_mention_text(body, data)
if (
msg_type == MessageType.VOICE
and cached_urls
and str(body).strip().lower() == "[ptt received]"
):
# The bridge synthesizes this placeholder for captionless voice
# notes. The cached audio is the real payload; retaining the
# placeholder makes the agent answer it as if it were user text.
body = ""
# If this is a reply, keep the quoted message in structured fields
# only. GatewayRunner._prepare_inbound_message_text owns rendering

34
tests/gateway/test_stt_config.py
View file

@ -90,6 +90,9 @@ async def test_enrich_message_with_transcription_avoids_bogus_no_provider_messag
with patch(
"tools.transcription_tools.transcribe_audio",
return_value={"success": False, "error": "VOICE_TOOLS_OPENAI_KEY not set"},
), patch(
"tools.transcription_tools.transcribe_audio_local_fallback",
return_value={"success": False, "error": "not installed"},
):
result, transcripts = await runner._enrich_message_with_transcription(
"caption",
@ -97,13 +100,42 @@ async def test_enrich_message_with_transcription_avoids_bogus_no_provider_messag
)
assert "No STT provider is configured" not in result
assert "[voice message could not be transcribed]" in result
assert "voice message could not be transcribed automatically" in result
assert "/tmp/voice.ogg" in result
# The opaque backend cause must NOT leak into the LLM-visible prompt.
assert "VOICE_TOOLS_OPENAI_KEY" not in result
assert "caption" in result
assert transcripts == []
@pytest.mark.asyncio
async def test_enrich_message_with_transcription_falls_back_to_installed_local_stt():
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner.config = GatewayConfig(stt_enabled=True)
with patch(
"tools.transcription_tools.transcribe_audio",
return_value={"success": False, "error": "configured provider unavailable"},
), patch(
"tools.transcription_tools.transcribe_audio_local_fallback",
return_value={
"success": True,
"transcript": "recovered locally",
"provider": "local",
},
) as local_fallback:
result, transcripts = await runner._enrich_message_with_transcription(
"",
["/tmp/voice.ogg"],
)
assert result == '"recovered locally"'
assert transcripts == ["recovered locally"]
local_fallback.assert_called_once_with("/tmp/voice.ogg")
@pytest.mark.asyncio
async def test_enrich_message_with_transcription_returns_tuple_for_empty_content_placeholder():
"""A successful transcription whose caption is the empty-content placeholder

29
tests/gateway/test_whatsapp_formatting.py
View file

@ -344,6 +344,35 @@ class TestBridgeEventMetadata:
assert event.raw_message["quotedRemoteJid"] == "15551234567@s.whatsapp.net"
assert event.raw_message["hasQuotedMessage"] is True
@pytest.mark.asyncio
async def test_captionless_voice_note_drops_bridge_placeholder(self, tmp_path, monkeypatch):
adapter = _make_adapter()
voice_path = tmp_path / "aud_voice.ogg"
voice_path.write_bytes(b"fake audio")
monkeypatch.setattr(
"plugins.platforms.whatsapp.adapter._is_allowed_bridge_path",
lambda path: path == str(voice_path),
)
data = {
"messageId": "voice-msg",
"chatId": "15551234567@s.whatsapp.net",
"senderId": "15551234567@s.whatsapp.net",
"senderName": "Tester",
"chatName": "Tester",
"isGroup": False,
"body": "[ptt received]",
"hasMedia": True,
"mediaType": "ptt",
"mime": "audio/ogg",
"mediaUrls": [str(voice_path)],
}
event = await adapter._build_message_event(data)
assert event is not None
assert event.text == ""
assert event.media_urls == [str(voice_path)]
# ---------------------------------------------------------------------------
# display_config tier classification

38
tests/tools/test_transcription.py
View file

@ -380,6 +380,44 @@ class TestTranscribeAudio:
assert "not found" in result["error"]
class TestLocalFallback:
def test_uses_installed_faster_whisper_without_changing_provider(self, tmp_path):
audio_file = tmp_path / "test.ogg"
audio_file.write_bytes(b"fake audio")
with patch(
"tools.transcription_tools._load_stt_config",
return_value={"provider": "openai", "local": {"model": "small"}},
), patch(
"tools.transcription_tools._HAS_FASTER_WHISPER",
True,
), patch(
"tools.transcription_tools._transcribe_local",
return_value={"success": True, "transcript": "local result"},
) as mock_local:
from tools.transcription_tools import transcribe_audio_local_fallback
result = transcribe_audio_local_fallback(str(audio_file))
assert result["transcript"] == "local result"
mock_local.assert_called_once_with(str(audio_file), "small")
def test_does_not_install_when_no_local_backend_exists(self, tmp_path):
audio_file = tmp_path / "test.ogg"
audio_file.write_bytes(b"fake audio")
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", False), patch(
"tools.transcription_tools._has_local_command", return_value=False
):
from tools.transcription_tools import transcribe_audio_local_fallback
result = transcribe_audio_local_fallback(str(audio_file))
assert result["success"] is False
assert "installed local STT" in result["error"]
# ---------------------------------------------------------------------------
# Model name normalisation for local providers
# ---------------------------------------------------------------------------

36
tools/transcription_tools.py
View file

@ -2326,6 +2326,42 @@ def _is_local_or_private_url(url: str) -> bool:
return False
def transcribe_audio_local_fallback(
file_path: str,
model: Optional[str] = None,
) -> Dict[str, Any]:
"""Try an already-installed local STT backend without changing config.
This is intended for passive inbound-media recovery after the configured
provider has failed. It deliberately does not lazy-install dependencies or
fall through to another cloud provider.
"""
error = _validate_audio_file(file_path)
if error:
return error
stt_config = _load_stt_config()
local_cfg = stt_config.get("local") or {}
local_model = model or local_cfg.get("model", DEFAULT_LOCAL_MODEL)
if _HAS_FASTER_WHISPER:
return _transcribe_local(
file_path,
_normalize_local_model(local_model),
)
if _has_local_command():
return _transcribe_local_command(
file_path,
_normalize_local_command_model(local_model),
)
return {
"success": False,
"transcript": "",
"error": "No installed local STT backend is available.",
"provider": "local",
}
def _resolve_openai_audio_client_config() -> tuple[str, str]:
"""Return direct OpenAI audio config or a managed gateway fallback."""
stt_config = _load_stt_config()