diff --git a/tests/unit/test_conversation_llm.py b/tests/unit/test_conversation_llm.py new file mode 100644 index 0000000..a51be37 --- /dev/null +++ b/tests/unit/test_conversation_llm.py @@ -0,0 +1,173 @@ +"""Offline real-SDK qualification of speech through durable Hope runs.""" +import asyncio +import json +import unittest +import httpx +from livekit.agents import llm +from agent.conversation_llm import ConversationLLM + +class ConversationLLMTests(unittest.IsolatedAsyncioTestCase): + async def test_single_turn_only_and_completed_public_output(self): + requests, progress = [], [] + def handler(request): + requests.append(request) + if request.method == 'POST': + return httpx.Response(202, json={'run': {'id': 'run1'}, 'message': {'id': 'user1'}}) + events = [ + ('message', {'message': {'id':'a','role':'assistant','status':'streaming','content':'draft'}}), + ('progress', {'message':'PRIVATE TOOL DETAILS'}), + ('message', {'message': {'id':'a','role':'assistant','status':'completed','content':'Final answer'}}), + ('done', {'run': {'id':'run1','status':'completed'}}), + ] + body = ''.join('event: '+name+'\ndata: '+json.dumps(data)+'\n\n' for name,data in events) + return httpx.Response(200, text=body, headers={'Content-Type':'text/event-stream'}) + async def feedback(payload): progress.append(payload) + bridge = ConversationLLM(control_url='https://control.test',secret='internal-secret',room_id='room1',on_progress=feedback) + await bridge.client.aclose() + bridge.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + context = llm.ChatContext([ + llm.ChatMessage(role='system',content=['DO NOT FORWARD']), + llm.ChatMessage(role='user',content=['earlier']), + llm.ChatMessage(role='assistant',content=['earlier answer']), + llm.ChatMessage(role='user',content=['latest question']), + ]) + async with bridge.chat(chat_ctx=context) as stream: + chunks = [chunk async for chunk in stream] + self.assertEqual([c.delta.content for c in chunks], ['Final answer']) + payload=json.loads(requests[0].content) + self.assertEqual(payload['content'],'latest question') + self.assertEqual(payload['modality'],'voice') + self.assertNotIn('messages',payload) + self.assertEqual(requests[0].headers['X-Control-Secret'],'internal-secret') + self.assertEqual(progress[-1]['status'],'completed') + self.assertNotIn('PRIVATE',json.dumps(progress)) + await bridge.aclose() + + async def test_rejected_admission_has_no_answer_or_retry(self): + calls=[] + def handler(request): + calls.append(request) + return httpx.Response(403,json={'detail':'private diagnostics'}) + async def progress(_): pass + bridge=ConversationLLM(control_url='https://control.test',secret='s',room_id='r',on_progress=progress) + await bridge.client.aclose() + bridge.client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ctx=llm.ChatContext([llm.ChatMessage(role='user',content=['hello'])]) + with self.assertRaises(Exception): + async with bridge.chat(chat_ctx=ctx) as stream: + async for _ in stream: pass + self.assertEqual(len(calls),1) + await bridge.aclose() + + async def test_repeated_stream_uses_stable_user_turn_id(self): + payloads = [] + def handler(request): + if request.method == 'POST': + payloads.append(json.loads(request.content)) + return httpx.Response(202, json={'run': {'id': 'offline-stable-run'}}) + body = ('event: message\ndata: {"message":{"role":"assistant","status":"completed","content":"Answer"}}\n\n' + 'event: done\ndata: {"run":{"status":"completed"}}\n\n') + return httpx.Response(200, text=body) + async def progress(_): pass + bridge = ConversationLLM(control_url='https://offline.test', secret='offline', room_id='room', on_progress=progress) + await bridge.client.aclose() + bridge.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + context = llm.ChatContext([llm.ChatMessage(role='user', content=['Same physical turn'])]) + try: + for _ in range(2): + async with bridge.chat(chat_ctx=context) as stream: + self.assertEqual([chunk.delta.content async for chunk in stream], ['Answer']) + self.assertEqual(payloads[0]['idempotency_key'], payloads[1]['idempotency_key']) + self.assertEqual(payloads[0]['idempotency_key'], context.items[-1].id) + finally: + await bridge.aclose() + + async def test_cancel_uncertain_admission_retries_identical_key_and_cancels(self): + entered, release = asyncio.Event(), asyncio.Event() + payloads, cancelled_paths = [], [] + async def handler(request): + if request.url.path.endswith('/messages'): + payloads.append(json.loads(request.content)) + if len(payloads) == 1: + entered.set() + await release.wait() + raise httpx.ReadError('Offline lost admission receipt', request=request) + return httpx.Response(202, json={'run': {'id': 'offline-recovered-id'}}) + if request.url.path.endswith('/cancel'): + cancelled_paths.append(request.url.path) + return httpx.Response(200, json={'run': {'status': 'cancelled'}}) + self.fail('Cancelled admission must not attach an answer stream') + async def progress(_): pass + bridge = ConversationLLM(control_url='https://offline.test', secret='offline', room_id='bound-room', on_progress=progress) + await bridge.client.aclose() + bridge.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + context = llm.ChatContext([llm.ChatMessage(role='user', content=['Uncertain admission'])]) + stream = bridge.chat(chat_ctx=context) + async def consume(): + async with stream: + return [chunk async for chunk in stream] + consumer = asyncio.create_task(consume()) + try: + await asyncio.wait_for(entered.wait(), 2) + closer = asyncio.create_task(stream.aclose()) + await asyncio.sleep(.02) + release.set() + await asyncio.wait_for(closer, 2) + await asyncio.gather(consumer, return_exceptions=True) + self.assertEqual(len(payloads), 2) + self.assertEqual(payloads[0], payloads[1]) + self.assertEqual(cancelled_paths, ['/internal/chat/rooms/bound-room/runs/offline-recovered-id/cancel']) + finally: + release.set() + if not consumer.done(): consumer.cancel() + await asyncio.gather(consumer, return_exceptions=True) + await bridge.aclose() + + async def test_cancel_blocked_admission_learns_and_cancels_remote_id(self): + admission_entered, release_admission, cancelled = asyncio.Event(), asyncio.Event(), asyncio.Event() + cancellations = [] + async def handler(request): + if request.url.path.endswith('/messages'): + admission_entered.set() + await release_admission.wait() + return httpx.Response(202, json={'run': {'id': 'offline-admitted-run'}}) + if request.url.path.endswith('/cancel'): + cancellations.append(request.url.path) + cancelled.set() + return httpx.Response(200, json={'run': {'status': 'cancelled'}}) + return httpx.Response(200, text='event: done\ndata: {"run":{"status":"cancelled"}}\n\n') + async def progress(_): pass + bridge = ConversationLLM(control_url='https://offline.test', secret='offline', room_id='room', on_progress=progress) + await bridge.client.aclose() + bridge.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + context = llm.ChatContext([llm.ChatMessage(role='user', content=['Cancel during admission'])]) + stream = bridge.chat(chat_ctx=context) + async def consume(): + async with stream: + return [chunk async for chunk in stream] + consumer = asyncio.create_task(consume()) + closer = second_closer = None + try: + await asyncio.wait_for(admission_entered.wait(), 2) + # Public SDK cancellation must not orphan an admission already accepted + # remotely but whose response has not reached the agent. + closer = asyncio.create_task(stream.aclose()) + await asyncio.sleep(.02) + second_closer = asyncio.create_task(stream.aclose()) + await asyncio.sleep(.02) + release_admission.set() + await asyncio.wait_for(asyncio.gather(closer, second_closer), 2) + await asyncio.gather(consumer, return_exceptions=True) + await asyncio.wait_for(cancelled.wait(), .3) + self.assertEqual(cancellations, ['/internal/chat/rooms/room/runs/offline-admitted-run/cancel']) + finally: + release_admission.set() + if closer and not closer.done(): + closer.cancel() + await asyncio.gather(closer, return_exceptions=True) + if not consumer.done(): + consumer.cancel() + await asyncio.gather(consumer, return_exceptions=True) + await bridge.aclose() + +if __name__=='__main__': unittest.main()