392 lines
20 KiB
Python
392 lines
20 KiB
Python
"""Offline bridge contracts against the real installed LiveKit SDK.
|
|
|
|
Run without deployment fixtures:
|
|
agent/.venv/bin/python -m unittest discover -s tests/unit -p test_hermes_llm.py -v
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from livekit.agents import llm as lk_llm
|
|
from livekit.agents.types import APIConnectOptions
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "agent"))
|
|
from hermes_llm import HermesLLM, HermesLLMError, WAIT_NOTICES, WAIT_NOTICE_SECONDS
|
|
|
|
|
|
def data(delta, **kwargs):
|
|
return ("data: " + json.dumps({"choices": [{"index": 0, "delta": delta, **kwargs}]}, ensure_ascii=False) + "\n\n").encode()
|
|
|
|
|
|
class FakeSSE(httpx.AsyncByteStream):
|
|
def __init__(self, chunks, *, gate=None):
|
|
self.chunks = chunks
|
|
self.gate = gate
|
|
self.closed = False
|
|
self.started = asyncio.Event()
|
|
|
|
async def __aiter__(self):
|
|
self.started.set()
|
|
for chunk in self.chunks:
|
|
yield chunk
|
|
if self.gate is not None:
|
|
await self.gate.wait()
|
|
|
|
async def aclose(self):
|
|
self.closed = True
|
|
|
|
|
|
class HermesLLMTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.progress = []
|
|
self.notices = []
|
|
self.requests = []
|
|
self.clients = []
|
|
|
|
async def asyncTearDown(self):
|
|
for client in self.clients:
|
|
await client.aclose()
|
|
|
|
async def progress_callback(self, event):
|
|
self.progress.append(event)
|
|
|
|
async def notice_callback(self, message):
|
|
self.notices.append(message)
|
|
|
|
def bridge(self, stream, *, status=200, headers=None, **kwargs):
|
|
def handler(request):
|
|
self.requests.append(request)
|
|
return httpx.Response(status, headers=headers or {"content-type": "text/event-stream"}, stream=stream)
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
self.clients.append(client)
|
|
return HermesLLM(
|
|
base_url="https://trusted.hermes/v1/", api_key="secret-key",
|
|
user_id="alice", room_id="stable-room", on_progress=self.progress_callback,
|
|
on_wait_notice=self.notice_callback, http_client=client, **kwargs,
|
|
)
|
|
|
|
def context(self):
|
|
return lk_llm.ChatContext([
|
|
lk_llm.ChatMessage(role="system", content=["EXACT system prompt\nDo not alter."]),
|
|
lk_llm.ChatMessage(role="user", content=["hello", "world"]),
|
|
lk_llm.ChatMessage(role="assistant", content=["Earlier answer"]),
|
|
lk_llm.ChatMessage(role="developer", content=["unsupported"]),
|
|
lk_llm.FunctionCall(call_id="call", name="local_tool", arguments='{"secret":true}'),
|
|
lk_llm.FunctionCallOutput(call_id="call", name="local_tool", output="local secret", is_error=False),
|
|
])
|
|
|
|
async def test_sdk_chunks_and_exact_request_contract(self):
|
|
body = data({"reasoning_content": "PRIVATE"}) + data({"content": "Hi "}) + data({"content": "there"}) + b"data: [DONE]\n\n"
|
|
fake = FakeSSE([body[:9], body[9:25], body[25:]])
|
|
bridge = self.bridge(fake)
|
|
context = self.context()
|
|
before = context.to_dict()
|
|
stream = bridge.chat(chat_ctx=context, tools=[object()], tool_choice="required", extra_kwargs={"model": "wrong", "tools": ["wrong"]})
|
|
async with stream:
|
|
chunks = [chunk async for chunk in stream]
|
|
self.assertTrue(isinstance(stream, lk_llm.LLMStream))
|
|
self.assertEqual([chunk.delta.content for chunk in chunks], ["Hi ", "there"])
|
|
self.assertTrue(all(chunk.delta.role == "assistant" and not chunk.delta.tool_calls for chunk in chunks))
|
|
request = self.requests[0]
|
|
self.assertEqual(str(request.url), "https://trusted.hermes/v1/chat/completions")
|
|
self.assertEqual(request.headers["authorization"], "Bearer secret-key")
|
|
self.assertEqual(request.headers["x-hermes-session-id"], "stable-room")
|
|
self.assertEqual(request.headers["x-hermes-session-key"], "hope:alice")
|
|
self.assertEqual(json.loads(request.content), {
|
|
"model": "auto", "provider": "custom:hope-upstream", "stream": True,
|
|
"messages": [
|
|
{"role": "system", "content": "EXACT system prompt\nDo not alter."},
|
|
{"role": "user", "content": "hello\nworld"},
|
|
{"role": "assistant", "content": "Earlier answer"},
|
|
], "model_options": {"reasoning": {"enabled": False}},
|
|
})
|
|
self.assertEqual(context.to_dict(), before)
|
|
self.assertEqual(stream.tools, [])
|
|
self.assertEqual(self.notices, [])
|
|
self.assertEqual(self.progress[-1]["status"], "completed")
|
|
self.assertEqual(self.progress[-1]["request_id"], chunks[0].id)
|
|
self.assertTrue(fake.closed)
|
|
|
|
async def test_sse_crlf_unicode_multiline_comments_unknown_events_and_quick_tools_silent(self):
|
|
tool = b'event: hermes.tool.progress\r\ndata: {"status":"started",\r\ndata: "tool_name":"search"}\r\n\r\n'
|
|
body = b":keepalive\r\n\r\nevent: unknown\r\ndata: not json\r\n\r\n" + tool + tool + data({"reasoning": "never speak"}) + data({"content": "caf\u00e9"}) + b"data: [DONE]\r\n\r\n"
|
|
fake = FakeSSE([body[i:i + 1] for i in range(len(body))])
|
|
bridge = self.bridge(fake)
|
|
result = await bridge.chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(result.text, "caf\u00e9")
|
|
self.assertEqual(self.notices, [])
|
|
self.assertEqual([event["status"] for event in self.progress], ["working", "working", "completed"])
|
|
self.assertTrue(all(event["type"] == "assistant_progress" for event in self.progress))
|
|
self.assertEqual(len({event["request_id"] for event in self.progress}), 1)
|
|
|
|
async def test_tool_after_answer_does_not_speak_notice(self):
|
|
fake = FakeSSE([data({"content": "Answer"}), b'event: hermes.tool.progress\ndata: {"status":"working","message":"Searching"}\n\n', b"data: [DONE]\n\n"])
|
|
await self.bridge(fake).chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(self.notices, [])
|
|
|
|
async def test_latency_notice_without_network_event(self):
|
|
fake = FakeSSE([], gate=asyncio.Event())
|
|
bridge = self.bridge(fake)
|
|
# Speed up only the twelve-second timer; network remains blocked.
|
|
sleep = asyncio.sleep
|
|
async def fast_sleep(seconds):
|
|
self.assertEqual(seconds, 12.0)
|
|
self.assertEqual(seconds, WAIT_NOTICE_SECONDS)
|
|
await sleep(0)
|
|
with patch("hermes_llm.asyncio.sleep", side_effect=fast_sleep):
|
|
stream = bridge.chat(chat_ctx=self.context())
|
|
await fake.started.wait()
|
|
for _ in range(10):
|
|
await sleep(0)
|
|
if self.notices:
|
|
break
|
|
self.assertEqual(len(self.notices), 1)
|
|
self.assertIn(self.notices[0], WAIT_NOTICES)
|
|
self.assertEqual(self.progress[0]["status"], "working")
|
|
self.assertEqual(self.progress[0]["message"], "Working on your request.")
|
|
await stream.aclose()
|
|
self.assertTrue(fake.closed)
|
|
self.assertEqual(self.progress[-1]["status"], "failed")
|
|
|
|
async def test_status_and_tools_wait_for_threshold_then_notice_once(self):
|
|
threshold = asyncio.Event()
|
|
noticed = asyncio.Event()
|
|
progress_seen = asyncio.Event()
|
|
tool = b'event: hermes.tool.progress\ndata: {"status":"started","tool_name":"search"}\n\n'
|
|
fake = FakeSSE([b'event: hermes.status\ndata: {"status":"working"}\n\n', tool, tool], gate=asyncio.Event())
|
|
bridge = self.bridge(fake)
|
|
|
|
async def timer_sleep(seconds):
|
|
self.assertEqual(seconds, 12.0)
|
|
await threshold.wait()
|
|
|
|
async def progress(event):
|
|
await self.progress_callback(event)
|
|
if len(self.progress) == 3:
|
|
progress_seen.set()
|
|
|
|
async def notice(text):
|
|
await self.notice_callback(text)
|
|
noticed.set()
|
|
|
|
bridge._on_progress = progress
|
|
bridge._on_wait_notice = notice
|
|
with patch("hermes_llm.asyncio.sleep", side_effect=timer_sleep):
|
|
stream = bridge.chat(chat_ctx=self.context())
|
|
try:
|
|
await asyncio.wait_for(progress_seen.wait(), timeout=1)
|
|
self.assertEqual(self.notices, [])
|
|
threshold.set()
|
|
await asyncio.wait_for(noticed.wait(), timeout=1)
|
|
await stream._notice() # repeated progress cannot duplicate it
|
|
self.assertEqual(len(self.notices), 1)
|
|
self.assertIn(self.notices[0], WAIT_NOTICES)
|
|
finally:
|
|
await stream.aclose()
|
|
|
|
async def test_delayed_notices_vary_across_turns_and_stay_out_of_history(self):
|
|
bridge = self.bridge(FakeSSE([], gate=asyncio.Event()))
|
|
sleep = asyncio.sleep
|
|
async def fast_sleep(seconds):
|
|
self.assertEqual(seconds, 12.0)
|
|
await sleep(0)
|
|
|
|
with patch("hermes_llm.asyncio.sleep", side_effect=fast_sleep):
|
|
for turn in range(8):
|
|
noticed = asyncio.Event()
|
|
async def notice(text):
|
|
await self.notice_callback(text)
|
|
noticed.set()
|
|
bridge._on_wait_notice = notice
|
|
context = self.context()
|
|
before = context.to_dict()
|
|
stream = bridge.chat(chat_ctx=context)
|
|
try:
|
|
await asyncio.wait_for(noticed.wait(), timeout=1)
|
|
self.assertEqual(len(self.notices), turn + 1)
|
|
self.assertIn(self.notices[-1], WAIT_NOTICES)
|
|
if turn:
|
|
self.assertNotEqual(self.notices[-1], self.notices[-2])
|
|
self.assertEqual(context.to_dict(), before)
|
|
messages = json.loads(self.requests[-1].content)["messages"]
|
|
self.assertFalse(any(m["content"] in WAIT_NOTICES for m in messages))
|
|
finally:
|
|
await stream.aclose()
|
|
|
|
async def test_timeout_closes_stream_and_never_retries(self):
|
|
fake = FakeSSE([b":keepalive\n\n"], gate=asyncio.Event())
|
|
stream = self.bridge(fake, timeout_seconds=0.02).chat(chat_ctx=self.context(), conn_options=APIConnectOptions(max_retry=10))
|
|
with self.assertRaises(HermesLLMError) as caught:
|
|
await stream.collect()
|
|
self.assertEqual(caught.exception.code, "timeout")
|
|
self.assertFalse(caught.exception.retryable)
|
|
self.assertEqual(len(self.requests), 1)
|
|
self.assertTrue(fake.closed)
|
|
self.assertEqual(self.progress[-1]["status"], "failed")
|
|
|
|
async def test_failures_are_explicit_redacted_and_not_retried(self):
|
|
cases = [
|
|
(401, b"secret-key sensitive payload", "http_401"),
|
|
(503, b"sensitive upstream failure", "http_503"),
|
|
(200, b"data: malformed-secret-key\n\n", "invalid_sse"),
|
|
(200, b'data: {"error":"secret-key"}\n\n', "upstream_error"),
|
|
(200, b'event: hermes.tool.progress\ndata: {"status":"failed","message":"secret-key"}\n\n', "tool_failed"),
|
|
(200, data({"content": "partial"}), "incomplete_stream"),
|
|
(200, data({"reasoning_content": "secret-key"}) + b"data: [DONE]\n\n", "empty_response"),
|
|
]
|
|
for status, body, code in cases:
|
|
with self.subTest(code=code):
|
|
before = len(self.requests)
|
|
fake = FakeSSE([body])
|
|
stream = self.bridge(fake, status=status).chat(chat_ctx=self.context())
|
|
with self.assertRaises(HermesLLMError) as caught:
|
|
await stream.collect()
|
|
self.assertEqual(caught.exception.code, code)
|
|
self.assertNotIn("secret-key", str(caught.exception))
|
|
self.assertNotIn("sensitive", repr(caught.exception))
|
|
self.assertEqual(len(self.requests), before + 1)
|
|
self.assertEqual(self.progress[-1]["status"], "failed")
|
|
self.assertTrue(fake.closed)
|
|
|
|
async def test_invalid_content_type_and_redirect_not_followed(self):
|
|
for status, expected in [(200, "invalid_content_type"), (302, "http_302")]:
|
|
bridge = self.bridge(FakeSSE([b"html"]), status=status, headers={"content-type": "text/html", "location": "https://attacker.invalid"})
|
|
with self.assertRaises(HermesLLMError) as caught:
|
|
await bridge.chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(caught.exception.code, expected)
|
|
self.assertEqual(len(self.requests), 2)
|
|
|
|
async def test_callbacks_fail_without_replaying_actions(self):
|
|
async def broken(_):
|
|
raise RuntimeError("secret-key")
|
|
fake = FakeSSE([b'event: hermes.tool.progress\ndata: {"status":"working"}\n\n', data({"content": "Done"}), b"data: [DONE]\n\n"])
|
|
bridge = self.bridge(fake)
|
|
bridge._on_progress = bridge._on_wait_notice = broken
|
|
result = await bridge.chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(result.text, "Done")
|
|
self.assertEqual(len(self.requests), 1)
|
|
|
|
async def test_sessions_stable_across_requests_and_caller_owned_client(self):
|
|
bridge = self.bridge(FakeSSE([data({"content": "Done"}), b"data: [DONE]\n\n"]))
|
|
ids = []
|
|
for _ in range(2):
|
|
stream = bridge.chat(chat_ctx=self.context())
|
|
ids.append(stream.request_id)
|
|
await stream.collect()
|
|
self.assertNotEqual(*ids)
|
|
self.assertTrue(all(req.headers["x-hermes-session-id"] == "stable-room" for req in self.requests))
|
|
await bridge.aclose()
|
|
self.assertFalse(bridge._client.is_closed)
|
|
|
|
async def test_origin_base_url_and_eof_finish_frame(self):
|
|
fake = FakeSSE([data({"content": "Done"}), b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}'])
|
|
bridge = self.bridge(fake)
|
|
origin = HermesLLM(base_url="https://trusted.hermes", api_key="secret-key", user_id="alice", room_id="stable-room", on_progress=self.progress_callback, http_client=bridge._client)
|
|
self.assertEqual((await origin.chat(chat_ctx=self.context()).collect()).text, "Done")
|
|
self.assertEqual(str(self.requests[0].url), "https://trusted.hermes/v1/chat/completions")
|
|
|
|
async def test_approval_and_error_finishes_close_immediately(self):
|
|
cases = [
|
|
(b'event: approval.request\ndata: {"secret":"hidden"}\n\n', "approval_unsupported"),
|
|
(b'event: hermes.approval.request\ndata: {}\n\n', "approval_unsupported"),
|
|
(data({}, finish_reason="error"), "upstream_error"),
|
|
(data({}, finish_reason="approval_required"), "approval_unsupported"),
|
|
(b'data: {"hermes":{"error":"secret"},"choices":[]}\n\n', "upstream_error"),
|
|
(b'event: hermes.error\ndata: {"message":"secret"}\n\n', "upstream_error"),
|
|
]
|
|
for body, code in cases:
|
|
with self.subTest(code=code):
|
|
fake = FakeSSE([body], gate=asyncio.Event())
|
|
with self.assertRaises(HermesLLMError) as caught:
|
|
await asyncio.wait_for(self.bridge(fake).chat(chat_ctx=self.context()).collect(), timeout=1)
|
|
self.assertEqual(caught.exception.code, code)
|
|
self.assertTrue(fake.closed)
|
|
self.assertEqual(self.progress[-1]["status"], "failed")
|
|
|
|
async def test_status_event_is_generic_and_never_answer_text(self):
|
|
fake = FakeSSE([b'event: hermes.status\ndata: {"message":"sensitive internals"}\n\n', data({"content": "Done"}), b"data: [DONE]\n\n"])
|
|
result = await self.bridge(fake).chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(result.text, "Done")
|
|
self.assertEqual(self.progress[0]["message"], "Working on your request.")
|
|
self.assertEqual(self.notices, [])
|
|
|
|
async def test_explicit_cancellation_closes_http_without_retry(self):
|
|
fake = FakeSSE([b":keepalive\n\n"], gate=asyncio.Event())
|
|
stream = self.bridge(fake).chat(chat_ctx=self.context())
|
|
await fake.started.wait()
|
|
await stream.aclose()
|
|
self.assertTrue(fake.closed)
|
|
self.assertEqual(len(self.requests), 1)
|
|
self.assertEqual(self.progress[-1]["status"], "failed")
|
|
self.assertEqual(self.progress[-1]["message"], "Request cancelled.")
|
|
self.assertTrue(stream._task.cancelled())
|
|
|
|
async def test_answer_start_hook_runs_once_before_any_content(self):
|
|
calls = []
|
|
async def answer_start():
|
|
calls.append("answer_start")
|
|
fake = FakeSSE([data({"reasoning_content": "ignored"}), data({"content": "First"}), data({"content": " second"}), b"data: [DONE]\n\n"])
|
|
bridge = self.bridge(fake, on_answer_start=answer_start)
|
|
async with bridge.chat(chat_ctx=self.context()) as stream:
|
|
async for chunk in stream:
|
|
self.assertEqual(calls[0], "answer_start")
|
|
calls.append(chunk.delta.content)
|
|
self.assertEqual(calls, ["answer_start", "First", " second"])
|
|
|
|
async def test_answer_start_failure_does_not_retry_or_drop_answer(self):
|
|
async def answer_start():
|
|
raise RuntimeError("secret-key")
|
|
fake = FakeSSE([data({"content": "Done"}), b"data: [DONE]\n\n"])
|
|
result = await self.bridge(fake, on_answer_start=answer_start).chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(result.text, "Done")
|
|
self.assertEqual(len(self.requests), 1)
|
|
|
|
async def test_saved_model_and_thinking_refreshed_per_turn(self):
|
|
selected = {'model': 'openai/gpt-example', 'thinking_level': 'high'}
|
|
async def settings():
|
|
return dict(selected)
|
|
bridge = self.bridge(FakeSSE([data({'content': 'Done'}), b'data: [DONE]\n\n']), get_model_settings=settings)
|
|
for effort in ('high', 'none', 'low'):
|
|
selected['thinking_level'] = effort
|
|
await bridge.chat(chat_ctx=self.context(), extra_kwargs={'provider': 'evil', 'model': 'evil'}).collect()
|
|
body = json.loads(self.requests[-1].content)
|
|
self.assertEqual(body['model'], 'openai/gpt-example')
|
|
self.assertEqual(body['provider'], 'custom:hope-upstream')
|
|
expected = {'enabled': False} if effort == 'none' else {'enabled': True, 'effort': effort}
|
|
self.assertEqual(body['model_options']['reasoning'], expected)
|
|
selected['model'] = 'intel-b70-1/Qwen3.8-27b'
|
|
await bridge.chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(json.loads(self.requests[-1].content)['model'], selected['model'])
|
|
|
|
async def test_settings_failure_does_not_call_hermes_or_leak(self):
|
|
async def broken():
|
|
raise RuntimeError('private-key')
|
|
with self.assertRaises(HermesLLMError) as caught:
|
|
await self.bridge(FakeSSE([]), get_model_settings=broken).chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(caught.exception.code, 'settings_unavailable')
|
|
self.assertNotIn('private-key', str(caught.exception))
|
|
self.assertEqual(self.requests, [])
|
|
|
|
async def test_transport_exception_redacted(self):
|
|
def broken(request):
|
|
self.requests.append(request)
|
|
raise httpx.ConnectError("secret-key sensitive payload", request=request)
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(broken))
|
|
self.clients.append(client)
|
|
bridge = HermesLLM(base_url="https://trusted.hermes", api_key="secret-key", user_id="alice", room_id="room", on_progress=self.progress_callback, http_client=client)
|
|
with self.assertRaises(HermesLLMError) as caught:
|
|
await bridge.chat(chat_ctx=self.context()).collect()
|
|
self.assertEqual(caught.exception.code, "transport_error")
|
|
self.assertTrue(caught.exception.__suppress_context__)
|
|
self.assertEqual(len(self.requests), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|