onIntroductionComplete

we're returning

QuestionAgent

as next agent, but QuestionAgent

on_enter

method is not getting called automatically... i've to re-trigger the agent by speaking something in order to trigger

on_enter

, why is that? PS: i've removed some codes, in order to send it on slack, since too much line of messages are not allowed on slack, so do let me know if you need to see any particular part of code before commenting anything

class IntroductionAgent(Agent):
   
    async def on_exit(self) -> None:
        exit_phrase = "Let's start with the interview."
        if not self._is_realtime_sts:
            self.session.say(exit_phrase)
        else:
            await safe_generate_reply(
                self.session, instructions=f"Say exactly: '{exit_phrase}'"
            )
        return await super().on_exit()

    @function_tool
    async def onIntroductionComplete(self, context: RunContext[InterviewSession]) -> Optional[Agent]:
        """Called when the introduction is complete and candidate is ready to move to questions."""
        session = context.userdata
        session.stage = InterviewStage.QUESTIONS
        job_ctx = get_job_context()
        room_name = job_ctx.room.name

        <http://logger.info|logger.info>("Starting recording after introduction stage...")
        result: Optional[RecordingResult] = await self._recording_service.start_recording(room_name)
        if result:
            <http://logger.info|logger.info>("Recording started successfully, egress_id=%s", result.egress_id)
            await save_recording_info_to_db(room_name, _recording_result_to_dict(result))
        else:
            logger.warning("Could not start recording - continuing without recording")

        await notify_client(
            room_name=room_name,
            event_type="stage_change",
            data={
                "previous_stage": InterviewStage.INTRODUCTION.value,
                "new_stage": InterviewStage.QUESTIONS.value,
                "session_state": get_session_state(session),
            },
        )

        current_question = session.get_current_question()
        if current_question:
            print(f"current_question: {current_question}")
            is_last = session.current_question_index == len(session.questions) - 1
            print(f"is_last: {is_last}")
            next_agent = QuestionAgent(
                question=current_question,
                pipeline_kwargs=self._pipeline_kwargs,
                is_realtime_sts=self._is_realtime_sts,
                is_last_question=is_last,
            )
            <http://logger.info|logger.info>("Moving to questions phase with first question: %s", current_question.question)
            return next_agent

        return AfterInterviewAgent(pipeline_kwargs=self._pipeline_kwargs)


class QuestionAgent(Agent):
    def __init__(
        self,
        question: Question,
        pipeline_kwargs: Dict[str, Any],
        is_realtime_sts: bool,
        is_last_question: bool = False,
    ) -> None:
        self.question = question
        self._pipeline_kwargs = pipeline_kwargs
        self._is_realtime_sts = is_realtime_sts
        self._is_last_question = is_last_question

        prefix = pipeline_kwargs.get("instruction_prefix", "")

        if is_last_question:
            wrap_instruction = (
                "When you're satisfied with the candidate's answers to this question and its follow-ups, "
                'say: "Do you want to add anything before we wrap up the interview?" '
                'Call "discussionEnded" when the candidate is ready to wrap up.'
            )
        else:
            wrap_instruction = (
                'When you\'re satisfied with the candidate\'s answers to a particular question and its follow-ups, '
                'say: "Do you want to add anything or should we move to another question?" '
                'Call "discussionEnded" when the candidate agreed to move to another question.'
            )

        super().__init__(
            instructions=(
                prefix
                + f"""
                You are an interviewer conducting interview with a candidate. Your responses will be converted to speech through text-to-speech technology, so keep your responses conversational and natural as if you're speaking directly to the candidate.

You have a predefined question and follow-up questions. Ask the main question first, then ask follow-up questions if the candidate hasn't already addressed those points in their response.

{wrap_instruction}

Only speak in english, if user speaks or asking to communicate in other languages, ask him to talk in english only


<question>
Question: {self.question.question}
Follow-ups:
{chr(10).join(self.question.follow_up_questions)}
</question>
"""
            ),
            **{k: v for k, v in pipeline_kwargs.items() if k != "instruction_prefix"},
        )

    async def on_enter(self) -> None:
        print(f"********************************on_enter: {self.question.asked}")
        if not self.question.asked:
            print(f"on_enter: {self.question.asked}")
            <http://logger.info|logger.info>("Entering question: %s", self.question.question)
            if not self._is_realtime_sts:
                await self.session.say(self.question.question)
            else:
                await safe_generate_reply(
                    self.session,
                    instructions=f"Ask the candidate this question exactly: '{self.question.question}'",
                )
            self.question.asked = True
        else:
            <http://logger.info|logger.info>("Continuing question: %s", self.question.question)
            continue_phrase = f"Let's continue with the interview. '{self.question.question}'"
            if not self._is_realtime_sts:
                await self.session.say(continue_phrase)
            else:
                await safe_generate_reply(
                    self.session,
                    instructions=f"Say exactly: '{continue_phrase}'",
                )

    @function_tool
    async def discussionEnded(self, context: RunContext[InterviewSession]) -> Any:
        """Called when the LLM is satisfied with user response."""
        session = context.userdata
        current_question_index = session.current_question_index
        session.move_to_next_question()
        job_ctx = get_job_context()
        room_name = job_ctx.room.name
        <http://logger.info|logger.info>("Moving to next question")

        if session.has_more_questions():
            next_question = session.get_current_question()
            if next_question:
                await notify_client(
                    room_name=room_name,
                    event_type="question_movement",
                    data={
                        "previous_question_index": current_question_index,
                        "new_question_index": session.current_question_index,
                        "session_state": get_session_state(session),
                    },
                )
                is_last = session.current_question_index == len(session.questions) - 1
                next_agent = QuestionAgent(
                    question=next_question,
                    pipeline_kwargs=self._pipeline_kwargs,
                    is_realtime_sts=self._is_realtime_sts,
                    is_last_question=is_last,
                )
                <http://logger.info|logger.info>("Moving to next question: %s", next_question.question)
                next_phrase = "Let's move on to the next question."
                if not self._is_realtime_sts:
                    await self.session.say(next_phrase)
                else:
                    await safe_generate_reply(
                        self.session,
                        instructions=f"Say exactly: '{next_phrase}'",
                    )
                return next_agent

        previous_stage = session.stage
        session.stage = InterviewStage.COMPLETED
        await notify_client(
            room_name=room_name,
            event_type="stage_change",
            data={
                "previous_stage": previous_stage.value,
                "new_stage": InterviewStage.COMPLETED.value,
                "session_state": get_session_state(session),
            },
        )
        after_agent = AfterInterviewAgent(
            chat_ctx=context.session._chat_ctx,
            pipeline_kwargs=self._pipeline_kwargs,
        )
        <http://logger.info|logger.info>("All questions completed, moving to wrap-up")
        return after_agent, "Excellent! We've covered all the technical questions. Let me wrap up the interview."


class AfterInterviewAgent(Agent):
    def __init__(
        self,
        pipeline_kwargs: Dict[str, Any],
        chat_ctx: Optional[ChatContext] = None,
    ) -> None:
        prefix = pipeline_kwargs.get("instruction_prefix", "")
        kwargs = {k: v for k, v in pipeline_kwargs.items() if k != "instruction_prefix"}
        super().__init__(
            instructions=(
                prefix
                + "You are wrapping up a technical interview. The candidate has completed all the questions.\n\n"
                "Your role is to:\n"
                "1. Thank the candidate for their time and participation\n"
                "2. Provide some positive feedback about their performance\n"
                "3. Explain the next steps (if any) - typically that the team will review and follow up\n"
                "4. Ask if they have any questions about the role or company\n"
                "5. Once the conversation feels complete, politely ask them to end the call by pressing the end button\n\n"
                "Be warm, professional, and leave them with a positive impression. "
                "Keep this phase relatively brief - 1-2 minutes maximum. "
                "Make sure they know this is the end of the interview."
            ),
            chat_ctx=chat_ctx,
            **kwargs,
        )

    async def on_enter(self) -> None:
        await safe_generate_reply(self.session, fallback_phrase="Let me wrap up.")

    @function_tool
    async def end_interview(self, context: RunContext[InterviewSession]) -> None:
        """Called to definitively end the interview session."""
        self.session.interrupt()
        await safe_generate_reply(
            self.session,
            instructions="Give a brief, warm goodbye and remind them to press the end call button",
        )
        job_ctx = get_job_context()
        await job_ctx.api.room.delete_room(api.DeleteRoomRequest(room=job_ctx.room.name))


def _recording_result_to_dict(result: RecordingResult) -> Dict[str, Any]:
    return {
        "egress_id": result.egress_id,
        "s3_url": result.s3_url,
        "s3_bucket": result.s3_bucket,
        "s3_key": result.s3_key,
        "filename": result.filename,
        "s3_region": result.s3_region,
    }

Thanks for reaching out, Rahul. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I've checked:- 1. on_enter not even getting called, when we switch agent from introduction to questio 2. we're using nova 2 sonic STS model

Thanks — this is helpful. You’ve confirmed two key facts I needed: (1)

on_enter

is not being called at all when you switch agents, and (2) you’re using the Nova 2 Sonic STS realtime model. That narrows the likely causes. What I’ll help you do next • Give the most likely reasons

on_enter

isn't being invoked for a handoff to a new agent in a realtime session. • Give short, concrete checks and a minimal set of instrumentation you can add to confirm where the transition is breaking. • Ask one targeted follow-up to pin down the remaining unknown that we need to debug. Why this is happening (most likely) • Realtime session lifecycle: realtime STS models can rely on an active realtime loop and explicit turn handling. If the agent activation code path that triggers lifecycle hooks (like

on_enter

) isn’t run on the realtime engine thread, the agent instance may be returned but never made "active" in the runtime loop so its lifecycle hooks won’t execute. • Handoff path being bypassed: if the handoff return value from

onIntroductionComplete

is not being processed by the runtime's agent dispatcher (for example because an exception was raised, the response was not returned to the dispatcher, or the dispatcher is still processing the previous agent's turn),

on_enter

will not be invoked. • Concurrency/await issue: if the runtime is awaiting something else (long-running task, blocking call, or canceled coroutine) between you returning the new agent and the dispatcher loop, the activation may be skipped. • Runtime-specific behavior with Nova 2 Sonic STS: realtime engines sometimes require an explicit "resume/activate" call for the new agent, or they require the agent to request generation (e.g., via

generate_reply

) to kick off its first action. If the runtime expects an explicit event to schedule agent work,

on_enter

might not be automatically scheduled. Quick checks and minimal instrumentation (do these first) 1. Confirm the runtime received the returned agent: ◦ Right before

return next_agent

in

onIntroductionComplete

, add: ▪︎

<http://logger.info|logger.info>("Returning next_agent: %s", type(next_agent).__name__)

◦ Immediately after that, in your global runtime/dispatcher code (the place that receives function tool returns), add a logger that logs when it receives a new agent assignment. If you don’t have access to that dispatcher, add instrumentation right after your call site where

onIntroductionComplete

is invoked. 1. Verify

on_enter

truly never runs: ◦ Replace your

print

in

on_enter

with

<http://logger.info|logger.info>("QuestionAgent.on_enter called for question: %s", self.question.question)

at the top.

print

can be swallowed; logger is more reliable in async runtimes. 1. Ensure there are no swallowed exceptions: ◦ Wrap the body of

onIntroductionComplete

(and the call sites that invoke function tools) with try/except and log any exception with a stack trace so you can see if something threw during handoff processing. 1. Confirm agent dispatcher scheduling: ◦ If you can, log inside the runtime's agent dispatcher when it calls lifecycle hooks. If the dispatcher never calls

on_enter

, we know the returned agent isn't being scheduled. 1. Short experiment to force execution: ◦ Temporarily, after

return next_agent

, also schedule a background task to call the agent’s

on_enter

manually (only for debugging). Example pattern: ▪︎ After returning, create an async task:

asyncio.create_task(manual_activate(next_agent, session))

where

manual_activate

awaits

next_agent.on_enter()

and logs results. If that triggers the expected behavior, the issue is the dispatcher not scheduling lifecycle hooks. Concrete things to try now • Add an unconditional

await safe_generate_reply(self.session, instructions="Ask the candidate exactly: '...'")

at the top of

QuestionAgent.on_enter

and ensure you see the logger entry for

on_enter

. If you see the logger but no speech, the generation is blocked by realtime turn logic; if you don't see the logger,

on_enter

is never invoked. • Confirm the runtime is actually consuming the return from the function tool. Log the function tool return value at the boundary where the return is received. Targeted follow-up I need from you • Please paste the AgentSession creation / session startup snippet (the code that creates the session, starts the agent runtime, and registers function tools or handlers). That’s the place where a realtime dispatcher is responsible for scheduling

on_enter

. Seeing that will let me point to the exact line where the returned agent should be scheduled and which call might be missing/blocked. If you prefer, paste only the bits that: • create/start the session, • register the agent function tool dispatcher, and • run the agent loop. Once I have that, I can tell you exactly where to add logging or a one-line change to ensure the dispatcher schedules the returned agent’s lifecycle hooks immediately.