import asyncio import textwrap import unittest.mock import pytest from livekit.agents import AgentSession, inference from agent import InterviewData, PastExperienceAgent, SelfIntroAgent def _llm() -> inference.LLM: return inference.LLM(model="openai/gpt-4.1-mini") @pytest.mark.asyncio async def test_self_intro_greeting() -> None: """SelfIntroAgent greets the candidate and invites a self-introduction.""" async with ( _llm() as llm, AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session, ): await session.start(SelfIntroAgent()) result = await session.run(user_input="Hello") await ( result.expect.next_event() .is_message(role="assistant") .judge( llm, intent=textwrap.dedent( """\ Greets the candidate warmly and invites them to introduce themselves, share their background, or describe their experience and motivations. """ ), ) ) result.expect.no_more_events() @pytest.mark.asyncio async def test_intro_complete_handoff() -> None: """Highest-priority: after a complete self-introduction, intro_complete fires once and control transfers to PastExperienceAgent.""" async with ( _llm() as llm, AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session, ): await session.start(SelfIntroAgent()) result = await session.run( user_input=( "Hi, I'm Sarah. I've been a software engineer for six years, " "mostly at fintech startups. My most recent role was at a payments company " "where I led backend API development. I'm interested in this position " "because I want to work on larger-scale distributed systems. " "That's my background!" ) ) # intro_complete must be called somewhere in this turn fnc = result.expect.next_event(type="function_call") assert fnc.event().item.name == "intro_complete" # Handoff to PastExperienceAgent must occur (position-independent search) result.expect.contains_agent_handoff(new_agent_type=PastExperienceAgent) @pytest.mark.asyncio async def test_no_premature_handoff() -> None: """A bare greeting must not trigger intro_complete or hand off to PastExperienceAgent.""" async with ( _llm() as llm, AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session, ): await session.start(SelfIntroAgent()) result = await session.run(user_input="Hi") # contains_agent_handoff should raise AssertionError when no handoff is present with pytest.raises(AssertionError): result.expect.contains_agent_handoff(new_agent_type=PastExperienceAgent) @pytest.mark.asyncio async def test_watchdog_cancelled_on_normal_handoff() -> None: """After intro_complete fires normally, on_exit cancels both watchdogs — no second transition replaces the resulting PastExperienceAgent.""" async with ( _llm() as llm, AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session, ): intro_agent = SelfIntroAgent(silence_budget=3.0, stage_budget=30.0) await session.start(intro_agent) result = await session.run( user_input=( "Hi, I'm Sarah. I've been a software engineer for six years, " "mostly at fintech startups. My most recent role was at a payments company " "where I led backend API development. That's my background!" ) ) result.expect.contains_agent_handoff(new_agent_type=PastExperienceAgent) # Give the event loop a tick to process on_exit after the handoff. await asyncio.sleep(0.1) # on_exit must have cancelled and cleared both watchdog tasks. assert intro_agent._silence_task is None, ( "on_exit did not clear _silence_task after intro_complete fired" ) assert intro_agent._stage_task is None, ( "on_exit did not clear _stage_task after intro_complete fired" ) assert isinstance(session.current_agent, PastExperienceAgent) @pytest.mark.asyncio async def test_fallback_handoff() -> None: """Silence fallback: when the silence budget elapses without intro_complete firing, control transfers to PastExperienceAgent. Stage budget is set high so it does not interfere with this silence-path test.""" async with ( _llm() as llm, AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session, ): # In text-test mode user_state is always "listening", so the silence timer # acts as a flat wall-clock countdown. 0.5 s fires quickly; 2 s headroom. await session.start(SelfIntroAgent(silence_budget=0.5, stage_budget=30.0)) await asyncio.sleep(2.0) assert isinstance(session.current_agent, PastExperienceAgent), ( f"Expected PastExperienceAgent after silence budget elapsed; " f"got {type(session.current_agent).__name__}" ) @pytest.mark.asyncio async def test_silence_reset_on_speech() -> None: """Code-path check: the silence timer resets while user_state is 'speaking', so it does not fire during candidate speech. NOTE: this test mocks user_state and does NOT prove real-VAD behavior — the live 25 s continuous-talk run is the acceptance criterion for VAD-driven resets.""" async with ( _llm() as llm, AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session, ): await session.start(SelfIntroAgent(silence_budget=0.5, stage_budget=30.0)) # Simulate candidate speaking for 0.7 s — past the 0.5 s budget. # Without the reset, the silence timer would have fired at 0.5 s. with unittest.mock.patch.object( type(session), "user_state", new_callable=unittest.mock.PropertyMock, return_value="speaking", ): await asyncio.sleep(0.7) # Immediately after "speech" ends: timer has not fired yet (counter restarted). assert isinstance(session.current_agent, SelfIntroAgent), ( "Silence timer fired during speech — reset logic not working" ) # Wait for full silence budget + poll headroom + say() → should fire. await asyncio.sleep(1.5) assert isinstance(session.current_agent, PastExperienceAgent), ( "Silence timer did not fire after full budget of silence post-speech" )