The @ctx.room.on("participant_connected") is not getting fired on livekit cloud. But when i run the agent in the dev mode it gets fired just perfectly

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

await ctx.connect() This is in the first line of my entrypoint and the handler is registered afterwards

Thanks — that explains the difference. Register the

@ctx.room.on("participant_connected")

handler before calling

await ctx.connect()

so the room listener is attached before any participants can join. If you connect first, the agent can miss events that occur immediately on connect in Cloud. Also try these quick checks: • Add logs right after registering the handler and immediately after

await ctx.connect()

to confirm ordering. • If participants might already be present when the agent starts, call

await ctx.wait_for_participant()

after connect to detect existing participants. • If

participant_connected

still doesn’t fire, register

@ctx.room.on("participant_joined")

as well. If you want, paste your entrypoint code (with secrets redacted) and I’ll point out the exact place to move the handler.

THen how is this working on dev?

Good question — that difference comes down to timing and how the environments deliver events. In dev mode the agent, room, and participants often run in the same process or with minimal network latency, so the participant connection event can arrive slightly later and your handler (registered after

await ctx.connect()

) still catches it. Cloud runs with more distributed timing: the participant may join immediately when the room becomes active, and events can be emitted before your handler is attached — so the handler misses them. What to do: • Register

@ctx.room.on("participant_connected")

before

await ctx.connect()

so you never miss events. • Keep the

await ctx.wait_for_participant()

fallback after connect to handle participants that were already present. • Add short startup logs around handler registration and

ctx.connect()

to confirm ordering in Cloud. If you want, paste your entrypoint and I’ll show exactly where to move the handler.

async def entrypoint(ctx: JobContext) -> None:
    # Setup and metadata parsing
    assistant = Assistant()
    metadata = Metadata.model_validate(...)
    
    # Send room_started event
    room_started_event = api.WebhookEvent(...)
    utils.send_webhook_event_background(...)
    
    # Send participant_connected events for existing participants
    for _participant_sid, participant in ctx.room.remote_participants.items():
        participant_connected_event = api.WebhookEvent(...)
        utils.send_webhook_event_background(...)
    
    # Start egress recording
    req = api.RoomCompositeEgressRequest(...)
    lkapi = api.LiveKitAPI()
    res = await lkapi.egress.start_room_composite_egress(req)
    webhook_event = api.WebhookEvent(...)
    utils.send_webhook_event_background(...)
    await lkapi.aclose()
    
    # Generate auth token and create session
    auth_token = generate_livekit_token(ttl_seconds=7200)
    session = AgentSession(...)
    
    # Event handlers registered
    @session.on("agent_false_interruption")
    def _on_agent_false_interruption(ev: AgentFalseInterruptionEvent) -> None:
        ...
    
    @session.on("user_input_transcribed")
    def _on_user_input_transcribed(event: UserInputTranscribedEvent) -> None:
        ...
    
    @session.on("conversation_item_added")
    def _on_conversation_item_added(event: SDKConversationItemAddedEvent) -> None:
        ...
    
    @session.on("user_state_changed")
    def _on_user_state_changed(event: SDKUserStateChangedEvent) -> None:
        ...
    
    @session.on("agent_state_changed")
    def _on_agent_state_changed(event: SDKAgentStateChangedEvent) -> None:
        ...
    
    # Metrics collector setup
    metrics_collector = MetricsCollector()
    metrics_collector.register_shutdown_callback(ctx)
    
    @session.on("metrics_collected")
    def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
        ...
    
    # Silence monitoring
    async def monitor_interaction() -> None:
        ...
    
    @session.on("close")
    def on_close(ev: Any) -> None:
        ...
    
    # Room event handlers
    @ctx.room.on("connected")
    def on_room_connected() -> None:
        ...
    
    @ctx.room.on("disconnected")
    def on_room_disconnected() -> None:
        ...
    
    async def async_on_participant_connected(participant: Any) -> None:
        ...
    
    @ctx.room.on("participant_connected")
    def on_participant_connected(participant: Any) -> None:
        ...
    
    @ctx.room.on("participant_disconnected")
    def on_participant_disconnected(participant: rtc.Participant) -> None:
        ...
    
    # Start session and initial greeting
    await session.start(...)
    await ctx.connect()
    await session.say(initial_call_greeting, allow_interruptions=False)
    
    # Start silence monitoring
    monitor_task = asyncio.create_task(monitor_interaction())
    
    # Transcript writing and cleanup
    async def write_transcript() -> None:
        ...
    
    ctx.add_shutdown_callback(write_transcript)

This is my entrypoint? What to correct here?

Thanks — this is very helpful. The main issue is that you call

await ctx.connect()

before the room event handlers are registered, so the agent can miss participant events in Cloud. Make these changes: 1. Register the room handlers before calling

await ctx.connect()

. 2. Make the participant handler async (you already have `async_on_participant_connected`; use that) so you can await inside it if needed. 3. After connect, handle existing participants with

await ctx.wait_for_participant()

or iterate

ctx.room.remote_participants

as a fallback. Suggested minimal change (move handlers above connect and use async handler):

# Room event handlers — place these BEFORE await ctx.connect()
@ctx.room.on("connected")
def on_room_connected() -> None:
    ...

@ctx.room.on("disconnected")
def on_room_disconnected() -> None:
    ...

@ctx.room.on("participant_connected")
async def on_participant_connected(participant: Any) -> None:
    # async logic or delegate to async_on_participant_connected
    await async_on_participant_connected(participant)

@ctx.room.on("participant_disconnected")
def on_participant_disconnected(participant: rtc.Participant) -> None:
    ...

Then later:

# Start session and connect after handlers are set
await session.start(...)
await ctx.connect()

# Optional: handle participants already present
await ctx.wait_for_participant(timeout=0)  # returns immediately if none
# or iterate ctx.room.remote_participants here

If you paste the body of

async_on_participant_connected

or your participant-handling logic, I’ll show the exact spot to move it and a final snippet with secrets redacted.

async def async_on_participant_connected( participant: Any, ) -> None: if session.agent_state == "listening": await session.say(metadata.conversation_data.continuation_call_greeting)

You're close — move registration of the room handlers (including the async participant handler) so they run before

await ctx.connect()

. Use the async handler directly and keep the existing check/`session.say`. Minimal corrected layout:

async def entrypoint(ctx: JobContext) -> None:
    # Setup and metadata parsing...
    assistant = Assistant()
    metadata = Metadata.model_validate(...)

    # Register room event handlers BEFORE connecting
    @ctx.room.on("connected")
    def on_room_connected() -> None:
        ...

    @ctx.room.on("disconnected")
    def on_room_disconnected() -> None:
        ...

    @ctx.room.on("participant_connected")
    async def on_participant_connected(participant: Any) -> None:
        if session.agent_state == "listening":
            await session.say(metadata.conversation_data.continuation_call_greeting)

    @ctx.room.on("participant_disconnected")
    def on_participant_disconnected(participant: rtc.Participant) -> None:
        ...

    # Start session and connect AFTER handlers are set
    await session.start(...)
    await ctx.connect()

    # Handle participants already present (optional)
    await ctx.wait_for_participant(timeout=0)
    for _sid, participant in ctx.room.remote_participants.items():
        # If you want to invoke the same logic for already-present participants:
        await on_participant_connected(participant)

    # Continue with greeting, monitoring, egress, shutdown, etc.

Notes: • Registering handlers before

ctx.connect()

prevents missing events in Cloud. • Use the async handler so

await session.say(...)

works. •

await ctx.wait_for_participant(timeout=0)

lets you handle participants who were already in the room when you connected. If you want, paste the surrounding code (with secrets removed) and I’ll return a ready-to-drop snippet.

Even now @ctx.room.on("participant_connected") is not getting fired

Thanks — you’ve done the right restructuring, but since it still isn't firing we need to narrow the cause. A few targeted checks and one quick change will tell us what's happening. Please run these checks and paste the outputs / logs you see: 1. Confirm handler registration order • Add these logs immediately before registering handlers, immediately after registering the participant handler, and immediately before/after `await ctx.connect()`: ◦ "registering room handlers" ◦ "participant handler registered" ◦ "calling ctx.connect()" ◦ "after ctx.connect()" • Share those log lines. 1. Log current participants right after connect • After

await ctx.connect()

, add: ◦ Print

list(ctx.room.remote_participants.keys())

and each participant.identity. • Share that output. 1. Verify the async handler is the one registered • Show the exact decorated handler in your deployed code (the function definition with

@ctx.room.on("participant_connected")

) — confirm it's

async def

and matches the function you tested locally. 1. Confirm you are using

ctx.wait_for_participant()

as a fallback • After connect run: ◦

p = await ctx.wait_for_participant(timeout=1)

and log whether

p

is None or a participant identity. • Share result. 1. SDK / deployment differences • Tell me the LiveKit agent SDK version in dev vs Cloud, and whether you deploy the exact same file (no conditional code paths). If you prefer a one-line quick test to validate event delivery immediately: • Move the simplest async handler above connect:

@ctx.room.on("participant_connected")
  async def _p(p):
      print("EVENT participant_connected", p.identity)

Deploy this and paste the Cloud startup logs. Send the logs/answers and I’ll point to the exact code line to change or the remaining environmental issue. Sources: Job lifecycle | LiveKit docs

await ctx.connect() await ctx.wait_for_participant(timeout=0, kind=rtc.ParticipantKind.PARTICIPANT_KIND_AGENT) is this correct?

Yes — that usage is correct, with one important note about ordering. What you wrote is fine:

await ctx.connect()
await ctx.wait_for_participant(timeout=0, kind=rtc.ParticipantKind.PARTICIPANT_KIND_AGENT)

Notes to ensure it behaves reliably on Cloud: • Register your

@ctx.room.on("participant_connected")

handler before calling

await ctx.connect()

so you don't miss any events that occur during connection. • Use an

async def

handler if it does

await

(e.g.,

await session.say(...)

). •

await ctx.wait_for_participant(timeout=0, kind=...)

is a correct fallback to block until a matching participant is present (timeout=0 waits indefinitely). • After connect, also log

list(ctx.room.remote_participants.keys())

to verify participants already present. If you want, paste the exact deployed handler and the connect/wait lines and I’ll confirm placement and provide the minimal change. Sources: Job lifecycle | LiveKit docs | Rooms, participants, and tracks | LiveKit docs | Agents Framework | livekit/livekit_composite | DeepWiki | How to Enable Agent-to-Agent Speech Detection in LiveKit

How do i know if a participant dropped and rejoin in a room?

You can detect a drop-and-rejoin by handling the participant lifecycle events and tracking participant identity vs participant SID. Use these patterns: 1) Register handlers (before connect) •

@ctx.room.on("participant_disconnected")

— fired when a participant leaves or disconnects. •

@ctx.room.on("participant_connected")

— fired when a participant (re)joins. 2) Track identity vs SID • Save participant.identity (stable for a user) and participant.sid (unique per connection). • On

participant_disconnected

, record the SID and timestamp; on

participant_connected

, compare the incoming participant.identity to any previously disconnected SIDs: ◦ If identity matches a stored identity but SID is different => user rejoined (new connection). ◦ If identity matches and SID is the same => rare (reconnect with same SID). ◦ If identity new => new participant. 3) Optional: short-term state store • Keep a small in-memory map: identity -> last_sid, last_seen_ts, disconnected_flag. Update on connect/disconnect to detect rejoin within your desired window. 4) Use

participant_state

or presence checks after connect • After

await ctx.connect()

, check

ctx.room.remote_participants

to reconcile current state with your store (handles cases where participant joined before agent started). •

await ctx.wait_for_participant()

can detect first joins reliably. 5) Example flow (pseudo):

# on connect registered before ctx.connect()
@ctx.room.on("participant_connected")
async def on_participant_connected(p):
    prev = participants.get(p.identity)
    if prev and prev["sid"] != p.sid:
        # user rejoined with new SID
    participants[p.identity] = {"sid": p.sid, "last_seen": now(), "connected": True}

@ctx.room.on("participant_disconnected")
def on_participant_disconnected(p):
    participants[p.identity]["connected"] = False
    participants[p.identity]["last_seen"] = now()
    participants[p.identity]["last_sid"] = p.sid

6) Edge cases • Network blips may cause short disconnects — consider a debounce window before treating as a true drop. • If participants can share identities, use additional metadata to disambiguate. If you’d like, paste your current connect/disconnect handlers and I’ll show the exact code to detect rejoin with your session logic.

@ctx.room.on("participant_connected") def on_participant_connected(participant: Any) -> None: print(f"Participant connected event: {participant}") webhook_event = api.WebhookEvent( id=str(uuid.uuid4()), created_at=int(time.time()), event="participant_joined", participant=api.ParticipantInfo( identity=participant.identity, name=participant.name, metadata=participant.metadata, kind=participant.kind, attributes=participant.attributes, disconnect_reason=participant.disconnect_reason ), room=api.Room( name=ctx.room.name, metadata=ctx.room.metadata, num_participants=ctx.room.num_participants, num_publishers=ctx.room.num_publishers, ), ) print(f"participant_connected webhook_event: {webhook_event}") if metadata.service_endpoints.livekit_events_url: utils.send_webhook_event_background( metadata.service_endpoints.livekit_events_url, webhook_event ) # Cast to RemoteParticipant since participant_connected only fires for remote participants if isinstance(participant, rtc.RemoteParticipant): asyncio.create_task(async_on_participant_connected(participant)) # noqa: RUF006 @ctx.room.on("participant_disconnected") def on_participant_disconnected(participant: rtc.Participant) -> None: print(f"Participant disconnected event: {participant.identity}") webhook_event = api.WebhookEvent( id=str(uuid.uuid4()), created_at=int(time.time()), event="participant_left", participant=api.ParticipantInfo( identity=participant.identity, name=participant.name, metadata=participant.metadata, kind=participant.kind, attributes=participant.attributes, disconnect_reason=participant.disconnect_reason, ), room=api.Room( name=ctx.room.name, metadata=ctx.room.metadata, num_participants=ctx.room.num_participants, num_publishers=ctx.room.num_publishers, ), ) print(f"participant_disconnected webhook_event: {webhook_event}") if metadata.service_endpoints.livekit_events_url: utils.send_webhook_event_background( metadata.service_endpoints.livekit_events_url, webhook_event ) Now i want to call async_on_participant_connected only when participant rejoins and not on first join

You're nearly there — track participants' previous SIDs (or identities) and only call

async_on_participant_connected

when a connecting participant's identity was seen before with a different SID (i.e., they disconnected and now returned). Minimal in-memory approach: 1) Add a small store at module or entrypoint scope:

# identity -> last_seen_info
participants_state: dict[str, dict] = {}
# example value: {"sid": "ABC", "connected": True, "last_seen": 1680000000}

2) Update handlers:

@ctx.room.on("participant_connected")
def on_participant_connected(participant: Any) -> None:
    identity = participant.identity
    prev = participants_state.get(identity)

    # If we've seen this identity before and SID changed -> rejoin
    is_rejoin = bool(prev and prev.get("sid") != participant.sid)

    # Update store as connected
    participants_state[identity] = {
        "sid": participant.sid,
        "connected": True,
        "last_seen": time.time(),
    }

    if is_rejoin and isinstance(participant, rtc.RemoteParticipant):
        asyncio.create_task(async_on_participant_connected(participant))

@ctx.room.on("participant_disconnected")
def on_participant_disconnected(participant: rtc.Participant) -> None:
    identity = participant.identity
    # Mark disconnected but keep last SID for rejoin detection
    participants_state.setdefault(identity, {}) .update({
        "sid": participant.sid,
        "connected": False,
        "last_seen": time.time(),
    })

3) Handle initial participants present when agent starts: • After connect, iterate

ctx.room.remote_participants

and populate

participants_state

so those first-join events aren't treated as rejoins later. 4) Optional improvements: • Use a debounce window (ignore reconnects within N seconds) to avoid treating brief network blips as true rejoin. • Persist state across restarts if you need rejoin detection after agent redeploys. If you paste your

async_on_participant_connected

body and how you populate state at startup, I’ll show the exact lines to add.

is there another way?

Yes — here are alternative approaches you can use instead of (or alongside) the in-memory SID tracking I suggested. Pick the one that fits your architecture: • Initialize from current room state after connect a. Register handlers before connect. b. After

await ctx.connect()

, populate a set of identities present at start:

initial_identities = {p.identity for _, p in ctx.room.remote_participants.items()}

1. In

participant_connected

, treat as rejoin only if

identity in initial_identities

or

identity in disconnected_set

. This avoids treating the very first join (while your agent was offline) as a rejoin. • Use SID comparison but persist minimal history to survive restarts Store last SID per identity in a small persistent store (Redis, sqlite, etc.). On connect/check: • If stored SID exists and differs from current SID => rejoin. • Update store on disconnect. This is robust across agent restarts. • Use a short disconnect debounce window (ignore quick reconnects) On disconnect, record timestamp. On connect, if previous disconnect.timestamp < N seconds ago treat as ephemeral blip (ignore) or as rejoin based on your policy. • Leverage participant metadata or custom attributes If clients can set metadata/attributes on join (e.g., a monotonically-increasing session id or connection counter), compare that value: changed value → new session/rejoin. This avoids relying on SIDs. • Rely on webhook event ordering for durable rejoin detection Send participant_joined/left webhooks to your backend. Use your backend to correlate joins/leaves (with timestamps) and decide if a join is a rejoin. Good when you already use webhooks for analytics. Which of these sounds best for your setup (ephemeral in-memory, persistent store, metadata from clients, or webhook-based)? I can give a short ready-to-drop snippet for the option you pick.

async def async_on_participant_connected(
        participant: Any,
    ) -> None:
        if session.agent_state == "listening":
            await session.say(metadata.conversation_data.continuation_call_greeting)

    @ctx.room.on("participant_connected")
    def on_participant_connected(participant: Any) -> None:
        print(f"Participant connected event: {participant}")
        webhook_event = api.WebhookEvent(
            id=str(uuid.uuid4()),
            created_at=int(time.time()),
            event="participant_joined",
            participant=api.ParticipantInfo(
                identity=participant.identity,
                name=participant.name,
                metadata=participant.metadata,
                kind=participant.kind,
                attributes=participant.attributes,
                disconnect_reason=participant.disconnect_reason
            ),
            room=api.Room(
                name=ctx.room.name,
                metadata=ctx.room.metadata,
                num_participants=ctx.room.num_participants,
                num_publishers=ctx.room.num_publishers,
            ),
        )
        print(f"participant_connected webhook_event: {webhook_event}")
        if metadata.service_endpoints.livekit_events_url:
            utils.send_webhook_event_background(
                metadata.service_endpoints.livekit_events_url, webhook_event
            )
        # Cast to RemoteParticipant since participant_connected only fires for remote participants
        if isinstance(participant, rtc.RemoteParticipant):
            asyncio.create_task(async_on_participant_connected(participant))  # noqa: RUF006

    @ctx.room.on("participant_disconnected")
    def on_participant_disconnected(participant: rtc.Participant) -> None:
        print(f"Participant disconnected event: {participant.identity}")
        webhook_event = api.WebhookEvent(
            id=str(uuid.uuid4()),
            created_at=int(time.time()),
            event="participant_left",
            participant=api.ParticipantInfo(
                identity=participant.identity,
                name=participant.name,
                metadata=participant.metadata,
                kind=participant.kind,
                attributes=participant.attributes,
                disconnect_reason=participant.disconnect_reason,
            ),
            room=api.Room(
                name=ctx.room.name,
                metadata=ctx.room.metadata,
                num_participants=ctx.room.num_participants,
                num_publishers=ctx.room.num_publishers,
            ),
        )
        print(f"participant_disconnected webhook_event: {webhook_event}")
        if metadata.service_endpoints.livekit_events_url:
            utils.send_webhook_event_background(
                metadata.service_endpoints.livekit_events_url, webhook_event
            )

.....

    await ctx.connect()
    await session.say(initial_call_greeting, allow_interruptions=False)

Now what i want is the initial_call_greeting should run on the initial participant join and the for every participant rejoin after that i need continuation_call_greeting to run. But right now, the continuation_call_greeting runs on the initial join as well

You're close — the continuation greeting runs on initial join because your code treats every participant_connected the same. Make the distinction by recording which identities have already received the initial greeting (populate that set after connect), and only run

async_on_participant_connected

when the identity was previously seen and now has a new SID (or was disconnected). Minimal, ready-to-drop change: 1) Add state at top of entrypoint:

seen_identities: set[str] = set()
last_sid_by_identity: dict[str, str] = {}

2) After connecting and after you run the initial greeting, populate seen identities from current participants so those first-joins are considered "seen":

await ctx.connect()
# play initial greeting to the first participant as you already do
await session.say(initial_call_greeting, allow_interruptions=False)

# seed seen participants and SIDs so future joins are rejoins
for _sid, p in ctx.room.remote_participants.items():
    seen_identities.add(p.identity)
    last_sid_by_identity[p.identity] = p.sid

3) Update handlers to call continuation only on rejoin:

@ctx.room.on("participant_connected")
def on_participant_connected(participant: Any) -> None:
    identity = participant.identity
    prev_sid = last_sid_by_identity.get(identity)

    # is rejoin if we've seen identity before and SID changed
    is_rejoin = (identity in seen_identities) and (prev_sid is not None and prev_sid != participant.sid)

    # mark seen and update last SID
    seen_identities.add(identity)
    last_sid_by_identity[identity] = participant.sid

    if metadata.service_endpoints.livekit_events_url:
        # send webhook as you already do...
        ...

    if is_rejoin and isinstance(participant, rtc.RemoteParticipant):
        asyncio.create_task(async_on_participant_connected(participant))

4) Mark disconnects so short reconnects are tracked:

@ctx.room.on("participant_disconnected")
def on_participant_disconnected(participant: rtc.Participant) -> None:
    # keep last_sid_by_identity[identity] so rejoin detection works
    # optional: record disconnect time if you want debounce logic
    ...

Behavior: • First join after you connect → identity not in

seen_identities

, so only the initial greeting runs. • Later reconnect with different SID →

is_rejoin

true → runs continuation greeting. • If you need to ignore quick blips, store disconnect timestamp and require a minimum elapsed time before treating reconnect as rejoin. If you want, I can convert this into a copy-paste patch for your exact file (redacting secrets).

last_sid_by_identity - do we need this?

You don't strictly need

last_sid_by_identity

. If you only care about "first time after agent startup" vs "subsequent reconnects while agent is running," a single

seen_identities

set is enough. Minimal approach (no SID tracking): 1. After

await ctx.connect()

and after sending the initial greeting, seed seen identities:

seen_identities = set(p.identity for _, p in ctx.room.remote_participants.items())

1. In handlers:

@ctx.room.on("participant_connected")
def on_participant_connected(participant):
    identity = participant.identity
    is_rejoin = identity in seen_identities

    # Mark seen so future joins are treated as rejoins
    seen_identities.add(identity)

    if is_rejoin and isinstance(participant, rtc.RemoteParticipant):
        asyncio.create_task(async_on_participant_connected(participant))
    else:
        # initial join behavior already handled by initial greeting
        pass

@ctx.room.on("participant_disconnected")
def on_participant_disconnected(participant):
    # keep identity in seen_identities so future joins are treated as rejoin
    # optionally record disconnect timestamp if you want debounce logic
    ...

When to keep `last_sid_by_identity`: • Use it if you need to distinguish a true rejoin (same identity, new SID) vs a transient reconnect where SID may remain the same, or if you want to detect rejoin windows and ignore quick blips. • Use it if your agent restarts and you want more precise detection across restarts (persist SIDs externally). If you want, tell me whether you need to ignore quick blips or survive agent restarts and I’ll provide the exact snippet for that case.