"""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": "hermes-agent", "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_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()