Skip to content
Merged
6 changes: 5 additions & 1 deletion agent/context_compressor.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,11 @@ def _generate_summary(self, turns_to_summarize: List[Dict[str, Any]]) -> Optiona
if self.summary_model:
call_kwargs["model"] = self.summary_model
response = call_llm(**call_kwargs)
summary = response.choices[0].message.content.strip()
content = response.choices[0].message.content
# Handle cases where content is not a string (e.g., dict from llama.cpp)
if not isinstance(content, str):
content = str(content) if content else ""
summary = content.strip()
if not summary.startswith("[CONTEXT SUMMARY]:"):
summary = "[CONTEXT SUMMARY]: " + summary
return summary
Expand Down
7 changes: 5 additions & 2 deletions gateway/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,10 +83,13 @@ def to_dict(self) -> Dict[str, Any]:

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SessionResetPolicy":
# Handle both missing keys and explicit null values (YAML null → None)
at_hour = data.get("at_hour")
idle_minutes = data.get("idle_minutes")
return cls(
mode=data.get("mode", "both"),
at_hour=data.get("at_hour", 4),
idle_minutes=data.get("idle_minutes", 1440),
at_hour=at_hour if at_hour is not None else 4,
idle_minutes=idle_minutes if idle_minutes is not None else 1440,
)


Expand Down
9 changes: 7 additions & 2 deletions hermes_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,8 +194,13 @@ def ensure_hermes_home():
},

"stt": {
"enabled": True,
"model": "whisper-1",
"provider": "local", # "local" (free, faster-whisper) | "openai" (Whisper API)
"local": {
"model": "base", # tiny, base, small, medium, large-v3
},
"openai": {
"model": "whisper-1", # whisper-1, gpt-4o-mini-transcribe, gpt-4o-transcribe
},
},

"human_delay": {
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ dependencies = [
"fal-client",
# Text-to-speech (Edge TTS is free, no API key needed)
"edge-tts",
"faster-whisper>=1.0.0",
# mini-swe-agent deps (terminal tool)
"litellm>=1.75.5",
"typer",
Expand Down
2 changes: 1 addition & 1 deletion run_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -2729,7 +2729,7 @@ def _build_api_kwargs(self, api_messages: list) -> dict:
"model": self.model,
"messages": api_messages,
"tools": self.tools if self.tools else None,
"timeout": 900.0,
"timeout": float(os.getenv("HERMES_API_TIMEOUT", 900.0)),
}

if self.max_tokens is not None:
Expand Down
41 changes: 41 additions & 0 deletions tests/agent/test_context_compressor.py
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,47 @@ def test_none_content_in_system_message_compress(self):
assert len(result) < len(msgs)


class TestNonStringContent:
"""Regression: content as dict (e.g., llama.cpp tool calls) must not crash."""

def test_dict_content_coerced_to_string(self):
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = {"text": "some summary"}

with patch("agent.context_compressor.get_model_context_length", return_value=100000):
c = ContextCompressor(model="test", quiet_mode=True)

messages = [
{"role": "user", "content": "do something"},
{"role": "assistant", "content": "ok"},
]

with patch("agent.context_compressor.call_llm", return_value=mock_response):
summary = c._generate_summary(messages)
assert isinstance(summary, str)
assert "CONTEXT SUMMARY" in summary

def test_none_content_coerced_to_empty(self):
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = None

with patch("agent.context_compressor.get_model_context_length", return_value=100000):
c = ContextCompressor(model="test", quiet_mode=True)

messages = [
{"role": "user", "content": "do something"},
{"role": "assistant", "content": "ok"},
]

with patch("agent.context_compressor.call_llm", return_value=mock_response):
summary = c._generate_summary(messages)
# None content → empty string → "[CONTEXT SUMMARY]: " prefix added
assert summary is not None
assert "CONTEXT SUMMARY" in summary


class TestCompressWithClient:
def test_summarization_path(self):
mock_client = MagicMock()
Expand Down
223 changes: 223 additions & 0 deletions tests/tools/test_transcription.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,223 @@
"""Tests for transcription_tools.py — local (faster-whisper) and OpenAI providers.

Tests cover provider selection, config loading, validation, and transcription
dispatch. All external dependencies (faster_whisper, openai) are mocked.
"""

import json
import os
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch, mock_open

import pytest


# ---------------------------------------------------------------------------
# Provider selection
# ---------------------------------------------------------------------------


class TestGetProvider:
"""_get_provider() picks the right backend based on config + availability."""

def test_local_when_available(self):
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", True):
from tools.transcription_tools import _get_provider
assert _get_provider({"provider": "local"}) == "local"

def test_local_fallback_to_openai(self, monkeypatch):
monkeypatch.setenv("VOICE_TOOLS_OPENAI_KEY", "sk-test")
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", False), \
patch("tools.transcription_tools._HAS_OPENAI", True):
from tools.transcription_tools import _get_provider
assert _get_provider({"provider": "local"}) == "openai"

def test_local_nothing_available(self, monkeypatch):
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", False), \
patch("tools.transcription_tools._HAS_OPENAI", False):
from tools.transcription_tools import _get_provider
assert _get_provider({"provider": "local"}) == "none"

def test_openai_when_key_set(self, monkeypatch):
monkeypatch.setenv("VOICE_TOOLS_OPENAI_KEY", "sk-test")
with patch("tools.transcription_tools._HAS_OPENAI", True):
from tools.transcription_tools import _get_provider
assert _get_provider({"provider": "openai"}) == "openai"

def test_openai_fallback_to_local(self, monkeypatch):
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", True), \
patch("tools.transcription_tools._HAS_OPENAI", True):
from tools.transcription_tools import _get_provider
assert _get_provider({"provider": "openai"}) == "local"

def test_default_provider_is_local(self):
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", True):
from tools.transcription_tools import _get_provider
assert _get_provider({}) == "local"


# ---------------------------------------------------------------------------
# File validation
# ---------------------------------------------------------------------------


class TestValidateAudioFile:

def test_missing_file(self, tmp_path):
from tools.transcription_tools import _validate_audio_file
result = _validate_audio_file(str(tmp_path / "nope.ogg"))
assert result is not None
assert "not found" in result["error"]

def test_unsupported_format(self, tmp_path):
f = tmp_path / "test.xyz"
f.write_bytes(b"data")
from tools.transcription_tools import _validate_audio_file
result = _validate_audio_file(str(f))
assert result is not None
assert "Unsupported" in result["error"]

def test_valid_file_returns_none(self, tmp_path):
f = tmp_path / "test.ogg"
f.write_bytes(b"fake audio data")
from tools.transcription_tools import _validate_audio_file
assert _validate_audio_file(str(f)) is None

def test_too_large(self, tmp_path):
import stat as stat_mod
f = tmp_path / "big.ogg"
f.write_bytes(b"x")
from tools.transcription_tools import _validate_audio_file, MAX_FILE_SIZE
real_stat = f.stat()
with patch.object(type(f), "stat", return_value=os.stat_result((
real_stat.st_mode, real_stat.st_ino, real_stat.st_dev,
real_stat.st_nlink, real_stat.st_uid, real_stat.st_gid,
MAX_FILE_SIZE + 1, # st_size
real_stat.st_atime, real_stat.st_mtime, real_stat.st_ctime,
))):
result = _validate_audio_file(str(f))
assert result is not None
assert "too large" in result["error"]


# ---------------------------------------------------------------------------
# Local transcription
# ---------------------------------------------------------------------------


class TestTranscribeLocal:

def test_successful_transcription(self, tmp_path):
audio_file = tmp_path / "test.ogg"
audio_file.write_bytes(b"fake audio")

mock_segment = MagicMock()
mock_segment.text = "Hello world"
mock_info = MagicMock()
mock_info.language = "en"
mock_info.duration = 2.5

mock_model = MagicMock()
mock_model.transcribe.return_value = ([mock_segment], mock_info)

with patch("tools.transcription_tools._HAS_FASTER_WHISPER", True), \
patch("tools.transcription_tools.WhisperModel", return_value=mock_model), \
patch("tools.transcription_tools._local_model", None):
from tools.transcription_tools import _transcribe_local
result = _transcribe_local(str(audio_file), "base")

assert result["success"] is True
assert result["transcript"] == "Hello world"

def test_not_installed(self):
with patch("tools.transcription_tools._HAS_FASTER_WHISPER", False):
from tools.transcription_tools import _transcribe_local
result = _transcribe_local("/tmp/test.ogg", "base")
assert result["success"] is False
assert "not installed" in result["error"]


# ---------------------------------------------------------------------------
# OpenAI transcription
# ---------------------------------------------------------------------------


class TestTranscribeOpenAI:

def test_no_key(self, monkeypatch):
monkeypatch.delenv("VOICE_TOOLS_OPENAI_KEY", raising=False)
from tools.transcription_tools import _transcribe_openai
result = _transcribe_openai("/tmp/test.ogg", "whisper-1")
assert result["success"] is False
assert "VOICE_TOOLS_OPENAI_KEY" in result["error"]

def test_successful_transcription(self, monkeypatch, tmp_path):
monkeypatch.setenv("VOICE_TOOLS_OPENAI_KEY", "sk-test")
audio_file = tmp_path / "test.ogg"
audio_file.write_bytes(b"fake audio")

mock_client = MagicMock()
mock_client.audio.transcriptions.create.return_value = "Hello from OpenAI"

with patch("tools.transcription_tools._HAS_OPENAI", True), \
patch("tools.transcription_tools.OpenAI", return_value=mock_client):
from tools.transcription_tools import _transcribe_openai
result = _transcribe_openai(str(audio_file), "whisper-1")

assert result["success"] is True
assert result["transcript"] == "Hello from OpenAI"


# ---------------------------------------------------------------------------
# Main transcribe_audio() dispatch
# ---------------------------------------------------------------------------


class TestTranscribeAudio:

def test_dispatches_to_local(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": "local"}), \
patch("tools.transcription_tools._get_provider", return_value="local"), \
patch("tools.transcription_tools._transcribe_local", return_value={"success": True, "transcript": "hi"}) as mock_local:
from tools.transcription_tools import transcribe_audio
result = transcribe_audio(str(audio_file))

assert result["success"] is True
mock_local.assert_called_once()

def test_dispatches_to_openai(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"}), \
patch("tools.transcription_tools._get_provider", return_value="openai"), \
patch("tools.transcription_tools._transcribe_openai", return_value={"success": True, "transcript": "hi"}) as mock_openai:
from tools.transcription_tools import transcribe_audio
result = transcribe_audio(str(audio_file))

assert result["success"] is True
mock_openai.assert_called_once()

def test_no_provider_returns_error(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={}), \
patch("tools.transcription_tools._get_provider", return_value="none"):
from tools.transcription_tools import transcribe_audio
result = transcribe_audio(str(audio_file))

assert result["success"] is False
assert "No STT provider" in result["error"]

def test_invalid_file_returns_error(self):
from tools.transcription_tools import transcribe_audio
result = transcribe_audio("/nonexistent/file.ogg")
assert result["success"] is False
assert "not found" in result["error"]
Loading
Loading