"""Agent loop machinery with a scripted LLM (no API key needed): prompt composition, tool dispatch, transcripts, outcome classification.""" import json from datetime import datetime, timedelta from zoneinfo import ZoneInfo from sqlalchemy import select from gogo.agent.llm import LLMResponse, ScriptedLLM, ToolUse from gogo.agent.loop import AgentConversation from gogo.agent.prompt import compose_greeting, compose_system_prompt from gogo.models import BookingRequest, MessageForOwner, Service TZ = ZoneInfo("Europe/Sarajevo") # -- prompt composition ------------------------------------------------------- async def test_system_prompt_contains_tenant_data(session, tenant): prompt = await compose_system_prompt(session, tenant, "voice") assert "Salon Merima" in prompt assert "Šišanje i feniranje" in prompt assert "45 min" in prompt assert "25–35 KM" in prompt assert "ponedjeljak: 09:00–18:00" in prompt assert "razgovor se snima" in prompt # recording disclosure in greeting (§5.4) assert "phone call" in prompt # voice channel suffix # service ids are exposed so the model can call tools with them svc = ( await session.execute(select(Service).where(Service.tenant_id == tenant.id)) ).scalars().first() assert str(svc.id) in prompt async def test_chat_prompt_requires_phone_collection(session, tenant): prompt = await compose_system_prompt(session, tenant, "chat") assert "web chat" in prompt assert "phone number" in prompt def test_greeting_variants(tenant): open_at = datetime(2026, 7, 15, 10, 0, tzinfo=TZ) # Wed 10:00 closed_at = datetime(2026, 7, 15, 21, 0, tzinfo=TZ) # Wed 21:00 g_open = compose_greeting(tenant, "voice", open_at) g_closed = compose_greeting(tenant, "voice", closed_at) g_chat = compose_greeting(tenant, "chat") assert "razgovor se snima" in g_open assert "zatvoren" not in g_open assert "trenutno zatvoren" in g_closed assert "radno vrijeme" in g_closed assert "razgovor se snima" not in g_chat # no recording disclosure in chat async def test_prompt_template_override(session, tenant): from gogo.models import TenantPromptOverride session.add( TenantPromptOverride(tenant_id=tenant.id, body="CUSTOM {salon_profile} END") ) await session.flush() prompt = await compose_system_prompt(session, tenant, "voice") assert prompt.startswith("CUSTOM") assert "Salon Merima" in prompt # -- conversation loop -------------------------------------------------------- async def booking_conversation(session, tenant, emails): """Scripted A.1-style flow: availability check → booking request.""" svc = ( await session.execute( select(Service).where(Service.tenant_id == tenant.id, Service.name.like("Šišanje%")) ) ).scalar_one() wed = datetime(2026, 7, 15, tzinfo=TZ) slot_start = wed.replace(hour=17) llm = ScriptedLLM( [ # turn 1: caller asks for a booking → model checks availability LLMResponse( text="", tool_uses=[ ToolUse( id="tu1", name="check_availability", input={ "service_id": str(svc.id), "date_from": "2026-07-15", "date_to": "2026-07-15", }, ) ], stop_reason="tool_use", ), LLMResponse(text="U srijedu poslijepodne slobodno je u pet. Odgovara?"), # turn 2: caller accepts → model submits the request and closes LLMResponse( text="", tool_uses=[ ToolUse( id="tu2", name="submit_booking_request", input={ "service_id": str(svc.id), "service_name_raw": "šišanje i feniranje", "client_name": "Amra Hodžić", "client_phone": "+38765123456", "slots": [ { "start": slot_start.isoformat(), "end": (slot_start + timedelta(minutes=45)).isoformat(), } ], "summary": "Šišanje i feniranje, srijeda u 17h.", }, ) ], stop_reason="tool_use", ), LLMResponse( text=( "Prosljeđujem salonu zahtjev: šišanje i feniranje, srijeda u pet. " "Kontaktiraće vas u najkraćem roku radi potvrde. Hvala i prijatno!" ) ), ] ) convo = AgentConversation( session, tenant, channel="voice", caller_phone="+38765123456", llm=llm ) await convo.greeting() r1 = await convo.user_turn("Htjela bih zakazati šišanje i feniranje u srijedu.") r2 = await convo.user_turn("U pet, može. Amra Hodžić, broj je ovaj s kojeg zovem.") return convo, llm, r1, r2 async def test_booking_flow_creates_request(session, tenant, emails, sms, clean_mock_provider): convo, llm, r1, r2 = await booking_conversation(session, tenant, emails) await session.commit() assert "slobodno je u pet" in r1 assert "prijatno" in r2.lower() # tool_result was fed back to the model tool_result_msg = llm.calls[1]["messages"][-2] # assistant tool_use, then user tool_result assert tool_result_msg["role"] == "assistant" results = llm.calls[1]["messages"][-1] assert results["role"] == "user" assert results["content"][0]["type"] == "tool_result" payload = json.loads(results["content"][0]["content"]) assert "slots" in payload # a real BookingRequest exists and the proposal email went out req = (await session.execute(select(BookingRequest))).scalar_one() assert req.client_name == "Amra Hodžić" assert req.source == "voice" assert len(emails) == 1 assert convo.outcome == "request_created" # transcript captures roles, text and tool calls roles = [t["role"] for t in convo.transcript] assert roles == ["assistant", "user", "assistant", "user", "assistant"] assert convo.transcript[-1]["tool_calls"][0]["name"] == "submit_booking_request" async def test_take_message_outcome(session, tenant, emails, sms): llm = ScriptedLLM( [ LLMResponse( text="", tool_uses=[ ToolUse( id="tu1", name="take_message", input={ "text": "Selma Kovač otkazuje sutrašnji termin u deset.", "client_name": "Selma Kovač", }, ) ], stop_reason="tool_use", ), LLMResponse(text="Prosljeđujem salonu poruku. Prijatno!"), ] ) convo = AgentConversation( session, tenant, channel="voice", caller_phone="+38765111222", llm=llm ) await convo.user_turn("Trebala bih otkazati termin za sutra u deset, Selma Kovač.") await session.commit() msg = (await session.execute(select(MessageForOwner))).scalar_one() assert "otkazuje" in msg.text assert msg.client_phone == "+38765111222" # caller-ID fallback assert convo.outcome == "message_taken" async def test_info_only_outcome(session, tenant): llm = ScriptedLLM([LLMResponse(text="Farbanje je od šezdeset do devedeset maraka.")]) convo = AgentConversation(session, tenant, channel="voice", llm=llm) await convo.user_turn("Koliko košta farbanje?") assert convo.outcome == "info_only" async def test_submit_without_phone_fails_in_chat(session, tenant, emails): """Chat has no caller-ID: submitting without a phone returns an error to the model.""" llm = ScriptedLLM( [ LLMResponse( text="", tool_uses=[ ToolUse( id="tu1", name="submit_booking_request", input={"client_name": "Ana", "summary": "manikir"}, ) ], stop_reason="tool_use", ), LLMResponse(text="Koji je vaš broj telefona za kontakt?"), ] ) convo = AgentConversation(session, tenant, channel="chat", llm=llm) await convo.user_turn("Može manikir sutra?") results = llm.calls[1]["messages"][-1] payload = json.loads(results["content"][0]["content"]) assert payload["error"] == "missing_phone" assert (await session.execute(select(BookingRequest))).scalar_one_or_none() is None assert convo.outcome == "info_only" async def test_hallucinated_service_id_is_dropped(session, tenant, emails): llm = ScriptedLLM( [ LLMResponse( text="", tool_uses=[ ToolUse( id="tu1", name="submit_booking_request", input={ "service_id": "not-a-real-uuid", "service_name_raw": "nešto čudno", "client_name": "Ana", "client_phone": "+38761000111", "summary": "test", }, ) ], stop_reason="tool_use", ), LLMResponse(text="Zahtjev proslijeđen."), ] ) convo = AgentConversation(session, tenant, channel="chat", llm=llm) await convo.user_turn("Zakazi mi nešto čudno") await session.commit() req = (await session.execute(select(BookingRequest))).scalar_one() assert req.service_id is None assert req.service_name_raw == "nešto čudno" async def test_tool_round_limit_guard(session, tenant): """A model stuck in a tool loop cannot spin forever.""" endless = LLMResponse( text="", tool_uses=[ToolUse(id="x", name="get_salon_info", input={})], stop_reason="tool_use", ) llm = ScriptedLLM([endless] * 20) convo = AgentConversation(session, tenant, channel="chat", llm=llm) reply = await convo.user_turn("zdravo") assert len(llm.calls) == 6 # MAX_TOOL_ROUNDS_PER_TURN assert reply # graceful fallback text, not an exception async def test_unknown_tool_returns_error(session, tenant): llm = ScriptedLLM( [ LLMResponse( text="", tool_uses=[ToolUse(id="x", name="book_now", input={})], stop_reason="tool_use", ), LLMResponse(text="Izvinite."), ] ) convo = AgentConversation(session, tenant, channel="chat", llm=llm) await convo.user_turn("test") payload = json.loads(llm.calls[1]["messages"][-1]["content"][0]["content"]) assert "unknown tool" in payload["error"]