mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(tts): route streaming-provider secrets through resolve_provider_secret; bound per-sentence stream bodies at 16 MiB
Follow-up integration for the #47588 salvage, aligning the new streamers with the post-campaign invariants: - All streaming key lookups go through _resolve_key -> tts_tool. _resolve_provider_key -> resolve_provider_secret (config > env/.env > credential pool, profile-scoped) — never bare get_env_value. xAI resolves via resolve_xai_http_credentials so OAuth users stream too. - _capped(): every provider's chunk iterator is bounded at 16 MiB per sentence, mirroring _read_tts_response_bytes' bounded-upstream-body invariant on the sync paths. - Tests updated for the resolver contract + new coverage for credential routing and the cap.
This commit is contained in:
Teknium
parent
bc4dcb1b02
commit
7800bb1a29
2 changed files with 122 additions and 19 deletions
|
|
@ -124,9 +124,11 @@ def test_never_swaps_provider_for_streaming(monkeypatch):
|
|||
|
||||
|
||||
def test_elevenlabs_available_reflects_key(monkeypatch):
|
||||
monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "key" if k == "ELEVENLABS_API_KEY" else None)
|
||||
# Key lookups now route through the provider-secret resolver
|
||||
# (config > env/.env > credential pool), not bare get_env_value.
|
||||
monkeypatch.setattr(ts, "_resolve_key", lambda env, pid: "key" if env == "ELEVENLABS_API_KEY" else "")
|
||||
assert ts.ElevenLabsStreamer.available() is True
|
||||
monkeypatch.setattr(ts, "get_env_value", lambda k, *a: None)
|
||||
monkeypatch.setattr(ts, "_resolve_key", lambda env, pid: "")
|
||||
assert ts.ElevenLabsStreamer.available() is False
|
||||
|
||||
|
||||
|
|
@ -384,6 +386,41 @@ def test_streaming_provider_auto_none_when_nothing_usable(monkeypatch):
|
|||
) is None
|
||||
|
||||
|
||||
# ── Credential routing: resolve_provider_secret, never bare env ──────────
|
||||
|
||||
|
||||
def test_elevenlabs_available_routes_through_secret_resolver(monkeypatch):
|
||||
calls = []
|
||||
|
||||
def _fake_resolve(env_var, provider_id):
|
||||
calls.append((env_var, provider_id))
|
||||
return "pool-key"
|
||||
|
||||
monkeypatch.setattr(ts, "_resolve_key", _fake_resolve)
|
||||
assert ts.ElevenLabsStreamer.available() is True
|
||||
assert ("ELEVENLABS_API_KEY", "elevenlabs") in calls
|
||||
|
||||
|
||||
def test_gemini_available_falls_back_to_google_key(monkeypatch):
|
||||
keys = {"GOOGLE_API_KEY": "g-key"}
|
||||
monkeypatch.setattr(ts, "_resolve_key", lambda env, pid: keys.get(env, ""))
|
||||
assert ts.GeminiStreamer.available() is True
|
||||
keys.clear()
|
||||
assert ts.GeminiStreamer.available() is False
|
||||
|
||||
|
||||
def test_xai_available_uses_oauth_credential_resolver(monkeypatch):
|
||||
import sys
|
||||
import types
|
||||
|
||||
fake = types.ModuleType("tools.xai_http")
|
||||
fake.resolve_xai_http_credentials = lambda: {"api_key": "xai-key"}
|
||||
monkeypatch.setitem(sys.modules, "tools.xai_http", fake)
|
||||
assert ts.XAIStreamer.available() is True
|
||||
fake.resolve_xai_http_credentials = lambda: {"api_key": ""}
|
||||
assert ts.XAIStreamer.available() is False
|
||||
|
||||
|
||||
# ── Gemini SSE parsing ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
@ -429,7 +466,7 @@ def test_gemini_streamer_decodes_sse_pcm_chunks(monkeypatch):
|
|||
fake_requests = types.ModuleType("requests")
|
||||
fake_requests.post = _post
|
||||
monkeypatch.setitem(sys.modules, "requests", fake_requests)
|
||||
monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "g-key" if k in ("GEMINI_API_KEY", "GOOGLE_API_KEY") else None)
|
||||
monkeypatch.setattr(ts, "_resolve_key", lambda env, pid: "g-key")
|
||||
|
||||
streamer = ts.GeminiStreamer({}, {"voice": "Kore"})
|
||||
assert list(streamer.stream("Hello there.")) == [pcm1, pcm2]
|
||||
|
|
@ -447,3 +484,18 @@ def test_xai_streamer_yields_collected_frames(monkeypatch):
|
|||
streamer.tts_config, streamer.section = {}, {}
|
||||
monkeypatch.setattr(streamer, "_collect_async", lambda text: list(frames))
|
||||
assert list(streamer.stream("A sentence.")) == frames
|
||||
|
||||
|
||||
# ── 16 MiB per-sentence stream cap ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_stream_cap_truncates_runaway_upstream(monkeypatch):
|
||||
monkeypatch.setattr(ts, "_STREAM_SENTENCE_BYTE_CAP", 100)
|
||||
|
||||
def _endless():
|
||||
while True:
|
||||
yield b"\x00" * 64
|
||||
|
||||
out = list(ts._capped(_endless(), "test"))
|
||||
assert len(out) == 1 # 64 ok, 128 > cap → stop
|
||||
assert sum(len(c) for c in out) <= 100
|
||||
|
|
|
|||
|
|
@ -32,6 +32,27 @@ from tools.tts_tool import _get_provider, _load_tts_config, get_env_value
|
|||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Upper bound on the PCM bytes accepted from one provider stream for one
|
||||
# sentence. Mirrors the 16 MiB bounded-upstream-body invariant of the sync
|
||||
# providers (``_read_tts_response_bytes`` in tools.tts_tool): a buggy or
|
||||
# hostile endpoint must not be able to feed us unbounded audio.
|
||||
_STREAM_SENTENCE_BYTE_CAP = 16 * 1024 * 1024
|
||||
|
||||
|
||||
def _resolve_key(env_var: str, provider_id: str) -> str:
|
||||
"""Provider secret lookup: config > env/.env > credential pool.
|
||||
|
||||
Thin, monkeypatchable seam over ``tools.tts_tool._resolve_provider_key``
|
||||
(which delegates to ``resolve_provider_secret``). ALL streaming-provider
|
||||
key lookups go through here — never bare ``get_env_value``.
|
||||
"""
|
||||
try:
|
||||
from tools.tts_tool import _resolve_provider_key
|
||||
|
||||
return _resolve_provider_key(env_var, provider_id) or ""
|
||||
except Exception:
|
||||
return get_env_value(env_var) or ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Interruption latch — lets the model know it was cut off mid-speech
|
||||
|
|
@ -204,7 +225,7 @@ class ElevenLabsStreamer(StreamingTTSProvider):
|
|||
|
||||
@staticmethod
|
||||
def available() -> bool:
|
||||
return bool(get_env_value("ELEVENLABS_API_KEY"))
|
||||
return bool(_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"))
|
||||
|
||||
def stream(self, text: str) -> Iterator[bytes]:
|
||||
from tools.tts_tool import (
|
||||
|
|
@ -215,7 +236,7 @@ class ElevenLabsStreamer(StreamingTTSProvider):
|
|||
)
|
||||
|
||||
client = _import_elevenlabs()(
|
||||
api_key=get_env_value("ELEVENLABS_API_KEY"),
|
||||
api_key=_resolve_key("ELEVENLABS_API_KEY", "elevenlabs"),
|
||||
**_elevenlabs_environment_kwargs(self.section),
|
||||
)
|
||||
voice_id = self.section.get("voice_id", DEFAULT_ELEVENLABS_VOICE_ID)
|
||||
|
|
@ -269,16 +290,33 @@ class OpenAIStreamer(StreamingTTSProvider):
|
|||
input=text,
|
||||
response_format="pcm",
|
||||
) as response:
|
||||
yield from response.iter_bytes()
|
||||
yield from _capped(response.iter_bytes(), "OpenAI streaming TTS")
|
||||
|
||||
|
||||
def _capped(chunks: Iterator[bytes], label: str) -> Iterator[bytes]:
|
||||
"""Pass chunks through, aborting past the 16 MiB per-sentence cap.
|
||||
|
||||
The streaming mirror of ``_read_tts_response_bytes``'s bounded-body
|
||||
invariant: one sentence of PCM should never approach the cap, so
|
||||
exceeding it means a runaway/hostile upstream — stop pulling.
|
||||
"""
|
||||
total = 0
|
||||
for chunk in chunks:
|
||||
total += len(chunk)
|
||||
if total > _STREAM_SENTENCE_BYTE_CAP:
|
||||
logger.warning("%s exceeded %d bytes for one sentence; truncating",
|
||||
label, _STREAM_SENTENCE_BYTE_CAP)
|
||||
return
|
||||
yield chunk
|
||||
|
||||
|
||||
@register("gemini")
|
||||
class GeminiStreamer(StreamingTTSProvider):
|
||||
"""Gemini ``streamGenerateContent?alt=sse`` → base64 PCM chunks (24 kHz).
|
||||
|
||||
Each SSE event carries one base64-encoded PCM chunk under
|
||||
``candidates[0].content.parts[*].inlineData.data``; we decode and yield
|
||||
each one as it arrives.
|
||||
Salvaged from PR #47588 (@Cdddo) and rebased onto the post-campaign
|
||||
infrastructure: credentials via the provider-secret resolver, requests
|
||||
(not httpx) with a bounded streamed body, and main's provider ABC.
|
||||
"""
|
||||
|
||||
sample_rate = 24000
|
||||
|
|
@ -286,7 +324,8 @@ class GeminiStreamer(StreamingTTSProvider):
|
|||
@staticmethod
|
||||
def available() -> bool:
|
||||
return bool(
|
||||
get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY")
|
||||
_resolve_key("GEMINI_API_KEY", "gemini")
|
||||
or _resolve_key("GOOGLE_API_KEY", "gemini")
|
||||
)
|
||||
|
||||
def stream(self, text: str) -> Iterator[bytes]:
|
||||
|
|
@ -302,7 +341,8 @@ class GeminiStreamer(StreamingTTSProvider):
|
|||
)
|
||||
|
||||
api_key = (
|
||||
get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY")
|
||||
_resolve_key("GEMINI_API_KEY", "gemini")
|
||||
or _resolve_key("GOOGLE_API_KEY", "gemini")
|
||||
)
|
||||
model = str(self.section.get("model", DEFAULT_GEMINI_TTS_MODEL)).strip() or DEFAULT_GEMINI_TTS_MODEL
|
||||
voice = str(self.section.get("voice", DEFAULT_GEMINI_TTS_VOICE)).strip() or DEFAULT_GEMINI_TTS_VOICE
|
||||
|
|
@ -354,26 +394,35 @@ class GeminiStreamer(StreamingTTSProvider):
|
|||
except (ValueError, TypeError) as exc:
|
||||
logger.warning("Gemini SSE: bad base64 audio: %s", exc)
|
||||
|
||||
yield from _sse_chunks()
|
||||
yield from _capped(_sse_chunks(), "Gemini streaming TTS")
|
||||
|
||||
|
||||
@register("xai")
|
||||
class XAIStreamer(StreamingTTSProvider):
|
||||
"""xAI WebSocket TTS → binary PCM frames (24 kHz mono int16).
|
||||
|
||||
xAI's chunked TTS API is WebSocket-only (``wss://api.x.ai/v1/tts``).
|
||||
The async WS loop is bridged to the sync iterator contract via
|
||||
``_collect_async`` — the seam unit tests monkeypatch.
|
||||
Salvaged from PR #47588 (@Cdddo): xAI's chunked TTS API is
|
||||
WebSocket-only (``wss://api.x.ai/v1/tts``). Credentials route through
|
||||
``resolve_xai_http_credentials`` (OAuth or XAI_API_KEY), same as the
|
||||
sync ``_generate_xai_tts`` path. The async WS loop is bridged to the
|
||||
sync iterator contract via ``_collect_async`` — the seam unit tests
|
||||
monkeypatch.
|
||||
"""
|
||||
|
||||
sample_rate = 24000
|
||||
|
||||
@staticmethod
|
||||
def available() -> bool:
|
||||
return bool(get_env_value("XAI_API_KEY"))
|
||||
try:
|
||||
from tools.xai_http import resolve_xai_http_credentials
|
||||
|
||||
creds = resolve_xai_http_credentials()
|
||||
return bool(str(creds.get("api_key") or "").strip())
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def stream(self, text: str) -> Iterator[bytes]:
|
||||
yield from self._collect_async(text)
|
||||
yield from _capped(iter(self._collect_async(text)), "xAI streaming TTS")
|
||||
|
||||
# -- async→sync bridge (test seam) ------------------------------------
|
||||
|
||||
|
|
@ -394,10 +443,12 @@ class XAIStreamer(StreamingTTSProvider):
|
|||
import websockets
|
||||
|
||||
from tools.tts_tool import DEFAULT_XAI_VOICE_ID
|
||||
from tools.xai_http import resolve_xai_http_credentials
|
||||
|
||||
api_key = (get_env_value("XAI_API_KEY") or "").strip()
|
||||
creds = resolve_xai_http_credentials()
|
||||
api_key = str(creds.get("api_key") or "").strip()
|
||||
if not api_key:
|
||||
raise RuntimeError("XAI_API_KEY not set; cannot use xAI streaming TTS")
|
||||
raise RuntimeError("No xAI credentials for streaming TTS")
|
||||
voice = str(self.section.get("voice_id", DEFAULT_XAI_VOICE_ID)).strip() or DEFAULT_XAI_VOICE_ID
|
||||
ws_url = str(
|
||||
self.section.get("streaming_url") or "wss://api.x.ai/v1/tts"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue