https://livekit.io logo
Join Slack
Powered by
Writing Livekit tests for a flow with multiple age...
# ask-ai
n
Writing Livekit tests for a flow with multiple agents. Agent handoff is not captured
t
Thanks for reaching out, Kareem. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I can help with that. You mentioned agent handoff isn’t being captured in your LiveKit tests; I’m going to help you narrow down the cause and give the right assertions to use. Quick checklist and where to look first • Are you triggering the handoff in the test (for example by returning a tool result that instructs a handoff)? If the handoff never occurs at runtime it won’t be recorded. • Are you using mocked tools (
mock_tools
) or the real agent/tool logic in the test? Mocking changes how/when events are emitted. • Which assertion method are you using: 
is_agent_handoff()
vs 
contains_agent_handoff()
? They behave differently (next-event vs searching the result). What to try (assertions and example) 1. If you expect the handoff to be the very next event after the turn, use: ◦ await result.expect.next_event().is_agent_handoff(new_agent_type=BillingAgent) 1. If the handoff may appear later in the sequence, use: ◦ result.expect.contains_agent_handoff(new_agent_type=BillingAgent) 1. Confirm the chat context contains an AgentHandoff item (it has 
old_agent_id
and 
new_agent_id
fields). If you want, assert on those fields explicitly after locating the handoff item. What I need from you to diagnose further Please share the following so I can pinpoint the problem: 1. The code that triggers the handoff (for example the tool return or agent logic you expect to cause the handoff). 2. The failing test code (the exact assertions you’re using). 3. Whether you’re running with 
mock_tools
or real tools/agents in this test. 4. If possible, a short transcript or the test run output showing the events emitted (so I can see whether an AgentHandoff event is present but not matched by your assertion). If you prefer, paste a minimal runnable snippet of the test + the handoff trigger and I’ll point out the exact change to capture the handoff. Next steps after you send the snippets • I’ll check whether the handoff event is emitted and if your assertion is targeting the correct event position. • If the event exists but your assertions miss it, I’ll suggest the correct assertion or show how to inspect the result’s events. • If the event never occurs, I’ll help adjust the trigger (mocked tool return or agent response) so LiveKit records the AgentHandoff. You’re not far — once I can see the trigger + test assertions I’ll give the exact code fix. Sources: Testing and evaluation | LiveKit Documentation | Agents and handoffs | LiveKit Documentation
n
We currently have an agent with has a loop of collecting facts. each loop run triggers a task to ask a question
t
Thanks — this helps. You said your agent runs a loop to collect facts and each loop run triggers a task to ask a question. To help you capture and assert the agent handoff in your tests, I need a few specifics so I can give an exact fix. Could you share the following? 1. The code that triggers the task/question inside the loop (the part that should cause the handoff). 2. The test code where you run the session and assert events (show the assertion you use now). 3. Are you using 
mock_tools
or real tools/agents in the test? 4. A short excerpt of the test run output or event transcript (if available) showing the emitted events — especially around where you expect the handoff. Quick tips that often resolve missed handoff captures • If the handoff is triggered by a tool return, ensure your mocked tool returns the exact structure that causes an AgentHandoff (include `old_agent_id`/`new_agent_id` if your test needs them). • Use 
await result.expect.next_event().is_agent_handoff(new_agent_type=YourAgent)
when the handoff should be the immediate next event. • Use 
result.expect.contains_agent_handoff(new_agent_type=YourAgent)
when the handoff may appear later in the event sequence. • If each loop iteration emits a task event before the handoff, the handoff may not be the next event — prefer 
contains_agent_handoff
or iterate 
result.expect.next_event()
until you find the handoff. • You can also inspect the full 
result.events
or chat context for an 
AgentHandoff
item and assert on its 
old_agent_id
/ 
new_agent_id
directly. If you paste a minimal snippet of your loop + trigger and the failing assertion, I’ll point out the exact change to capture the handoff reliably (mock return shape, which assertion to use, or how to scan events).
n
agent handoff is done in task complete. The result events only include the FunctionCallOutputEvent as the last event but not handoff from Task to Agent
t
Thanks — this is helpful context. You said the task completes (you see the 
FunctionCallOutputEvent
) but the handoff from the task back to the agent isn’t present in 
RunResult.events
. I’ll explain why that can happen and give concrete ways to assert the handoff in tests plus targeted questions so I can give a precise fix. Why you may not see an AgentHandoff event in RunResult • When a task calls 
self.complete(...)
, control returns to the parent agent, but that return is often internal and does not always surface as a separate event in the 
RunResult.events
stream. The session state changes and the handoff is recorded in the chat context/history, but you might only see the tool/function output as the last emitted event. • If the parent agent resumes without emitting a new assistant message, the only visible event remains the function/tool output — no explicit handoff event appears in 
RunResult
. • Agent handoffs are recorded as an 
AgentHandoff
item in the conversation history/chat context, not guaranteed as a standalone 
RunResult
event. What to test instead (practical options) Use one of these approaches depending on what you need to validate: • Option A — Assert the parent agent resumes (recommended) a. Assert the task/tool call and its output occurred. b. Assert the next assistant message is from the parent agent and contains the expected behavior/content. Example flow: 1. expect function call 2. expect function output 3. expect next event is assistant message (confirm it’s the parent agent resuming) • Option B — Inspect the session/chat history for an AgentHandoff item ◦ After the run, check 
session.history
(or 
chat_ctx
) for an 
AgentHandoff
item and assert its 
old_agent_id
/ 
new_agent_id
. ◦ This directly verifies the handoff metadata even if no event was emitted. • Option C — Make the handoff observable (only if you need an explicit event) ◦ In your 
AgentTask.on_exit()
or right when you 
complete()
, generate a short assistant message indicating return of control. That will produce an observable assistant event you can assert in 
RunResult
. ◦ This is more of a UX/test signal and not required for correct behavior. Concrete assertions to try • If handoff should be immediate next event: ◦ await result.expect.next_event().is_agent_handoff(new_agent_type=YourAgent) — use only if your flow emits such an event. • If handoff may be later: ◦ result.expect.contains_agent_handoff(new_agent_type=YourAgent) • To assert resume instead of explicit handoff: ◦ await result.expect.next_event().is_function_call(...) ◦ await result.expect.next_event().is_function_call_output() ◦ await result.expect.next_event().is_message(role="assistant") # validate content/intent matches parent agent Targeted questions so I can give an exact code fix 1. Can you paste the task’s completion code (where you call 
self.complete(...)
) and the task class’ 
on_exit
if present? 2. Share the exact test snippet you’re running (the 
session.run(...)
call and the assertions you used). 3. Are you using 
mock_tools
or real tools in this test? 4. If possible, paste the 
RunResult.events
list (or a short transcript) showing the last few events around the task completion. If you paste those snippets I’ll: • Tell you whether the handoff should be visible as an event in your current setup. • Suggest the exact assertion to use (or the minimal change to the task to produce a visible handoff event). • Provide a minimal corrected test snippet you can drop into your suite. You’re on the right track — once I see the task completion code + test assertions I’ll give the precise change to reliably capture the handoff. Sources: Tasks and task groups | LiveKit Documentation | Agents and handoffs | LiveKit Documentation | Collect recording consent with tasks | LiveKit Documentation | Testing and evaluation | LiveKit Documentation
n
I am not keen on the handoff event as much as I am keen on capturing the agent speech when a new task is triggered. Here is the current flow: Agent => Task A => Agent => Task B => Agent. every Task starts with an agent question.
Copy code
await initial_result.expect.contains_message(role="assistant").judge(
            judge_llm,
            intent="Greets the patient by name and asks what is the reason for their call",
        )

turn1 = await session.run(user_input="User Speech")
# Task completed
Copy code
#Task two starts

#Agent speech not captured! but user speech captured. 

        # Turn 2 — 
        turn2 = await session.run(user_input="User Speech")
        await turn2
also I see RuntimeError: cannot schedule new speech, the speech scheduling is draining/pausing exception
t
Thanks — this is very helpful. I’ll focus on two things you care about: (A) reliably capturing the agent’s speech at the start of each task, and (B) fixing the RuntimeError about speech scheduling. I’ll also ask for a couple of small snippets so I can give the exact change. Why the agent question may not be captured • If a task is started but the parent agent (or the task) does not emit a visible assistant message before the test run ends, the test’s 
RunResult.events
will not include that assistant message. You’ll often see only the user input and the task/tool output. • If speech is scheduled but not actually produced because the system is paused/draining, the assistant message will not appear in the event stream. • When you call 
session.run(...)
for the next user turn immediately after a task completes, the agent’s scheduled speech may still be in the speech queue (or blocked), so the speech event is not emitted before the next run starts. How to capture agent speech at the start of each task Use one or a combination of these approaches depending on your architecture: • Option 1 — Assert the agent message that starts the task is emitted before you call the next `session.run()`: a. After the task is triggered, wait for the assistant message that asks the question: • await result.expect.next_event().is_message(role="assistant") a. Only then proceed to call the next 
session.run(user_input=...)
. This ensures you capture the agent’s prompt as an event before the next user turn begins. • Option 2 — Poll the result/event stream until the assistant message appears: ◦ If the message may be emitted asynchronously after several micro-events (task logs, tool outputs), iterate 
await result.expect.next_event()
until you see an assistant message with the question. ◦ Alternatively use 
result.expect.contains_message(role="assistant", predicate=...)
to search the result for the question. • Option 3 — Make the task produce the question as a guaranteed visible event when it starts: ◦ Emit the agent question from the task startup (for example in 
on_enter()
or immediately after task initiation) using the session API that produces a message rather than only scheduling speech. ◦ Example: call a method that generates an assistant message (not only speech scheduling) so it’s placed into 
RunResult.events
. • Option 4 — Inspect session/chat history for assistant messages ◦ If 
RunResult.events
misses the event, 
session.history
or 
chat_ctx
usually contains the assistant message items. Assert against that history to validate the agent question exists. Addressing the RuntimeError: cannot schedule new speech (speech scheduling is draining/pausing) • That RuntimeError means the speech scheduler is in a draining or paused state and refuses new speech requests. This typically happens when: ◦ The session is shutting down, draining, or transitioning state. ◦ Speech was scheduled but not flushed before you tried to schedule more speech. ◦ Tests are starting subsequent turns before the scheduler has finished processing previous speech items. • Fix patterns: a. Ensure the agent’s speech is flushed/emitted before starting the next user turn. Wait for the assistant message event (see Option 1 above). b. If the Task schedules speech internally, avoid scheduling overlapping speech requests. Make sure each scheduled speech completes or is awaited before scheduling the next. c. If you control the task code, change it to emit a text message event (so the test can assert on it) and then schedule speech from that message if needed. d. If the runtime provides an explicit flush/await on the speech queue, call that (e.g., await session.flush_speech() or similar) before the next turn. If you don’t have this API, waiting for the assistant message event is the practical substitute. Concrete test pattern you can try now 1. Run turn1 and assert the task completes and the agent response for task start appears: ◦ await initial_result.expect.contains_message(role="assistant").judge(...) ◦ turn1 = await session.run(user_input="User Speech") ◦ # After turn1 completes, explicitly wait for the assistant question that begins Task B: ▪︎ await turn1.expect.next_event().is_message(role="assistant") # validate content/intent 1. Only then call turn2: ◦ turn2 = await session.run(user_input="User Speech") ◦ await turn2.expect.contains_message(role="assistant").judge(...) If the assistant message is produced by the Task startup rather than by the parent agent, ensure the task code emits a message event immediately when starting. Targeted info I need to give a precise code fix Please paste these small snippets: 1. The Task A and Task B startup code (especially where they schedule or generate the initial question — e.g., 
on_enter()
, 
on_start()
, or where you call 
self.schedule_speech(...)
or 
self.complete(...)
). 2. The test code around turn1 and turn2 (the 
session.run(...)
calls and your assertions). 3. Whether you use 
mock_tools
or real tools, and whether speech is disabled in tests (some tests mock out speech). 4. The full traceback of the RuntimeError (so I can see which function raises it). What I’ll do after you paste those • If the task only schedules speech, I’ll show how to emit an assistant message event on task start (small change) so tests can assert reliably. • If the scheduler is being drained/paused because of ordering, I’ll show how to wait for/flush the speech queue or change the test ordering to avoid the RuntimeError. • I’ll provide the exact minimal test snippet you can drop in to reliably capture the agent question for each task. You’re close — once I can see the task startup code and the small test snippet, I’ll give the exact code change to capture the agent’s question and remove the RuntimeError.
n
in the task, we have session.say in the on_enter
t
Thanks — this is perfect context. You said the task calls 
session.say()
in 
on_enter
, you don’t see the agent question in the test events, and you get a RuntimeError about speech scheduling being draining/pausing. Below I’ll explain what’s happening, give concrete test patterns you can use immediately, and suggest minimal code changes to both capture the speech reliably and avoid the scheduler error. Why this happens (short) • 
session.say()
schedules speech output, but scheduling is asynchronous: the speech may be queued and not turned into an observable assistant message event before your test advances to the next turn. If no assistant message is emitted (text event) the test only sees user input and function/tool outputs. • The RuntimeError ("cannot schedule new speech, the speech scheduling is draining/pausing") occurs when 
session.say()
is called while the speech queue is draining or the session is closing/transitioning; new speech requests are rejected in that state. Concrete ways to capture the agent question (pick one) • Option A — Wait for the assistant message event before starting the next turn (recommended) a. After triggering the task (first run), wait for the assistant message that contains the task question: • await result.expect.next_event().is_message(role="assistant") a. Then call the next 
session.run(...)
for the user reply. This ensures the scheduled speech was surfaced as a message event (or that the agent produced a message) before you proceed. • Option B — Make the task produce an immediate visible assistant message instead of only speech ◦ In 
on_enter
, emit a message event (text) that represents the question (this is deterministic and test-friendly), then optionally schedule speech for playout: ▪︎ session.send_message("What is the reason for your call?") # or the API that creates a message event ▪︎ session.say("What is the reason for your call?") # optional speech playout ◦ Tests assert the presence of the message event (fast, reliable). • Option C — Await the SpeechHandle/playout completion if you must rely on speech ◦ If 
session.say()
returns a 
SpeechHandle
, await its completion before moving on: ▪︎ handle = session.say("...") ▪︎ await handle.wait_for_playout() ◦ This guarantees speech finished (and often that corresponding events were emitted) before the next run. • Option D — Inspect session history/chat_ctx for the assistant message ◦ If events aren’t in 
RunResult.events
, check 
session.history
for assistant messages emitted during 
on_enter
. Assert against history items for the task question. How to avoid the "cannot schedule new speech" RuntimeError • Ensure you do not call 
session.say()
while the scheduler is draining or the session is closing: ◦ Guard speech calls: ▪︎ if not session.is_closing: session.say(...) • Only schedule speech once you know the session is active (on enter is fine if the session isn’t shutting down). • Avoid scheduling overlapping speech during transitions: ◦ Don’t trigger the next task/user turn until the current speech has been emitted or awaited (use Option A or C above). • If your tests quickly start a new run right after a task completes, wait for the assistant message or speech playout before calling the next 
session.run(...)
. This prevents the scheduler from being overwhelmed or put into draining state. Minimal code patterns to use in tests 1) Wait-for-assistant pattern (recommended)
Copy code
turn1 = await session.run(user_input="User Speech")
# Wait for the assistant message from Task B on_enter
await turn1.expect.next_event().is_message(role="assistant")
# Now proceed with the user's reply to task question
turn2 = await session.run(user_input="User Speech")
2) Emit text message in task on_enter (most robust)
Copy code
# inside Task.on_enter()
session.send_message("What is the reason for your call?")
# optionally:
session.say("What is the reason for your call()")  # schedule speech AFTER creating message
3) Await speech playout when using speech-only
Copy code
# inside Task.on_enter()
handle = session.say("What is the reason for your call?")
# in test, after turn finishes:
await handle.wait_for_playout()
# then start next session.run(...)
Targeted suggestions based on your setup • If you can change task code: emit a message event in 
on_enter
(Option B). Tests can assert against messages deterministically and you avoid scheduling races. • If you cannot change task code and must rely on `session.say()`: return and await the 
SpeechHandle
in tests (Option C) or wait for the assistant message event (Option A) which may be produced when speech is scheduled. • If 
session.say()
is sometimes called during shutdown/drain, add a guard 
if not session.is_closing:
around it to prevent the RuntimeError. Small check-list for you to try now • Modify your test to wait for the assistant message after the task starts (use 
await turn1.expect.next_event().is_message(role="assistant")
) before calling 
session.run()
for the next user input. • If the task only schedules speech, change 
on_enter
to also create a message event (or have 
session.say()
return a handle you can await). • Add a guard around 
session.say()
to avoid scheduling during session shutdown. If you want, paste these two tiny snippets and I’ll give the exact lines to add: 1. Your Task.on_enter (the part with 
session.say(...)
). 2. The test around turn1/turn2 (the 
session.run(...)
calls and assertions). With those I’ll give the minimal code edits to either emit a visible message event or to wait properly for the speech handle so your test reliably captures the agent question and the RuntimeError goes away. Sources: How to detect when an agent has finished speaking | LiveKit
n
livekit.agentsagent.py728 The asyncio.Task finished before FactsQuestionnaire was completed.
t
Thanks — this is helpful detail. You’re seeing two related runtime issues: the missing agent speech event and the asyncio warning "The asyncio.Task finished before FactsQuestionnaire was completed." Both point to asynchronous work (speech playback / reply generation) not finishing before the task ends or a handoff occurs. Below I’ll explain the root causes, give concrete fixes (minimal code changes), and list a short checklist and targeted questions so I can give a precise patch if needed. Why this is happening (short) • Calling 
session.say()
/ 
session.generate_reply()
schedules asynchronous speech/playout. If the task calls 
self.complete(...)
(or the agent hands off) before that speech handle finishes, the speech scheduler may be drained/paused and new speech requests are rejected. • The asyncio warning means the background asyncio.Task that was running the task’s async work returned while other awaited operations (speech handles, reply generation) were still pending. That leaves speech unfinished and produces the RuntimeError later. Minimal, reliable fixes 1. Await the speech/generate_reply handle before completing the task ◦ When you call 
session.say()
or 
session.generate_reply()
, capture the returned 
SpeechHandle
/ coroutine and 
await
it (or call 
await handle.wait_for_playout()
if that API is provided) before calling 
self.complete(...)
. Example pattern:
Copy code
async def on_enter(self):
       handle = self.session.generate_reply(instructions="What is the reason for your call?")
       await handle.wait_for_playout()  # or `await handle`
       # Now safe to continue; the message/speech is emitted

   async def some_task_action(self):
       # finish processing
       self.complete(result)
1. Use allow_interruptions=False when you need speech to finish ◦ If the system supports it, pass 
allow_interruptions=False
to ensure the reply/speech finishes without being interrupted by subsequent turns. 1. Guard speech calls during shutdown/drain ◦ Protect speech scheduling with a check so you don’t try to schedule while session is closing:
Copy code
if not self.session.is_closing:
       handle = self.session.say("...")
1. Emit a deterministic message event, then schedule speech ◦ For tests, it’s often simplest to create a text message event immediately (so tests can assert it) then call 
session.say()
for audio playout. This avoids race conditions between scheduling and test assertions. Example in task `on_enter`:
Copy code
# Emit message event immediately for test visibility
   self.session.send_message("What is the reason for your call?")
   # Then schedule speech (optional), and await it if needed
   handle = self.session.say("What is the reason for your call?")
   await handle.wait_for_playout()
Concrete test patterns to avoid races • Wait for assistant message before calling next run:
Copy code
turn1 = await session.run(user_input="User Speech")
  # Ensure the task's agent question was emitted
  await turn1.expect.next_event().is_message(role="assistant")
  # Now safe to run next user turn
  turn2 = await session.run(user_input="User Speech")
• If using speech-only, have your test await the SpeechHandle or use 
await handle.wait_for_playout()
to ensure playout finished before the next turn. Short checklist to apply now • Update the task to await any speech/reply handles in 
on_enter
(or before calling 
complete()
). • If you cannot await in 
on_enter
, emit a text message first (for test assertions) and only then schedule/await speech. • Add 
if not session.is_closing:
guards around 
session.say()
if shutdown races are possible. • Change your tests to wait for the assistant message or speech handle completion before starting the next user turn. Targeted questions — paste these so I can give exact lines to change 1. The Task class code for FactsQuestionnaire — especially 
on_enter
, where you call 
session.say()
and where/when you call 
self.complete(...)
. 2. The test snippet around turn1/turn2 (the 
session.run(...)
calls and assertions). 3. If 
session.say()
returns a handle in your runtime, paste the exact call so I can show how to await it (or tell me if you already have access to 
wait_for_playout()
). If you paste the two small snippets (task on_enter + the test around turn1/turn2), I’ll return a minimal patch you can drop in that: • awaits speech properly, • removes the asyncio warning, • and guarantees the agent question is visible to your test (either as a message event or via awaited speech playout). Sources: Tasks and task groups | LiveKit Documentation | Agent speech and audio | LiveKit Documentation | Transcriber | LiveKit Documentation
n
Copy code
self.session.say(ack_message, allow_interruptions=False)

        await context.wait_for_playout()
        self.complete(result)
Still triggers ERROR livekit.agentsagent.py728 The asyncio.Task finished before FactsQuestionnaire was completed. =========================================================================================================================== short test summary info ============================================================================================================================ FAILED tests/workflows/sns/test_respiratory_triage.py::test_respiratory_triage_happy_path - RuntimeError: cannot schedule new speech, the speech scheduling is draining/pausing
t
Thanks — and I appreciate you sticking with this. You’ve shown that you already call 
self.session.say(..., allow_interruptions=False)
and then 
await context.wait_for_playout()
before 
self.complete(result)
, yet you still get: • The asyncio warning: "The asyncio.Task finished before FactsQuestionnaire was completed." • RuntimeError: "cannot schedule new speech, the speech scheduling is draining/pausing" These signals mean some asynchronous speech or related background work is still pending or being scheduled while the task finishes or during a handoff, and the scheduler has entered a draining/closing state. Let’s focus on concrete checks and minimal changes to make the flow deterministic in tests. What’s likely happening (short) • 
session.say()
schedules speech; it may return a SpeechHandle or schedule an internal task. • 
context.wait_for_playout()
may not be awaiting the exact handle created by 
session.say()
(it may await a different queue drain or return earlier). • Meanwhile, another part of the task/agent lifecycle (handoff, session shutdown, or test code) causes the asyncio.Task driving the task to finish, leaving outstanding speech work and causing the scheduler to be in a draining state. Subsequent attempts to schedule speech fail with RuntimeError. Concrete steps to fix (try in this order) 1) Capture and await the speech handle returned by session.say directly • If 
session.say()
returns a handle, keep that handle and await it (or call its 
wait_for_playout()
), not just 
context.wait_for_playout()
. Example:
Copy code
handle = self.session.say(ack_message, allow_interruptions=False)
await handle.wait_for_playout()   # or `await handle` if the API awaits directly
self.complete(result)
Why: awaiting the exact handle ensures the specific speech task finished before you complete the task. 2) If session.say does not return a handle, ensure you await the exact playout API used by your runtime • Confirm that 
context.wait_for_playout()
awaits the same queue item created by 
session.say()
. If not, replace it with the proper handle await or an explicit session API that flushes speech. 3) Ensure no other background speech is scheduled after you await playout and before complete • Search the task for other 
session.say()
/ 
session.generate_reply()
calls. Every such call must be awaited (its handle) before 
self.complete(...)
. • If any code schedules speech in response to events (callbacks, on_exit, or parent agent), make sure those either finish or are prevented when the task completes. 4) Avoid racing with test code: wait for emitted assistant message or playout in test before starting next run • In the test, after the run that triggers the task, assert and wait for the assistant message (or speech playout) before calling the next `session.run(...)`:
Copy code
turn1 = await session.run(user_input="User Speech")
await turn1.expect.next_event().is_message(role="assistant")
# now safe to run turn2
turn2 = await session.run(user_input="User Speech")
5) Guard speech scheduling when session is closing • Add a check to avoid scheduling speech during shutdown:
Copy code
if not self.session.is_closing:
    handle = self.session.say(ack_message, allow_interruptions=False)
    await handle.wait_for_playout()
6) If you cannot await playout synchronously in on_enter, emit a message event first then schedule speech • Create a message event that tests can assert deterministically:
Copy code
self.session.send_message(ack_message)  # immediate event for tests
handle = self.session.say(ack_message, allow_interruptions=False)
await handle.wait_for_playout()
self.complete(result)
What I need from you to give the exact one-line patch Please paste these two small snippets: 1. The FactsQuestionnaire task code (full or relevant parts: 
on_enter
, any 
session.say
/ 
generate_reply
calls, and where 
self.complete(...)
is called). 2. The test snippet around turn1/turn2 (the 
session.run(...)
calls and any event assertions). With those I will: • Mark exactly which await/guard lines to add. • Show the minimal change so the asyncio warning disappears and the Scheduler RuntimeError stops. • Provide a short test change (wait for assistant message / await handle) to make the test deterministic. If you prefer, paste the full traceback for the RuntimeError too — that helps locate where scheduling is attempted while the scheduler is draining.
n
Copy code
turn1 = await session.run(user_input="User Speech")
await turn1.expect.next_event().is_message(role="assistant")
next event is a function call
t
Thanks — that helps. You said 
await turn1.expect.next_event().is_message(role="assistant")
but the next event is a function/tool call instead of the assistant message. That confirms the task’s speech is being scheduled but the function/tool call is emitted before the assistant message event is visible to the test. To give the exact minimal fix, I need two small snippets so I can pinpoint the race and recommend the precise await/emit change: 1. The FactsQuestionnaire task code (relevant parts): ◦ The full 
on_enter
method (where you call 
session.say(...)
). ◦ Any other 
session.say()
or 
session.generate_reply()
calls in the task. ◦ Where/when 
self.complete(...)
is called. 1. The test snippet around turn1/turn2: ◦ The 
turn1 = await session.run(...)
call and the immediate assertions you run after it. ◦ The 
turn2 = await session.run(...)
call. ◦ Any test-level setup for mocking tools or speech. While you paste those, try these quick checks/changes — often one of these is the root cause and will fix the ordering immediately: • Capture and await the exact speech handle returned by `session.say()`: ◦ Replace ▪︎ 
self.session.say(ack_message, allow_interruptions=False)
▪︎ 
await context.wait_for_playout()
◦ With ▪︎ 
handle = self.session.say(ack_message, allow_interruptions=False)
▪︎ 
await handle.wait_for_playout()
(or 
await handle
if that’s the API) ◦ This ensures you wait for the specific speech scheduled by this call, not a general queue drain. • If 
session.say()
does not return a handle in your runtime, emit a synchronous message event before scheduling speech: ◦ 
self.session.send_message(ack_message)
# immediate event the test can assert ◦ 
handle = self.session.say(ack_message, allow_interruptions=False)
◦ 
await handle.wait_for_playout()
◦ 
self.complete(result)
• Ensure you do not call 
self.complete(result)
until after the awaited handle returns (and after any other outstanding speech handles in the task are awaited). • In your test, if the function/tool call is expected to happen before the assistant message, search the result events rather than expecting the assistant to be immediate: ◦ Iterate 
await turn1.expect.next_event()
until you encounter the assistant message, or use 
turn1.expect.contains_message(role="assistant")
. Why I need the snippets: The most common root causes are: • 
context.wait_for_playout()
is not awaiting the same internal handle created by 
session.say()
. • 
self.complete()
or a handoff is being invoked while some speech scheduling task still exists or while another tool triggers immediately after the task starts. • The task emits a function/tool call as part of startup logic before it generates the assistant message. Paste the two snippets and I’ll return: • The single-line change you need (awaiting handle vs context.wait_for_playout or adding send_message). • The minimal test change if needed (wait for message or scan events). • A short explanation of why that change fixes the asyncio/task scheduler error you’re seeing.
n
Copy code
on_enter:

    async def on_enter(self) -> None:
        handle = self.session.say(text= self.question_text, allow_interruptions=False)
        await handle.wait_for_playout()
Copy code
if ack_message:
            handle = self.session.say(ack_message, allow_interruptions=False)
            await handle.wait_for_playout()

        self.complete(result)
t
Thanks — this is very clear and helpful. You’re already awaiting the speech handle in both places, so the remaining race is likely elsewhere: something is scheduling speech (or triggering work that schedules speech) after you await and before the task fully finishes, or the session is transitioning to draining concurrently. Let’s narrow it down and make a small, safe change to remove the race and guarantee the agent question is observable in tests. What’s happening now (short) • Your code awaits the playout handle, which is correct. • The asyncio warning and the RuntimeError mean some async work that schedules speech is still pending (or new speech is attempted) while the task/agent is finishing or the session is draining. • In tests the next visible event is a function call (tool) before the assistant message — the assistant message is scheduled but not visible at the time your test checks next_event(). Immediate, minimal changes to make the flow deterministic Apply these three small changes together — they are safe and fix the common causes: 1) Add an explicit test-visible message event before scheduling speech • In on_enter, emit a chat message synchronously so tests can assert it immediately:
Copy code
# inside on_enter
self.session.send_message(self.question_text)   # immediate event for tests
handle = self.session.say(text=self.question_text, allow_interruptions=False)
await handle.wait_for_playout()
Why: Tests observe message events reliably; this avoids relying solely on speech scheduling. 2) Ensure no speech is scheduled after complete: move complete only after ALL awaits • You already await, but verify there are no other places scheduling speech (on_exit, callbacks). If any exist, await those handles too before calling complete. Example:
Copy code
if ack_message:
    handle = self.session.say(ack_message, allow_interruptions=False)
    await handle.wait_for_playout()

# ensure no other speech handles pending here
self.complete(result)
Why: Any stray speech scheduled after complete can trigger scheduler draining. 3) Add a defensive guard around scheduling to avoid calls during drain • Protect calls that might run during shutdown/handoff:
Copy code
if not self.session.is_closing:
    handle = self.session.say(..., allow_interruptions=False)
    await handle.wait_for_playout()
Why: Prevents attempts to schedule speech when the scheduler is refusing new tasks. Test-side changes (one-line) — ensure you wait for the message event or scan for it • Prefer this in test after turn1:
Copy code
turn1 = await session.run(user_input="User Speech")
await turn1.expect.contains_message(role="assistant")   # searches events for assistant message
# then proceed to turn2
turn2 = await session.run(user_input="User Speech")
Why: 
contains_message
finds the assistant message even if it’s not the immediate next event (useful when function/tool events appear first). Targeted checks to find remaining race sources • Search your task and agent code for any other 
session.say()
, 
generate_reply()
, 
session.send_message()
or asynchronous callbacks (including 
on_exit
) that might schedule speech after your awaited handles. • Check any tool/tool-mock callbacks invoked on task start/complete; they may schedule speech or make calls that lead to speech scheduling. • If any background tasks are spawned, ensure they’re awaited or cancelled before calling 
self.complete()
. If this still fails, please paste: 1. The full FactsQuestionnaire task (on_enter, on_exit, any tool callbacks) so I can spot stray speech calls. 2. The small test snippet (turn1/turn2) you use now. 3. The full traceback of the RuntimeError (the frame where scheduling is attempted) — that tells me which call is trying to schedule after draining. If you prefer, start with adding the immediate 
send_message(...)
in 
on_enter
and change your test to 
contains_message(...)
. That combination usually makes tests stable immediately and prevents the scheduler error by avoiding timing races. I can craft the exact one-line edits to your task and test as soon as you paste the two snippets — or I can produce the patch based on the snippets you provided if you want me to. Sources: Agent speech and audio | LiveKit Documentation | How to detect when an agent has finished speaking | LiveKit | Testing and evaluation | LiveKit Documentation
n
=========================================================================================================================== short test summary info ============================================================================================================================ FAILED tests/workflows/sns/test_respiratory_triage.py::test_respiratory_triage_happy_path - TypeError: object ChatMessageAssert can't be used in 'await' expression
t
You're already doing most of the right things; the errors you're seeing come from two separate issues: 1) Using await on an assertion object (ChatMessageAssert) — that's why you get TypeError: object ChatMessageAssert can't be used in 'await' expression. • Fix: don't await the assertion itself; only await 
.judge(...)
when using LLM judgement. Example — wrong:
Copy code
# raises TypeError because contains_message(...) returns an assert object (not awaitable)
   await result.expect.contains_message(role="assistant", content="shortness of breath")
Example — correct:
Copy code
# synchronous assertion object; no await
   result.expect.contains_message(role="assistant", content="shortness of breath")

   # if you want fuzzy/LLM-based judgement, await judge()
   await result.expect.contains_message(role="assistant").judge(
       judge_llm, intent="Asks about shortness of breath"
   )
2) Speech scheduling / race conditions around task completion — the RuntimeError and asyncio warning • You correctly await the speech handle in 
on_enter
, but scheduling races can still occur if: ◦ other code schedules speech after your awaited handle, ◦ 
self.complete(...)
triggers cleanup/drain while other speech tasks remain, ◦ or the test advances to the next run before the assistant message is observable. Minimal, reliable changes to make tests deterministic 1. Emit a test-visible message event immediately in on_enter, then schedule speech:
Copy code
# inside FactsQuestionnaire.on_enter
      self.session.send_message(self.question_text)   # immediate, test-observable
      handle = self.session.say(text=self.question_text, allow_interruptions=False)
      await handle.wait_for_playout()
• Rationale: message events are deterministic and appear in `RunResult.events`; this avoids relying solely on TTS timing. 1. Capture and await the exact handle returned by 
say()
(you already do this; keep it).
Copy code
handle = self.session.say(ack_message, allow_interruptions=False)
      await handle.wait_for_playout()
      self.complete(result)
• Rationale: ensure the specific scheduled speech finishes before completing the task. 1. Add a defensive guard to avoid scheduling speech during shutdown/drain:
Copy code
if not self.session.is_closing:
          handle = self.session.say(..., allow_interruptions=False)
          await handle.wait_for_playout()
• Rationale: prevents attempts to schedule speech while the scheduler is refusing new tasks. 1. Update the test to search for the assistant message (not necessarily the immediate next event):
Copy code
turn1 = await session.run(user_input="User Speech")
      # Use contains_message to find the assistant question even if a function call appears first
      result.expect.contains_message(role="assistant")         # no await
Or, if you want LLM-based verification:
Copy code
await turn1.expect.contains_message(role="assistant").judge(
          judge_llm, intent="Greets the patient by name and asks reason for call"
      )
Why these changes fix your errors • The TypeError disappears because you stop awaiting a non-awaitable assert object. • The RuntimeError and asyncio warning disappear because: ◦ every speech scheduling is awaited before task completion, ◦ message events make the assistant question visible to tests immediately, ◦ guarded scheduling avoids calling say() while the scheduler is draining. If you still see the RuntimeError after these changes, please paste: • The full FactsQuestionnaire task (on_enter, on_exit and any callbacks) so I can spot any stray 
say()
calls. • The test snippet around turn1/turn2 you’re running now (so I can suggest the exact replacement assertions). I’ll then give the one-line edits to the task and the minimal test adjustment to make your tests stable. Sources: Testing and evaluation | LiveKit Documentation | How to detect when an agent has finished speaking | LiveKit
n
Copy code
turn1 = await session.run(user_input="Agent speech")
        await turn1.expect.contains_message(role="assistant").judge(
            judge_llm,
            intent="Thanks the patient and explains that some questions will be asked to understand their situation",
        )


        # Turn 2 — Patient denies severe shortness of breath
        turn2 = await session.run(user_input="No")
        await turn2.expect.contains_message(role="assistant").judge(
            judge_llm, intent="Short acknowledgment message"
        )


        # Turn 3 — Patient denies coughing blood
        turn3 = await session.run(user_input="No, ")
        await turn3.expect.contains_message(role="assistant").judge(
            judge_llm, intent="Short acknowledgment message"
        )
Now this test passes but it does not capture the agent question at a new turn
t
Thanks — this is clear. You’re nearly there. The root cause is timing: your task schedules speech in 
on_enter
, but the message event isn’t always visible in the test event stream before the next turn or tool events occur. That’s why your test passes (facts are captured) but doesn’t see the agent question. Two safe, minimal options to make the agent question reliably visible in tests: Option A — Emit a synchronous message event (recommended) • Add the assistant question directly into the chat context before scheduling speech. This makes the message immediately present in 
RunResult.events
. Example change in on_enter:
Copy code
self.session.chat_ctx.add_message(role="assistant", content=self.question_text)
handle = self.session.say(text=self.question_text, allow_interruptions=False)
await handle.wait_for_playout()
• Test: use 
await turn1.expect.contains_message(role="assistant").judge(...)
(no await on contains_message itself). Option B — Ensure speech handle completion and test waits for it • If you prefer not to add a message, ensure every 
say()
returns a handle you await, and make the test scan for the assistant message (or await playout). • In your test, use 
turn1.expect.contains_message(...)
instead of assuming it’s the immediate next event. Additionally — defensive guard to prevent scheduling during drain:
Copy code
if not self.session.is_closing:
    handle = self.session.say(...)
    await handle.wait_for_playout()
Why this fixes your failures • Adding the message synchronously guarantees the agent question is in the conversation history and discoverable by 
contains_message
. • Awaiting the exact speech handle prevents premature task completion and avoids "draining/pausing" RuntimeErrors. • Using 
contains_message
(not awaiting it) with 
.judge()
awaited ensures correct assertion usage and removes the TypeError. If you want, I can: • Provide the exact one-line edits to your 
on_enter
and the minimal test line to change. • Or, if you paste the full 
on_enter
and the small test snippet again, I’ll return a ready-to-drop patch. Which would you prefer?
n
AttributeError: 'AgentSession' object has no attribute 'chat_ctx'. Did you mean: '_chat_ctx'?
t
Thanks — great detail. You hit the AttributeError because 
AgentSession
doesn’t expose 
chat_ctx
directly. Use the active agent’s chat context instead and make a couple small, safe changes so the agent question is reliably visible in tests and you avoid the speech scheduling races. Quick fixes to apply (minimal, safe) 1. Access the chat context from the active agent:
Copy code
chat_ctx = self.session.agent.chat_ctx
1. Emit a synchronous message before scheduling speech in on_enter (makes the question immediately discoverable by tests):
Copy code
# inside FactsQuestionnaire.on_enter
self.session.agent.chat_ctx.add_message(role="assistant", content=self.question_text)
handle = self.session.say(text=self.question_text, allow_interruptions=False)
await handle.wait_for_playout()
1. Continue to await the exact speech handle before completing the task:
Copy code
if ack_message:
    handle = self.session.say(ack_message, allow_interruptions=False)
    await handle.wait_for_playout()

self.complete(result)
1. Defensive guard to avoid scheduling during shutdown:
Copy code
if not self.session.is_closing:
    handle = self.session.say(...)
    await handle.wait_for_playout()
Test-side adjustments (one-line, correct usage) • Do not 
await
the assertion object. Await 
.judge()
only.
Copy code
# correct: do not await contains_message(); await judge()
await turn1.expect.contains_message(role="assistant").judge(judge_llm, intent="...")
• If the assistant message can be emitted later in the event stream (e.g., function calls appear first), prefer:
Copy code
turn1.expect.contains_message(role="assistant")  # no await
# or use the awaited judge form shown above
Why this will fix your failures • 
self.session.agent.chat_ctx.add_message(...)
places the assistant question synchronously into the conversation history, so 
contains_message
will find it even if TTS or other events are delayed. • Awaiting the specific speech handle prevents task completion from racing with speech playout and avoids the "draining/pausing" scheduler error and the asyncio warning. • The defensive guard prevents attempts to schedule speech while the session is closing. If you want, paste: • The FactsQuestionnaire 
on_enter
and where 
self.complete(...)
is called (I can mark the exact lines to add). • The small test snippet (turn1/turn2) you’re running now. I’ll return a one-line patch for the task and the precise test line to change. Sources: External data and RAG | LiveKit Documentation | Collect recording consent with tasks | LiveKit Documentation | Testing and evaluation | LiveKit Documentation | How to build an agent with speech-to-text input and text-only output | Live...
n
Copy code
session.say by defaults adds to chat context as per your documentation

await session.say(
   "Hello. How can I help you today?",
   allow_interruptions=False,
)

await session.say(
  'Hello. How can I help you today?',
  {
    allowInterruptions: false,
  }
);

Parameters 
You can call session.say() with the following options:

text only: Synthesizes speech using TTS, which is added to the transcript and chat context (unless add_to_chat_ctx=False).
t
Thanks — this is useful confirmation. You're right: 
session.say()
does add text to the chat context by default. Given that, the remaining causes for the agent question not appearing in your test are timing and ordering (speech/event emission vs. tool/function events and task lifecycle). Here’s a short checklist and the minimal, deterministic changes to make the agent question reliably visible in tests and stop the scheduler errors. 1) Verify you await the speech handle returned by session.say • Correct (keep doing this):
Copy code
handle = self.session.say(self.question_text, allow_interruptions=False)
await handle.wait_for_playout()
• Why: this ensures that the specific TTS task finishes before you move on or call self.complete(). 2) Ensure no other code schedules speech after you await the handle and before you complete the task • Search for other 
session.say()
/ 
generate_reply()
calls in the task, callbacks, on_exit, or parent-agent code. Any such call must be awaited before calling 
self.complete(...)
. • If any background task may schedule speech, make it finish or cancel before completing. 3) Guard against scheduling during shutdown/drain • Add a defensive guard so you don’t call say() while the scheduler is refusing new speech:
Copy code
if not self.session.is_closing:
    handle = self.session.say(..., allow_interruptions=False)
    await handle.wait_for_playout()
• Why: prevents "cannot schedule new speech, ... draining/pausing" when session is closing. 4) Make the question immediately discoverable for tests (recommended, minimal change) • Even though say() adds to chat context, event ordering can still cause the assistant message to appear too late. Emit a synchronous chat message immediately so tests can find it deterministically:
Copy code
# Use the agent's chat context (session.chat_ctx is not available)
self.session.agent.chat_ctx.add_message(role="assistant", content=self.question_text)

# Then schedule speech (still await it)
handle = self.session.say(self.question_text, allow_interruptions=False)
await handle.wait_for_playout()
• Why: adding the message synchronously guarantees it is present in RunResult.events / chat history, avoiding timing races with function/tool events. 5) Test assertion patterns — avoid TypeError and be tolerant to event ordering • Don’t await the assert object. Await judge() only:
Copy code
# Correct: await the .judge() call
await turn1.expect.contains_message(role="assistant").judge(judge_llm, intent="...")
• If the assistant message may not be the immediate next event (a function call can appear first), use contains_message (which searches the events) rather than next_event():
Copy code
turn1.expect.contains_message(role="assistant")   # no await
# or with judge:
await turn1.expect.contains_message(role="assistant").judge(judge_llm, intent="...")
6) Quick debug steps if problem persists • Print the events to inspect ordering:
Copy code
print([type(e).__name__ for e in turn1.events])
print([e.model_dump() for e in turn1.events])  # or equivalent
• Confirm where speech scheduling is attempted in the traceback if you still see the RuntimeError. If you want, I can: • Give the exact one-line edits for your task (where to add chat_ctx.add_message and the guard). • Show the single test-line replacement (contains_message + judge) to make the test stable. Which would you prefer — the exact patch to drop into your task and test, or I should inspect the small task/test snippets you have and produce the precise changes? Sources: Agent speech and audio | LiveKit Documentation | Job lifecycle | LiveKit Documentation | Testing and evaluation | LiveKit Documentation
n
we need capture_run=True for session.run similar to session.start
t
Thanks — that's a great idea. Using capture_run=True with session.run is exactly the right way to make the single-turn RunResult include all events (messages, function calls, speech items, and state changes) so your assertions can reliably see the agent question, tool outputs, and any handoffs. How to use it (minimal example)
Copy code
# capture all events for this single turn
turn1 = await session.run(user_input="Agent speech", capture_run=True)

# now you can search/assert across all events for the assistant question
await turn1.expect.contains_message(role="assistant").judge(
    judge_llm,
    intent="Thanks the patient and explains that some questions will be asked to understand their situation",
)
Why this helps • 
capture_run=True
collects the full event stream produced during that run — including messages added by 
session.say()
(chat context entries), function/tool events, and any AgentHandoff metadata — so you don't miss events that are emitted slightly later or not as the immediate next_event. • It makes assertions tolerant to ordering (function call may come before the assistant message) because 
contains_message()
searches the captured events rather than expecting the very next event. Two small test / task best practices to use with capture_run • Continue awaiting the exact speech handles in your task:
Copy code
handle = self.session.say(self.question_text, allow_interruptions=False)
  await handle.wait_for_playout()
This prevents scheduling races and the "draining/pausing" RuntimeError. • Use 
contains_message()
(not awaited) and await `.judge()`:
Copy code
# correct: do not await contains_message() itself
  await turn1.expect.contains_message(role="assistant").judge(judge_llm, intent="...")
When to prefer capture_run=True • Use it when a turn triggers async events that may not appear instantly (speech playout, tool outputs, task handoffs). • It’s especially useful for multi-agent/task flows where event ordering can vary. If you want, I can: • Show a one-line edit to your existing test to add 
capture_run=True
in each 
session.run(...)
call, or • Produce a small example test showing capture_run across your turn1/turn2/turn3 flow. Which would you prefer? Sources: Using the Testing Framework with Realtime Models | LiveKit
n
but capture run is only available for start and not run method
t
I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!
n
[6] AgentHandoffEvent(old_agent=<tasks.chief_complaint.verify.VerifyMainComplaint object at 0x12b60c050>, new_agent=<agents.greet_and_route.GreetAndRouteAgent object at 0x12b603110>) [7] AgentHandoffEvent(old_agent=<agents.greet_and_route.GreetAndRouteAgent object at 0x12b603110>, new_agent=<agents.triage_agent.TriageAgent object at 0x12e8d0860>) [8] AgentHandoffEvent(old_agent=<agents.triage_agent.TriageAgent object at 0x12e8d0860>, new_agent=<tasks.facts_questionaire.FactsQuestionnaire object at 0x12e8ec3e0>) Why does it not include the agent chat message on eneter?
are you still here?
2 Views
Open in Slack
PreviousNext
Powered by