From 4d73ef07dd8b3b65913134ff6e1bd28c3a7bfa36 Mon Sep 17 00:00:00 2001 From: Azain Khan Date: Thu, 20 Aug 2026 15:39:28 +0530 Subject: [PATCH 1/2] fix(simulate): harden hosted voice-sim reliability Native LiveKit->LiveKit, VAPI/Retell bridge, engine robustness, recorder, and observability fixes for the hosted runner (see PR body for the full breakdown). - native path: forward target turns via generate_reply(user_input=) so they reach the simulator LLM (were transcript-only -> simulator was deaf); defer target dispatch until the session is live; room_disconnected hangup detection + stop-reason fixes; typed dispatch failures - bridge: detect VAPI provider hangup/end frames; decouple the reader + add a provider_audio_timeout watchdog in LiveKitAudioBridge (covers VAPI + Retell) - engine: no_conversation guard, cleanup-timeout clamp, shared off-loop silero VAD, and an env-tunable per-child case-concurrency ceiling (ALK_VOICE_MAX_CASE_CONCURRENCY, caps the config-driven max_parallel_cases) - child_entrypoint: force-exit on cancel (no leaked children); logging config + job-boundary observability (start/completed/failed/cancelled/crashed) - results sink: submission-seam logging (http error / exception / ok / missing) - recording: hidden + subscribe-only + cancel-safe room recorder --- src/fi/simulate/hosted/child_entrypoint.py | 55 ++ src/fi/simulate/recording/room_recorder.py | 50 +- src/fi/simulate/results/futureagi.py | 37 ++ src/fi/simulate/simulation/bridge/livekit.py | 80 ++- src/fi/simulate/simulation/bridge/vapi.py | 24 + src/fi/simulate/simulation/engines/livekit.py | 295 ++++++++-- tests/runtime/test_livekit_engine.py | 532 +++++++++++++++++- tests/test_hosted_runner.py | 43 ++ tests/test_vapi_websocket_bridge.py | 59 ++ 9 files changed, 1091 insertions(+), 84 deletions(-) diff --git a/src/fi/simulate/hosted/child_entrypoint.py b/src/fi/simulate/hosted/child_entrypoint.py index a4a4e1a0..a945ed5f 100644 --- a/src/fi/simulate/hosted/child_entrypoint.py +++ b/src/fi/simulate/hosted/child_entrypoint.py @@ -18,6 +18,7 @@ import argparse import asyncio import json +import logging import os import signal import sys @@ -37,6 +38,22 @@ from fi.simulate.runtime.spec import SimulationSpec _HEARTBEAT_INTERVAL_SECONDS = 10.0 +_CANCEL_GRACE_SECONDS = 30.0 + +logger = logging.getLogger("fi.simulate.hosted.runner") + + +def _job_log_fields(job: StartRunnerJob) -> dict[str, Any]: + fields: dict[str, Any] = {"job_id": job.job_id, "mode": job.mode.value} + if job.voice is not None: + target = dict(job.voice.agent_definition or {}).get("target") or {} + fields["provider"] = target.get("provider") + dataset = dict(job.voice.scenario or {}).get("dataset") or [] + fields["cases"] = len(dataset) + if job.sink is not None: + fields["run_test_id"] = job.sink.run_test_id + fields["test_execution_id"] = job.sink.test_execution_id + return fields class _StatusReporter: @@ -157,6 +174,7 @@ async def _heartbeat(reporter: _StatusReporter) -> None: async def _execute(job: StartRunnerJob, reporter: _StatusReporter) -> int: reporter.emit(RunnerJobPhase.PREPARING) + logger.info("hosted job start", extra=_job_log_fields(job)) sink = _build_sink(job) if job.mode is RunnerMode.CHAT: @@ -174,6 +192,17 @@ async def _execute(job: StartRunnerJob, reporter: _StatusReporter) -> int: report: SimulationReport = await run_task except asyncio.CancelledError: reporter.emit(RunnerJobPhase.CANCELED, detail="cancelled") + logger.warning("hosted job cancelled", extra={"job_id": job.job_id}) + # Cancelling this coroutine does not cancel ``run_task``; without an + # explicit cancel ``asyncio.run`` shutdown waits on it forever and the + # child leaks past SIGTERM. + run_task.cancel() + try: + await asyncio.wait({run_task}, timeout=_CANCEL_GRACE_SECONDS) + except asyncio.CancelledError: + pass + if not run_task.done(): + os._exit(2) raise finally: heartbeat_task.cancel() @@ -193,6 +222,17 @@ async def _execute(job: StartRunnerJob, reporter: _StatusReporter) -> int: detail = report.failure.code if report.failure else "run_failed" else: detail = f"submission_{submission_status or 'missing'}" + outcome_fields = { + **_job_log_fields(job), + "run_status": getattr(report.status, "value", str(report.status)), + "submission_status": submission_status, + "report_hash": report.report_hash, + "detail": detail, + } + if completed: + logger.info("hosted job completed", extra=outcome_fields) + else: + logger.error("hosted job failed", extra=outcome_fields) reporter.emit( RunnerJobPhase.COMPLETED if completed else RunnerJobPhase.FAILED, detail=detail, @@ -228,7 +268,21 @@ async def _main_async(job: StartRunnerJob, reporter: _StatusReporter) -> int: return 2 +def _configure_logging() -> None: + """The child runs with no logging config, so INFO seams (job start/outcome, + engine dispatch/join/stop_reason) were silently dropped by the WARNING-level + lastResort handler and never reached the runner's log capture.""" + root = logging.getLogger() + if not root.handlers: + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s")) + root.addHandler(handler) + root.setLevel(logging.WARNING) + logging.getLogger("fi.simulate").setLevel(logging.INFO) + + def main(argv: list[str] | None = None) -> int: + _configure_logging() parser = argparse.ArgumentParser(prog="fi.simulate.hosted.child_entrypoint") parser.add_argument("job", help="path to the StartRunnerJob JSON file") parser.add_argument("--status-file", default=None) @@ -241,6 +295,7 @@ def main(argv: list[str] | None = None) -> int: try: return asyncio.run(_main_async(job, reporter)) except Exception as exc: # noqa: BLE001 + logger.exception("hosted job crashed", extra={"job_id": job.job_id}) reporter.emit( RunnerJobPhase.FAILED, detail=f"{type(exc).__name__}: {exc}" ) diff --git a/src/fi/simulate/recording/room_recorder.py b/src/fi/simulate/recording/room_recorder.py index 95ac49d3..ff38f029 100644 --- a/src/fi/simulate/recording/room_recorder.py +++ b/src/fi/simulate/recording/room_recorder.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import contextlib import logging import re import time @@ -83,27 +84,56 @@ async def start(self) -> None: raise ImportError("LiveKit recording requires the 'livekit' extra") self._running = True await asyncio.sleep(max(0.0, self._join_delay_s)) - token = ( - AccessToken(self._api_key, self._api_secret) - .with_identity(self._identity) - .with_grants(VideoGrants(room_join=True, room=self._room_name)) - .to_jwt() - ) + token = self._build_token() room = rtc.Room() - await room.connect(self._url, token) + try: + await room.connect( + self._url, + token, + options=rtc.RoomOptions(auto_subscribe=False), + ) + except BaseException: + with contextlib.suppress(Exception): + await room.disconnect() + raise self._room = room self._recording_started_at = time.time() self._output_dir.mkdir(parents=True, exist_ok=True) + @room.on("track_published") + def _on_track_published(publication, participant) -> None: + self._subscribe_audio(publication) + @room.on("track_subscribed") def _on_track_subscribed(track, publication, participant) -> None: self._start_recording(track, publication, participant) for participant in tuple(room.remote_participants.values()): for publication in tuple(participant.track_publications.values()): - track = getattr(publication, "track", None) - if track is not None: - self._start_recording(track, publication, participant) + self._subscribe_audio(publication) + + def _build_token(self) -> str: + return ( + AccessToken(self._api_key, self._api_secret) + .with_identity(self._identity) + .with_grants( + VideoGrants( + room_join=True, + room=self._room_name, + hidden=True, + recorder=True, + can_publish=False, + can_publish_data=False, + can_update_own_metadata=False, + ) + ) + .to_jwt() + ) + + def _subscribe_audio(self, publication: Any) -> None: + if getattr(publication, "kind", None) != rtc.TrackKind.KIND_AUDIO: + return + publication.set_subscribed(True) def paths_for_participant( self, diff --git a/src/fi/simulate/results/futureagi.py b/src/fi/simulate/results/futureagi.py index bd2cb7f4..7c049858 100644 --- a/src/fi/simulate/results/futureagi.py +++ b/src/fi/simulate/results/futureagi.py @@ -25,6 +25,7 @@ from __future__ import annotations import json +import logging import os from datetime import datetime, timezone from pathlib import Path @@ -41,6 +42,8 @@ from .filesystem import LocalFilesystemResultSink +logger = logging.getLogger("fi.simulate.results.futureagi") + _STATUS_MAP = { "completed": "completed", "failed": "failed", @@ -223,6 +226,14 @@ def submit_case(self, index: int, case: Any) -> None: "status_code": resp.status_code, "body": _safe_body(resp), } + logger.warning( + "case submission http error", + extra={ + "case_index": index, + "call_execution_id": call_id, + "status_code": resp.status_code, + }, + ) return self._streamed_indices.add(index) self._stream_failures.pop(index, None) @@ -232,6 +243,14 @@ def submit_case(self, index: int, case: Any) -> None: "call_execution_id": call_id, "error": f"{type(exc).__name__}: {exc}", } + logger.warning( + "case submission failed", + extra={ + "case_index": index, + "call_execution_id": call_id, + "error": f"{type(exc).__name__}: {exc}", + }, + ) def case_started(self, index: int) -> None: """PATCH a pre-allocated CallExecution row to ONGOING the moment its case @@ -345,6 +364,9 @@ def submit(self, report: SimulationReport) -> dict[str, Any]: if missing: submission["status"] = "not_configured" submission["reason"] = "missing_config: " + ",".join(missing) + logger.warning( + "submission not configured", extra={"missing": ",".join(missing)} + ) _write_submission(run_directory, submission) return submission @@ -360,9 +382,24 @@ def submit(self, report: SimulationReport) -> dict[str, Any]: ) submission.update(outcome) submission["status"] = "submitted" + logger.info( + "submission ok", + extra={ + "run_test_id": self._run_test_id, + "test_execution_id": self._test_execution_id, + }, + ) except Exception as exc: submission["status"] = "failed" submission["reason"] = f"submission_error: {exc.__class__.__name__}: {exc}" + logger.error( + "submission failed", + extra={ + "run_test_id": self._run_test_id, + "test_execution_id": self._test_execution_id, + "error": f"{exc.__class__.__name__}: {exc}", + }, + ) _write_submission(run_directory, submission) return submission diff --git a/src/fi/simulate/simulation/bridge/livekit.py b/src/fi/simulate/simulation/bridge/livekit.py index 41301b82..7caad192 100644 --- a/src/fi/simulate/simulation/bridge/livekit.py +++ b/src/fi/simulate/simulation/bridge/livekit.py @@ -16,6 +16,8 @@ TRACK_TIMEOUT_SECONDS = 30.0 WATCHDOG_TIMEOUT_SECONDS = 60.0 PROVIDER_READY_BUFFER_FRAMES = 3000 +PROVIDER_AUDIO_TIMEOUT_SECONDS = 120.0 +PROVIDER_QUEUE_MAX_CHUNKS = 200 class LiveKitAudioBridge: @@ -42,6 +44,7 @@ def __init__( self._closed = False self._close_lock = asyncio.Lock() self._last_audio_at = time.monotonic() + self._last_provider_audio_at = time.monotonic() @property def call_id(self) -> str | None: @@ -172,27 +175,56 @@ async def _room_to_provider(self) -> None: async def _provider_to_room(self) -> None: if self._audio_source is None: raise RuntimeError("bridge_not_connected") + # The websocket reader must never block on room playback: a stalled + # ``capture_frame`` would stop close/hangup frames from being seen and + # keep a dead provider call alive. Live audio, so drop oldest when full. + queue: asyncio.Queue[tuple[bytes, int] | None] = asyncio.Queue( + maxsize=PROVIDER_QUEUE_MAX_CHUNKS + ) + + async def _pump() -> None: + try: + async for chunk in self._connector.recv_audio(): + now = time.monotonic() + self._last_audio_at = now + self._last_provider_audio_at = now + if queue.full(): + queue.get_nowait() + queue.put_nowait(chunk) + finally: + if queue.full(): + queue.get_nowait() + queue.put_nowait(None) + + pump = asyncio.create_task(_pump()) resamplers: dict[int, PCMResampler] = {} - async for pcm, sample_rate in self._connector.recv_audio(): - self._last_audio_at = time.monotonic() - if sample_rate != ROOM_SAMPLE_RATE: - resampler = resamplers.setdefault( - sample_rate, - PCMResampler( - from_rate=sample_rate, - to_rate=ROOM_SAMPLE_RATE, - channels=ROOM_CHANNELS, - ), - ) - pcm = resampler.convert(pcm) - await self._audio_source.capture_frame( - rtc.AudioFrame( - data=pcm, - sample_rate=ROOM_SAMPLE_RATE, - num_channels=ROOM_CHANNELS, - samples_per_channel=len(pcm) // 2, + try: + while True: + chunk = await queue.get() + if chunk is None: + break + pcm, sample_rate = chunk + if sample_rate != ROOM_SAMPLE_RATE: + resampler = resamplers.setdefault( + sample_rate, + PCMResampler( + from_rate=sample_rate, + to_rate=ROOM_SAMPLE_RATE, + channels=ROOM_CHANNELS, + ), + ) + pcm = resampler.convert(pcm) + await self._audio_source.capture_frame( + rtc.AudioFrame( + data=pcm, + sample_rate=ROOM_SAMPLE_RATE, + num_channels=ROOM_CHANNELS, + samples_per_channel=len(pcm) // 2, + ) ) - ) + finally: + pump.cancel() + await asyncio.gather(pump, return_exceptions=True) async def _send_silence_until_track(self) -> None: frame = b"\x00" * int(16000 * 0.02 * 2) @@ -201,10 +233,18 @@ async def _send_silence_until_track(self) -> None: await asyncio.sleep(0.02) async def _watchdog(self) -> None: + # ``_last_audio_at`` refreshes on simulator silence frames too, so it is + # blind to a dead provider; track provider-received audio separately. while True: await asyncio.sleep(5.0) - if time.monotonic() - self._last_audio_at > WATCHDOG_TIMEOUT_SECONDS: + now = time.monotonic() + if now - self._last_audio_at > WATCHDOG_TIMEOUT_SECONDS: raise RuntimeError("bridge_audio_watchdog_timeout") + if ( + now - self._last_provider_audio_at + > PROVIDER_AUDIO_TIMEOUT_SECONDS + ): + raise RuntimeError("provider_audio_timeout") async def _wait_for_room_disconnect(self) -> None: if self._room_disconnected is None: diff --git a/src/fi/simulate/simulation/bridge/vapi.py b/src/fi/simulate/simulation/bridge/vapi.py index fa2644bd..fcb2f9af 100644 --- a/src/fi/simulate/simulation/bridge/vapi.py +++ b/src/fi/simulate/simulation/bridge/vapi.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import logging import os from collections.abc import AsyncIterator @@ -129,14 +130,37 @@ async def recv_audio(self) -> AsyncIterator[tuple[bytes, int]]: async for message in self._ws: if message.type == aiohttp.WSMsgType.BINARY: yield message.data, VAPI_SAMPLE_RATE + elif message.type == aiohttp.WSMsgType.TEXT: + if self._is_call_end_event(message.data): + logger.info( + "vapi_websocket_call_ended", + extra={"call_id": self._call_id}, + ) + break elif message.type in { aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.ERROR, }: break self._connected = False + @staticmethod + def _is_call_end_event(payload: str) -> bool: + try: + data = json.loads(payload) + except ValueError: + return False + if not isinstance(data, dict): + return False + event_type = str(data.get("type") or "").lower() + if event_type in {"hangup", "call-ended", "end-of-call-report"}: + return True + return event_type == "status-update" and str( + data.get("status") or "" + ).lower() in {"ended", "ending"} + async def disconnect(self) -> None: self._connected = False if self._ws and not self._ws.closed: diff --git a/src/fi/simulate/simulation/engines/livekit.py b/src/fi/simulate/simulation/engines/livekit.py index 46c9c07f..1b33e25d 100644 --- a/src/fi/simulate/simulation/engines/livekit.py +++ b/src/fi/simulate/simulation/engines/livekit.py @@ -5,6 +5,7 @@ import logging import os import re +import threading from dataclasses import dataclass, field from urllib.parse import urlsplit from pathlib import Path @@ -88,6 +89,39 @@ # finishes playing), then delete the room so neither side keeps talking into a # call the other has already left. _FINAL_TURN_COMMIT_WAIT_SECONDS = 30.0 +# The hosted platform inflates ``cleanup_timeout`` to carry the whole run +# budget (observed 1470s); as a per-step cleanup bound it must stay capped. +_MAX_CLEANUP_TIMEOUT_SECONDS = 60.0 +_NO_CONVERSATION_TIMEOUT_SECONDS = 120.0 +# Each web case drives a full voice pipeline (STT/LLM/TTS + LiveKit conns) in one +# child; too many starve the pod's CPU. This is an OPS CEILING on the +# config-driven ``max_parallel_cases`` (not a replacement for it) — tune +# ``ALK_VOICE_MAX_CASE_CONCURRENCY`` to the pod's cores. Caps web cases only. +_VOICE_MAX_CASE_CONCURRENCY_DEFAULT = 4 + + +def _voice_max_case_concurrency() -> int: + raw = os.environ.get("ALK_VOICE_MAX_CASE_CONCURRENCY", "").strip() + if not raw: + return _VOICE_MAX_CASE_CONCURRENCY_DEFAULT + try: + value = int(raw) + except ValueError: + return _VOICE_MAX_CASE_CONCURRENCY_DEFAULT + return value if value >= 1 else _VOICE_MAX_CASE_CONCURRENCY_DEFAULT + +_silero_vad: Any | None = None +_silero_vad_guard = threading.Lock() + + +def _load_silero_vad_sync() -> Any: + """One shared VAD per process; per-case ``VAD.load()`` ran a synchronous + model load on the event loop for every concurrent case.""" + global _silero_vad + with _silero_vad_guard: + if _silero_vad is None: + _silero_vad = silero.VAD.load() + return _silero_vad @dataclass(frozen=True) @@ -394,6 +428,7 @@ async def run( "sip_inbound_room_template_required: multi-case inbound runs " "need {run_id} or {test_case_id} in room_name" ) + cleanup_timeout = min(cleanup_timeout, _MAX_CLEANUP_TIMEOUT_SECONDS) current_run_id = run_id or new_run_id() if recording_case_directory is not None and len(scenario.dataset) != 1: raise ValueError( @@ -409,7 +444,14 @@ async def run( case_concurrency = ( 1 if profile.is_sip - else max(1, min(int(max_concurrency or 1), len(scenario.dataset))) + else max( + 1, + min( + int(max_concurrency or 1), + _voice_max_case_concurrency(), + len(scenario.dataset), + ), + ) ) case_semaphore = asyncio.Semaphore(case_concurrency) @@ -740,24 +782,13 @@ def buffer_target_transcription( ), ) if outcome is None and profile.uses_external_room: - # agent_first: defer the target dispatch until AFTER the early - # buffer handler is registered (post room.connect), so the - # target's greeting transcription is never dropped for lack of - # a handler. simulator_first dispatches now, unchanged. - if conversation_direction == "agent_first": - target_dispatch_deferred = True - else: - await asyncio.wait_for( - api_client.agent_dispatch.create_dispatch( - api.CreateAgentDispatchRequest( - agent_name=agent_definition.agent_name - or agent_definition.name, - room=room_name, - metadata=_dispatch_metadata_json(agent_definition), - ) - ), - timeout=connect_timeout, - ) + # Defer the target dispatch until AFTER the early buffer + # handler is registered and the session is live (both + # directions): a native target may greet the moment it + # joins even when the simulator is meant to open, and the + # LiveKit client drops a text-stream header that arrives + # with no handler — the greeting would be lost. + target_dispatch_deferred = True elif outcome is None and profile.receives_inbound_call: try: ( @@ -811,7 +842,7 @@ def buffer_target_transcription( timeout=connect_timeout, ) room_connected = True - if conversation_direction == "agent_first" and profile.uses_external_room: + if profile.uses_external_room: # Buffer any target greeting that arrives before readiness; the # LiveKit client drops a text-stream header with no handler. room.register_text_stream_handler( @@ -882,16 +913,55 @@ def buffer_target_transcription( # Session + early buffer handler are live; now dispatch the target # so its greeting stream is captured, not dropped. assert api_client is not None - await asyncio.wait_for( - api_client.agent_dispatch.create_dispatch( - api.CreateAgentDispatchRequest( - agent_name=agent_definition.agent_name - or agent_definition.name, - room=room_name, - metadata=_dispatch_metadata_json(agent_definition), - ) - ), - timeout=connect_timeout, + dispatch_agent_name = ( + agent_definition.agent_name or agent_definition.name + ) + try: + await asyncio.wait_for( + api_client.agent_dispatch.create_dispatch( + api.CreateAgentDispatchRequest( + agent_name=dispatch_agent_name, + room=room_name, + metadata=_dispatch_metadata_json(agent_definition), + ) + ), + timeout=connect_timeout, + ) + except asyncio.TimeoutError: + outcome = _failure_outcome( + TestCaseStatus.TIMED_OUT, + FailureStage.PREPARING, + "livekit_dispatch_timeout", + "Target agent dispatch exceeded its deadline", + retryable=True, + ) + return outcome + except Exception as exc: + logger.warning( + "LiveKit target dispatch failed", + exc_info=redacted_exc_info(exc), + extra={ + "run_id": run_id, + "test_case_id": test_case_id, + "room_name": room_name, + }, + ) + outcome = _failure_outcome( + TestCaseStatus.FAILED, + FailureStage.PREPARING, + "livekit_dispatch_failed", + "Failed to dispatch the target agent", + details=_safe_provider_error_details( + exc, operation="agent_dispatch" + ), + ) + return outcome + logger.info( + "livekit_target_dispatched agent=%s room=%s run=%s case=%s", + dispatch_agent_name, + room_name, + run_id, + test_case_id, ) if profile.uses_web_audio_bridge: try: @@ -1044,6 +1114,14 @@ def buffer_target_transcription( target_identity=effective_target_identity, timeout=effective_readiness_timeout, ) + logger.info( + "livekit_target_joined identity=%s sid=%s track=%s run=%s case=%s", + target.identity, + target.sid, + target.audio_track_sid, + run_id, + test_case_id, + ) # RoomIO auto-links to the first participant that joined — the # recorder, which publishes no audio — so the simulator's STT never # hears the target. Re-point it at the target readiness selected. @@ -1117,6 +1195,12 @@ def on_target_transcription( agent_first_silence_timeout_seconds=agent_first_silence_timeout_seconds, provider_task=bridge_task, ) + logger.info( + "livekit_conversation_ended stop_reason=%s run=%s case=%s", + stop_reason, + run_id, + test_case_id, + ) # End the call cleanly. First let the party that just spoke commit its # own final turn — a LiveKit turn only lands in history once its TTS # finishes — bounded so we do not wait on the other side. We do NOT @@ -1448,6 +1532,14 @@ def on_target_transcription( ), } ) + logger.info( + "livekit_case_outcome status=%s stop_reason=%s failure=%s run=%s case=%s", + outcome.status.value, + outcome.metadata.get("stop_reason"), + outcome.failure.code if outcome.failure is not None else None, + run_id, + test_case_id, + ) return outcome async def _create_customer_agent( @@ -1517,7 +1609,7 @@ async def _create_customer_agent( stt_config=stt_config, tts_config=tts_config, ) - vad = silero.VAD.load() + vad = await asyncio.to_thread(_load_silero_vad_sync) agent = _TestRunnerAgent( persona=persona, min_turn_messages=min_turn_messages, @@ -1590,21 +1682,27 @@ async def _forward_target_transcription( # trailing target turn survives regardless of session state. if captured_target_turns is not None: captured_target_turns.append(transcript) - # Best-effort commit onto the chat context too (keeps the live - # conversation coherent while the session is still running). - try: - session.history.add_message(role="user", content=transcript) - except Exception: # noqa: BLE001 - pass # Only elicit a simulator response while the conversation is live; once # it has ended the target's turn is recorded but the simulator stays - # silent. + # silent. The turn MUST travel through ``generate_reply(user_input=...)``: + # the reply pipeline reads the agent's own chat context, not + # ``session.history``, so a turn only added to the history is invisible + # to the simulator LLM (it answers as if it heard nothing). The pipeline + # then persists the message into both contexts once the reply schedules. if conversation_ended is None or not conversation_ended.is_set(): try: - session.generate_reply() + session.generate_reply(user_input=transcript) except RuntimeError: # Session is already closing; the turn is captured above. pass + else: + return + # Conversation over (or the session rejected the reply): record the + # turn on the transcript without eliciting a response. + try: + session.history.add_message(role="user", content=transcript) + except Exception: # noqa: BLE001 + pass except Exception as exc: # noqa: BLE001 logger.warning( "Failed to consume target transcription stream", @@ -1673,6 +1771,7 @@ async def _wait_for_conversation_end( ) -> str: closed = asyncio.Event() target_disconnected = asyncio.Event() + room_disconnected = asyncio.Event() def on_close(_event) -> None: closed.set() @@ -1681,15 +1780,38 @@ def on_participant_disconnected(participant) -> None: if str(participant.identity) == target_identity: target_disconnected.set() + def on_room_disconnected(*_args) -> None: + room_disconnected.set() + session.on("close", on_close) room.on("participant_disconnected", on_participant_disconnected) + # A native target commonly hangs up by DELETING the room (the LiveKit + # hangup recipe); the simulator then sees a room disconnect, not a + # participant_disconnected, and without this watcher the case idled + # through the silence backstop before ending. + room.on("disconnected", on_room_disconnected) + # The target may have left in the gap between readiness and this + # registration — the event is gone, so recheck presence once. + remote_participants = getattr(room, "remote_participants", None) + if isinstance(remote_participants, dict) and not any( + str(participant.identity) == target_identity + for participant in remote_participants.values() + ): + target_disconnected.set() tasks = { "closed": asyncio.create_task(closed.wait()), "target_disconnected": asyncio.create_task(target_disconnected.wait()), + "room_disconnected": asyncio.create_task(room_disconnected.wait()), "simulator_end_call": asyncio.create_task(customer_agent.end_requested.wait()), "conversation_settled": asyncio.create_task( _wait_for_conversation_silence(session) ), + "no_conversation": asyncio.create_task( + _wait_for_conversation_never_started( + session, + timeout_seconds=_NO_CONVERSATION_TIMEOUT_SECONDS, + ) + ), } if conversation_direction == "agent_first": tasks["conversation_silence_timeout"] = asyncio.create_task( @@ -1717,17 +1839,40 @@ def on_participant_disconnected(participant) -> None: await asyncio.gather(*owned_pending, return_exceptions=True) if not done: return "timeout" + # A crashed monitor is also "done"; it must not count as its condition. + completed: set[str] = set() + monitor_failures: dict[str, BaseException] = {} + for name, task in tasks.items(): + if task not in done or task.cancelled(): + continue + exc = task.exception() + if exc is None: + completed.add(name) + else: + monitor_failures[name] = exc + for name, exc in monitor_failures.items(): + logger.warning( + "conversation end monitor failed", + exc_info=redacted_exc_info(exc), + extra={"monitor": name, "target_identity": target_identity}, + ) + # A bridge task error is still a real provider-side disconnect. + if "provider_disconnected" in monitor_failures: + completed.add("provider_disconnected") for reason in ( "simulator_end_call", "target_disconnected", + "room_disconnected", + "no_conversation", "conversation_silence_timeout", "conversation_settled", "provider_disconnected", "closed", ): - task = tasks.get(reason) - if task is not None and task in done: + if reason in completed: return "session_closed" if reason == "closed" else reason + if monitor_failures: + return "monitor_failed" return "session_closed" finally: _remove_room_listener( @@ -1735,6 +1880,7 @@ def on_participant_disconnected(participant) -> None: "participant_disconnected", on_participant_disconnected, ) + _remove_room_listener(room, "disconnected", on_room_disconnected) async def _wait_for_conversation_silence( @@ -1750,6 +1896,10 @@ async def _wait_for_conversation_silence( message count — a run is never cut off at a floor, it runs as long as turns keep flowing. The timer resets on every new message and while either side is speaking, so only a real ``quiet_seconds`` gap of nothing ends the call. + + Parks until the first non-empty turn: a call where nobody ever spoke is the + ``no_conversation`` monitor's condition, and this backstop firing first + mislabeled dead calls as merely settled. """ last_signature: tuple[tuple[str, str], ...] | None = None stable_since: float | None = None @@ -1757,6 +1907,11 @@ async def _wait_for_conversation_silence( while True: messages = _session_messages(session) signature = tuple((message["role"], message["content"]) for message in messages) + if not any(message["content"] for message in messages): + last_signature = signature + stable_since = None + await asyncio.sleep(0.1) + continue participant_speaking = ( getattr(session, "agent_state", None) == "speaking" or getattr(session, "user_state", None) == "speaking" @@ -1771,6 +1926,21 @@ async def _wait_for_conversation_silence( await asyncio.sleep(0.1) +async def _wait_for_conversation_never_started( + session: AgentSession, + *, + timeout_seconds: float, +) -> None: + """Completes only when no non-empty turn has ever been committed; parks + forever (until cancelled) once the conversation has actually started.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout_seconds + while loop.time() < deadline: + if any(message["content"] for message in _session_messages(session)): + await asyncio.Event().wait() + await asyncio.sleep(0.5) + + async def _wait_for_agent_first_silence( session: AgentSession, *, @@ -1781,7 +1951,13 @@ async def _wait_for_agent_first_silence( while True: messages = _session_messages(session) signature = tuple((message["role"], message["content"]) for message in messages) - if signature != last_signature: + participant_speaking = ( + getattr(session, "agent_state", None) == "speaking" + or getattr(session, "user_state", None) == "speaking" + ) + # A turn lands in history only after its TTS finishes, so an in-flight + # utterance longer than the timeout must count as activity. + if signature != last_signature or participant_speaking: last_signature = signature last_change = asyncio.get_running_loop().time() roles = {message["role"] for message in messages if message["content"]} @@ -1985,13 +2161,29 @@ def _conversation_outcome( messages=messages, retryable=True, ) - if stop_reason in {"conversation_silence_timeout", "session_closed"}: + if stop_reason in { + "conversation_silence_timeout", + "session_closed", + "no_conversation", + "monitor_failed", + }: code = stop_reason - message = ( - "Agent-first conversation stalled after it began" - if stop_reason == "conversation_silence_timeout" - else "Conversation session closed before a natural end condition" - ) + message = { + "conversation_silence_timeout": ( + "Agent-first conversation stalled after it began" + ), + "session_closed": ( + "Conversation session closed before a natural end condition" + ), + "no_conversation": ( + "No conversation turns were committed before the inactivity " + "deadline" + ), + "monitor_failed": ( + "Conversation end monitoring failed before a natural end " + "condition" + ), + }[stop_reason] return _failure_outcome( TestCaseStatus.FAILED, FailureStage.RUNNING, @@ -2003,8 +2195,8 @@ def _conversation_outcome( ) if len(messages) < min_turn_messages or not _has_role_alternation(messages): code = ( - "target_disconnected" - if stop_reason == "target_disconnected" + stop_reason + if stop_reason in {"target_disconnected", "room_disconnected"} else "insufficient_conversation" ) return _failure_outcome( @@ -2014,7 +2206,8 @@ def _conversation_outcome( "Conversation ended before the required alternating turns completed", transcript=transcript, messages=messages, - retryable=stop_reason in {"target_disconnected", "session_closed"}, + retryable=stop_reason + in {"target_disconnected", "room_disconnected", "session_closed"}, details={ "stop_reason": stop_reason, "turn_count": str(len(messages)), diff --git a/tests/runtime/test_livekit_engine.py b/tests/runtime/test_livekit_engine.py index 6ff298a8..8ddfcf11 100644 --- a/tests/runtime/test_livekit_engine.py +++ b/tests/runtime/test_livekit_engine.py @@ -17,7 +17,10 @@ from fi.simulate.runtime import TestCaseStatus as CaseStatus from fi.simulate.simulation import bridge as _bridge from fi.simulate.simulation.engines import livekit -from fi.simulate.simulation.engines.livekit import LiveKitEngine +from fi.simulate.simulation.engines.livekit import ( + LiveKitEngine, + _voice_max_case_concurrency, +) from fi.simulate.simulation import livekit_models from fi.simulate.simulation.models import Persona, Scenario @@ -466,6 +469,34 @@ def test_stereo_mix_leaves_missing_side_silent(tmp_path: Path) -> None: assert samples.tolist() == [0, 2000, 0, 2000] +def test_room_recorder_token_is_hidden_subscribe_only() -> None: + import jwt + + from fi.simulate.recording.room_recorder import RoomRecorder + + recorder = RoomRecorder( + url="wss://livekit.example.com", + api_key="key", + api_secret="secret-0123456789abcdef0123456789abcdef", + room_name="room-1", + identity="fagi-recorder-abc123", + ) + claims = jwt.decode( + recorder._build_token(), + "secret-0123456789abcdef0123456789abcdef", + algorithms=["HS256"], + ) + video = claims["video"] + assert video["room"] == "room-1" + assert video["roomJoin"] is True + assert video["hidden"] is True + assert video["recorder"] is True + assert video["canPublish"] is False + assert video["canPublishData"] is False + assert video["canUpdateOwnMetadata"] is False + assert video.get("canSubscribe", True) is True + + def test_stereo_mix_returns_none_when_both_sides_empty(tmp_path: Path) -> None: destination = tmp_path / "stereo.wav" result = mix_recordings_stereo([], [], destination, sample_rate=8000) @@ -1852,8 +1883,9 @@ def test_cases_run_concurrently_and_preserve_dataset_order(monkeypatch) -> None: ) ) - # Actual overlap happened, capped at the ceiling. - assert state["max_in_flight"] == 5 + # Web cases overlap but are clamped to the voice ceiling even though the + # platform requested 5 — one 4-CPU child cannot drive more pipelines. + assert state["max_in_flight"] == _voice_max_case_concurrency() # Despite inverted completion order, results stay in dataset order. order = [r.persona.persona["name"] for r in report.results] assert order == [f"Caller {i}" for i in range(6)] @@ -2005,3 +2037,497 @@ def test_dispatch_metadata_forwarded_when_set(): SimpleNamespace(dispatch_metadata={"b": 2, "a": 1}) ) assert out == json.dumps({"a": 1, "b": 2}, sort_keys=True) + + +def test_crashed_monitor_is_not_mistaken_for_its_condition(monkeypatch) -> None: + # Regression: a watcher that raised counted as "done" and was mapped to + # conversation_silence_timeout, reporting a dead call as merely "stalled". + class FakeRoom: + def on(self, _event, _callback): + return None + + def off(self, _event, _callback): + return None + + class FakeSession: + history = SimpleNamespace( + items=[ + SimpleNamespace(type="message", role="assistant", text_content="Hi"), + SimpleNamespace(type="message", role="user", text_content="Hello"), + ] + ) + + def on(self, _event, _callback): + return None + + async def _crash(session, **kwargs): + raise RuntimeError("engine is closed") + + async def _park(session, **kwargs): + await asyncio.Event().wait() + + monkeypatch.setattr(livekit, "_wait_for_agent_first_silence", _crash) + monkeypatch.setattr(livekit, "_wait_for_conversation_silence", _park) + + async def run() -> str: + return await livekit._wait_for_conversation_end( + FakeRoom(), + FakeSession(), + customer_agent=SimpleNamespace(end_requested=asyncio.Event()), + target_identity="target-agent", + timeout=1, + conversation_direction="agent_first", + agent_first_silence_timeout_seconds=30, + ) + + reason = asyncio.run(run()) + + assert reason == "monitor_failed" + outcome = livekit._conversation_outcome(reason, [], min_turn_messages=4) + assert outcome.status == CaseStatus.FAILED + assert outcome.failure.code == "monitor_failed" + + +def test_dead_call_with_no_turns_ends_as_no_conversation(monkeypatch) -> None: + # Regression: a target that greets then hangs up unseen (VAPI + # silence-timed-out) left the case running for the full talk budget with an + # empty transcript. + class FakeRoom: + def on(self, _event, _callback): + return None + + def off(self, _event, _callback): + return None + + class FakeSession: + history = SimpleNamespace(items=[]) + + def on(self, _event, _callback): + return None + + monkeypatch.setattr(livekit, "_NO_CONVERSATION_TIMEOUT_SECONDS", 0.05) + + async def run() -> str: + return await livekit._wait_for_conversation_end( + FakeRoom(), + FakeSession(), + customer_agent=SimpleNamespace(end_requested=asyncio.Event()), + target_identity="target-agent", + timeout=5, + conversation_direction="agent_first", + agent_first_silence_timeout_seconds=30, + ) + + reason = asyncio.run(run()) + + assert reason == "no_conversation" + outcome = livekit._conversation_outcome(reason, [], min_turn_messages=4) + assert outcome.status == CaseStatus.FAILED + assert outcome.failure.code == "no_conversation" + + +def test_no_conversation_guard_parks_once_turns_exist() -> None: + session = SimpleNamespace( + history=SimpleNamespace( + items=[ + SimpleNamespace(type="message", role="user", text_content="Hi"), + ] + ) + ) + + async def run() -> None: + task = asyncio.create_task( + livekit._wait_for_conversation_never_started( + session, + timeout_seconds=0.05, + ) + ) + done, _pending = await asyncio.wait({task}, timeout=0.3) + assert not done + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + asyncio.run(run()) + + +class _FakeTranscriptionReader: + def __init__(self, text: str) -> None: + self._text = text + self.info = SimpleNamespace(attributes={}) + + async def read_all(self) -> str: + return self._text + + +class _FakeReplySession: + def __init__(self, *, reply_error: Exception | None = None) -> None: + self.reply_inputs: list[object] = [] + self.history_adds: list[tuple[str, str]] = [] + self._reply_error = reply_error + session = self + + class _History: + def add_message(self, *, role: str, content: str) -> None: + session.history_adds.append((role, content)) + + self.history = _History() + + def generate_reply(self, *, user_input=None): + if self._reply_error is not None: + raise self._reply_error + self.reply_inputs.append(user_input) + + +def test_target_transcription_feeds_simulator_llm_context() -> None: + # Regression: the target's turns were added to session.history (the + # transcript context) with a bare generate_reply(); the reply pipeline + # reads the AGENT's chat context, so the simulator LLM never saw a single + # target turn and answered every one with "I can't hear you". + session = _FakeReplySession() + captured: list[str] = [] + + asyncio.run( + livekit._forward_target_transcription( + _FakeTranscriptionReader(" How can I help you today? "), + session, + conversation_ended=asyncio.Event(), + captured_target_turns=captured, + ) + ) + + assert session.reply_inputs == ["How can I help you today?"] + # The pipeline persists the turn into both contexts itself; a manual + # history add here would double the transcript entry. + assert session.history_adds == [] + assert captured == ["How can I help you today?"] + + +def test_target_transcription_after_end_records_without_reply() -> None: + session = _FakeReplySession() + ended = asyncio.Event() + ended.set() + captured: list[str] = [] + + asyncio.run( + livekit._forward_target_transcription( + _FakeTranscriptionReader("Goodbye!"), + session, + conversation_ended=ended, + captured_target_turns=captured, + ) + ) + + assert session.reply_inputs == [] + assert session.history_adds == [("user", "Goodbye!")] + assert captured == ["Goodbye!"] + + +def test_target_transcription_falls_back_to_history_when_session_closing() -> None: + session = _FakeReplySession(reply_error=RuntimeError("scheduling is paused")) + captured: list[str] = [] + + asyncio.run( + livekit._forward_target_transcription( + _FakeTranscriptionReader("One last thing."), + session, + conversation_ended=asyncio.Event(), + captured_target_turns=captured, + ) + ) + + assert session.history_adds == [("user", "One last thing.")] + assert captured == ["One last thing."] + + +class _FakeEventRoom: + def __init__(self, participants: dict | None = None) -> None: + self.listeners: dict[str, list] = {} + if participants is not None: + self.remote_participants = participants + + def on(self, event, callback=None): + self.listeners.setdefault(event, []).append(callback) + return callback + + def off(self, event, callback): + self.listeners.get(event, []).remove(callback) + + def fire(self, event, *args) -> None: + for callback in list(self.listeners.get(event, [])): + callback(*args) + + +_BALANCED_HISTORY = SimpleNamespace( + items=[ + SimpleNamespace(type="message", role="assistant", text_content="One"), + SimpleNamespace(type="message", role="user", text_content="Two"), + SimpleNamespace(type="message", role="assistant", text_content="Three"), + SimpleNamespace(type="message", role="user", text_content="Four"), + SimpleNamespace(type="message", role="assistant", text_content="Five"), + SimpleNamespace(type="message", role="user", text_content="Six"), + ] +) + + +def test_room_disconnect_ends_conversation() -> None: + # A native target commonly hangs up by deleting the room; the simulator + # sees a room disconnect, not a participant_disconnected, and the case + # idled through the silence backstop before ending. + room = _FakeEventRoom( + participants={"t": SimpleNamespace(identity="target-agent")} + ) + + class FakeSession: + history = _BALANCED_HISTORY + + def on(self, _event, _callback): + return None + + async def run() -> str: + loop = asyncio.get_running_loop() + loop.call_later(0.05, room.fire, "disconnected", "ROOM_DELETED") + return await livekit._wait_for_conversation_end( + room, + FakeSession(), + customer_agent=SimpleNamespace(end_requested=asyncio.Event()), + target_identity="target-agent", + timeout=5, + conversation_direction="simulator_first", + agent_first_silence_timeout_seconds=30, + ) + + reason = asyncio.run(run()) + + assert reason == "room_disconnected" + assert room.listeners["disconnected"] == [] + completed = livekit._conversation_outcome( + reason, + livekit._session_messages(SimpleNamespace(history=_BALANCED_HISTORY)), + min_turn_messages=6, + ) + assert completed.status == CaseStatus.COMPLETED + failed = livekit._conversation_outcome(reason, [], min_turn_messages=4) + assert failed.status == CaseStatus.FAILED + assert failed.failure.code == "room_disconnected" + assert failed.failure.retryable is True + + +def test_target_absent_at_watch_start_counts_as_disconnected() -> None: + # The target can leave between readiness and listener registration; the + # event is gone by then, so presence is rechecked once. + room = _FakeEventRoom(participants={}) + + class FakeSession: + history = _BALANCED_HISTORY + + def on(self, _event, _callback): + return None + + async def run() -> str: + return await livekit._wait_for_conversation_end( + room, + FakeSession(), + customer_agent=SimpleNamespace(end_requested=asyncio.Event()), + target_identity="target-agent", + timeout=5, + conversation_direction="simulator_first", + agent_first_silence_timeout_seconds=30, + ) + + assert asyncio.run(run()) == "target_disconnected" + + +def test_silence_backstop_parks_until_first_turn() -> None: + # A call where nobody ever spoke belongs to the no_conversation monitor; + # the backstop firing first mislabeled dead calls as merely settled. + session = SimpleNamespace(history=SimpleNamespace(items=[])) + + async def run() -> None: + task = asyncio.create_task( + livekit._wait_for_conversation_silence(session, quiet_seconds=0.01) + ) + done, _pending = await asyncio.wait({task}, timeout=0.3) + assert not done + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + asyncio.run(run()) + + +def test_agent_first_silence_ignores_inflight_speech() -> None: + # A turn lands in history only after its TTS finishes, so a long in-flight + # utterance used to trip the agent-first stall watcher mid-speech. + session = SimpleNamespace( + agent_state="speaking", + user_state="listening", + history=SimpleNamespace( + items=[ + SimpleNamespace(type="message", role="assistant", text_content="Hi"), + SimpleNamespace(type="message", role="user", text_content="Hello"), + ] + ), + ) + + async def run() -> None: + task = asyncio.create_task( + livekit._wait_for_agent_first_silence(session, timeout_seconds=0.05) + ) + await asyncio.sleep(0.2) + assert not task.done() + session.agent_state = "listening" + await asyncio.wait_for(task, timeout=1) + + asyncio.run(run()) + + +def _order_probe_engine(monkeypatch, calls, *, dispatch_error=None): + audio_kind = livekit.rtc.TrackKind.KIND_AUDIO + + class FakeRoom: + def __init__(self): + self.remote_participants = { + "target": SimpleNamespace( + identity="target-agent", + sid="participant-target", + track_publications={ + "track-target": SimpleNamespace( + sid="track-target", kind=audio_kind + ) + }, + ) + } + self.listeners = {} + + async def connect(self, url, token): + calls.append("connect") + + async def disconnect(self): + calls.append("disconnect") + + def on(self, event, callback=None): + self.listeners.setdefault(event, []).append(callback) + return callback + + def off(self, event, callback): + self.listeners.get(event, []).remove(callback) + + def register_text_stream_handler(self, topic, handler=None): + calls.append("register_text_handler") + return handler + + def unregister_text_stream_handler(self, topic): + pass + + class _Rooms: + async def create_room(self, request): + calls.append("create_room") + + async def delete_room(self, request): + calls.append("delete_room") + + class _Dispatch: + async def create_dispatch(self, request): + if dispatch_error is not None: + raise dispatch_error + calls.append("dispatch") + + class _Api: + def __init__(self): + self.room = _Rooms() + self.agent_dispatch = _Dispatch() + + async def aclose(self): + pass + + class FakeSession: + history = SimpleNamespace( + items=[ + SimpleNamespace(type="message", role="user", text_content="hi"), + SimpleNamespace(type="message", role="assistant", text_content="ok"), + ] + ) + + def on(self, event, callback): + return None + + def shutdown(self, *, drain=True): + pass + + class FakeCustomerAgent: + def __init__(self): + self.end_requested = asyncio.Event() + self.end_requested.set() + + async def start_session(self, _room, **_kwargs): + calls.append("start_session") + return FakeSession() + + def open_conversation(self): + calls.append("open") + + monkeypatch.setenv("LIVEKIT_API_KEY", "key") + monkeypatch.setenv("LIVEKIT_API_SECRET", "secret") + monkeypatch.setattr(livekit.rtc, "Room", lambda: FakeRoom()) + monkeypatch.setattr(livekit.api, "LiveKitAPI", lambda *_a: _Api()) + monkeypatch.setattr(livekit, "AccessToken", _fake_access_token()) + engine = LiveKitEngine() + + async def _fake_create(_p, _s, **_kwargs): + return FakeCustomerAgent(), None + + monkeypatch.setattr(engine, "_create_customer_agent", _fake_create) + return engine + + +def test_managed_dispatch_waits_for_session_and_buffer_handler(monkeypatch) -> None: + # Dispatching before the session + buffer handler are live loses a native + # target's greeting for BOTH directions: the LiveKit client drops a + # text-stream header that arrives with no registered handler. + calls: list = [] + engine = _order_probe_engine(monkeypatch, calls) + + report = asyncio.run( + engine.run( + agent_definition=_agent( + room_mode="managed", + agent_name="registered-agent", + target_participant_identity="target-agent", + ), + scenario=_scenario(), + run_id="run_dispatch_order", + min_turn_messages=2, + ) + ) + + assert report.results[0].metadata["status"] == CaseStatus.COMPLETED.value + assert calls.index("connect") < calls.index("register_text_handler") + assert calls.index("register_text_handler") < calls.index("dispatch") + assert calls.index("start_session") < calls.index("dispatch") + + +def test_dispatch_failure_is_typed_preparing_failure(monkeypatch) -> None: + calls: list = [] + engine = _order_probe_engine( + monkeypatch, + calls, + dispatch_error=RuntimeError("twirp: agent not registered"), + ) + + report = asyncio.run( + engine.run( + agent_definition=_agent( + room_mode="managed", + agent_name="registered-agent", + target_participant_identity="target-agent", + ), + scenario=_scenario(), + run_id="run_dispatch_fail", + min_turn_messages=2, + ) + ) + + metadata = report.results[0].metadata + assert metadata["status"] == CaseStatus.FAILED.value + assert metadata["failure"]["code"] == "livekit_dispatch_failed" + assert metadata["failure"]["stage"] == "preparing" + assert "delete_room" in calls diff --git a/tests/test_hosted_runner.py b/tests/test_hosted_runner.py index ddc85b79..378ccbb7 100644 --- a/tests/test_hosted_runner.py +++ b/tests/test_hosted_runner.py @@ -502,3 +502,46 @@ def _import(ref: str): module_name, _, attr = ref.partition(":") return getattr(importlib.import_module(module_name), attr) + + +def test_child_cancel_propagates_to_run_task(tmp_path, monkeypatch): + # Regression: cancelling _execute left run_task alive, so asyncio.run's + # shutdown waited on the engine forever and the child leaked past SIGTERM. + from fi.simulate.hosted import child_entrypoint as ce + + spec = _chat_spec("callable", {"target": "mod:fn"}) + job = StartRunnerJob( + job_id="job-cancel", + mode=RunnerMode.CHAT, + spec=spec, + sink={"root_directory": str(tmp_path / "runs")}, + ) + reporter = ce._StatusReporter("job-cancel", tmp_path / "status.jsonl") + state: dict[str, bool] = {} + + class _ParkedRunner: + async def run(self, *args, **kwargs): + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + state["run_task_cancelled"] = True + raise + + monkeypatch.setattr(ce, "SimulationRunner", _ParkedRunner) + monkeypatch.setattr(ce, "resolve_chat_target", lambda _spec: object()) + + async def scenario() -> None: + task = asyncio.create_task(ce._execute(job, reporter)) + await asyncio.sleep(0.05) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(scenario()) + + assert state.get("run_task_cancelled") is True + statuses = [ + json.loads(line) + for line in (tmp_path / "status.jsonl").read_text().splitlines() + ] + assert statuses[-1]["phase"] == "canceled" diff --git a/tests/test_vapi_websocket_bridge.py b/tests/test_vapi_websocket_bridge.py index 67ed4e7b..8217f6fd 100644 --- a/tests/test_vapi_websocket_bridge.py +++ b/tests/test_vapi_websocket_bridge.py @@ -140,3 +140,62 @@ def test_vapi_websocket_connector_requires_credentials(monkeypatch) -> None: with pytest.raises(ValueError, match="VAPI_API_KEY, VAPI_ASSISTANT_ID"): VapiWebSocketConnector.from_env() + + +def test_call_end_text_event_is_recognized() -> None: + import json + + is_end = VapiWebSocketConnector._is_call_end_event + assert is_end(json.dumps({"type": "hangup"})) + assert is_end(json.dumps({"type": "call-ended"})) + assert is_end(json.dumps({"type": "end-of-call-report"})) + assert is_end(json.dumps({"type": "status-update", "status": "ended"})) + assert not is_end(json.dumps({"type": "status-update", "status": "in-progress"})) + assert not is_end(json.dumps({"type": "transcript", "text": "hi"})) + assert not is_end("not json") + assert not is_end(json.dumps(["hangup"])) + + +def test_recv_audio_stops_on_call_end_text_event() -> None: + import json + + connector = VapiWebSocketConnector( + ConnectorConfig(api_key="k", assistant_id="a", api_url="https://api/call") + ) + + class _EndedWS: + closed = False + + def __aiter__(self): + self._messages = iter( + [ + SimpleNamespace( + type=vapi.aiohttp.WSMsgType.BINARY, data=b"audio" + ), + SimpleNamespace( + type=vapi.aiohttp.WSMsgType.TEXT, + data=json.dumps({"type": "hangup"}), + ), + SimpleNamespace( + type=vapi.aiohttp.WSMsgType.BINARY, data=b"late" + ), + ] + ) + return self + + async def __anext__(self): + try: + return next(self._messages) + except StopIteration as exc: + raise StopAsyncIteration from exc + + connector._ws = _EndedWS() + connector._connected = True + + async def collect(): + return [chunk async for chunk in connector.recv_audio()] + + chunks = asyncio.run(collect()) + + assert chunks == [(b"audio", vapi.VAPI_SAMPLE_RATE)] + assert connector.is_connected is False From 83b22bac03dee0a598c90c04124111ef9eec331f Mon Sep 17 00:00:00 2001 From: Azain Khan Date: Thu, 20 Aug 2026 16:22:53 +0530 Subject: [PATCH 2/2] fix(simulate): restore native target-turn timing metrics Native target turns reach the report via generate_reply(user_input=text) -- a text input with no audio metrics -- so target ("assistant") turns had stopped_speaking_at == started_speaking_at (zero duration), leaving bot_wpm, avg_latency_ms, and talk_ratio unpopulated (only simulator/"user" turns, which keep AgentSession ChatMessage.metrics, computed). Capture receiver-side wall-clock start/stop for each target transcription emission (playback-synced via TranscriptSynchronizer) and patch it onto the matching history turns in _merge_captured_target_turns -- aggregating min-start/max-stop across partial/extended emissions, filling only turns that lack a real stop, never overriding genuine metrics. VAPI/Retell unaffected (captured_target_turns is only populated by the native transcription handler). Verified on a local native sim: every assistant turn now carries real duration (4.7-15.4s), zero zero-duration. --- src/fi/simulate/simulation/engines/livekit.py | 103 ++++++++++++++---- tests/runtime/test_livekit_engine.py | 75 ++++++++++++- 2 files changed, 153 insertions(+), 25 deletions(-) diff --git a/src/fi/simulate/simulation/engines/livekit.py b/src/fi/simulate/simulation/engines/livekit.py index 1b33e25d..be683077 100644 --- a/src/fi/simulate/simulation/engines/livekit.py +++ b/src/fi/simulate/simulation/engines/livekit.py @@ -6,6 +6,7 @@ import os import re import threading +import time from dataclasses import dataclass, field from urllib.parse import urlsplit from pathlib import Path @@ -671,7 +672,7 @@ async def _run_single_test_case( # target closing delivered after the simulator is done never reaches the # chat context. These are merged into the report so the trailing target # turn is never lost. - captured_target_turns: list[str] = [] + captured_target_turns: list[dict[str, Any]] = [] # agent_first (target greets first): the target can publish its greeting # transcription before the main handler is registered post-readiness, and # the LiveKit client DROPS a text-stream header that arrives with no @@ -1669,10 +1670,17 @@ async def _forward_target_transcription( session: "AgentSession", *, conversation_ended: "asyncio.Event | None" = None, - captured_target_turns: list[str] | None = None, + captured_target_turns: list[dict[str, Any]] | None = None, ) -> None: + # Receiver-side wall clock — same clock domain as the simulator's + # ChatMessage.metrics, and the target's transcript IO is playback-synced + # (TranscriptSynchronizer), so stream-open ~= speech start and read_all() + # completion ~= speech end. Timestamps embedded in the stream are the + # sender's (laptop) clock; skew there would corrupt the derived latencies. + started_at = time.time() try: transcript = (await reader.read_all()).strip() + stopped_at = time.time() if not transcript: return # Capture the target's turn independently of the simulator session FIRST. @@ -1681,7 +1689,13 @@ async def _forward_target_transcription( # reaches the chat context. This list is merged into the report so the # trailing target turn survives regardless of session state. if captured_target_turns is not None: - captured_target_turns.append(transcript) + captured_target_turns.append( + { + "content": transcript, + "started_speaking_at": started_at, + "stopped_speaking_at": stopped_at, + } + ) # Only elicit a simulator response while the conversation is live; once # it has ended the target's turn is recorded but the simulator stays # silent. The turn MUST travel through ``generate_reply(user_input=...)``: @@ -2084,22 +2098,68 @@ def _canonical_report_messages(session: AgentSession) -> list[dict[str, Any]]: def _merge_captured_target_turns( messages: list[dict[str, Any]], - captured_target_turns: list[str] | None, + captured_target_turns: list[dict[str, Any]] | None, ) -> list[dict[str, Any]]: - """Append target turns captured off the transcription stream that never + """Restore native target-turn timing, and append any target turn that never reached the session history. - The target's closing is often delivered after the simulator session has - started draining (it rejects new input with "speech scheduling is paused"), - so it is recorded in ``captured_target_turns`` but missing from the report. - Each captured turn is the target (``assistant`` in the report perspective). - Deduped against existing content — including partial/extended emissions of - the same turn — so nothing already recorded is doubled. Appended after the - existing turns (they are trailing utterances) with a synthetic timing just - past the last message so downstream ms-offset ordering keeps them last. + Native target turns are fed to the simulator via ``generate_reply( + user_input=text)`` — a text input with no audio metrics — so their report + entries carry a start but no ``stopped_speaking_at``: zero-duration turns + that leave bot WPM, latency, and talk-ratio unpopulated. Each captured turn + carries receiver-side wall-clock timing (see ``_forward_target_transcription``); + here we (a) patch it onto the matching ``assistant`` turns missing a real + stop, and (b) append the trailing turn delivered after the simulator drained + ("speech scheduling is paused"). One turn can arrive as several partial or + extended emissions, so match by containment and aggregate min-start/max-stop. + Only populated by the native transcription handler — VAPI/Retell are untouched. """ if not captured_target_turns: return messages + + def _matching(text: str) -> list[dict[str, Any]]: + result = [] + for captured in captured_target_turns: + cap_text = (captured.get("content") or "").strip() + if cap_text and (text in cap_text or cap_text in text): + result.append(captured) + return result + + # (a) Fill timing onto existing assistant turns that lack a real stop; never + # override genuine audio metrics if livekit-agents ever populates them. + for message in messages: + if message.get("role") != "assistant": + continue + text = (message.get("content") or "").strip() + if not text: + continue + started = message.get("started_speaking_at") + stopped = message.get("stopped_speaking_at") + if ( + isinstance(started, (int, float)) + and isinstance(stopped, (int, float)) + and stopped > started + ): + continue + matched = _matching(text) + starts = [ + c["started_speaking_at"] + for c in matched + if isinstance(c.get("started_speaking_at"), (int, float)) + ] + stops = [ + c["stopped_speaking_at"] + for c in matched + if isinstance(c.get("stopped_speaking_at"), (int, float)) + ] + if starts: + message["started_speaking_at"] = min(starts) + if not isinstance(message.get("created_at"), (int, float)): + message["created_at"] = min(starts) + if stops: + message["stopped_speaking_at"] = max(stops) + + # (b) Append target turns that never reached the report at all. assistant_texts = [ (m.get("content") or "").strip() for m in messages @@ -2117,18 +2177,23 @@ def _already_present(text: str) -> bool: last_ts = value merged = list(messages) - for offset, raw in enumerate(captured_target_turns, start=1): - text = (raw or "").strip() + for offset, captured in enumerate(captured_target_turns, start=1): + text = (captured.get("content") or "").strip() if not text or _already_present(text): continue - ts = (last_ts + offset) if last_ts else None + started = captured.get("started_speaking_at") + if not isinstance(started, (int, float)): + started = (last_ts + offset) if last_ts else None + stopped = captured.get("stopped_speaking_at") + if not isinstance(stopped, (int, float)): + stopped = started merged.append( { "role": "assistant", "content": text, - "created_at": ts, - "started_speaking_at": ts, - "stopped_speaking_at": ts, + "created_at": started, + "started_speaking_at": started, + "stopped_speaking_at": stopped, "interrupted": False, "e2e_latency": None, } diff --git a/tests/runtime/test_livekit_engine.py b/tests/runtime/test_livekit_engine.py index 8ddfcf11..1c8ec729 100644 --- a/tests/runtime/test_livekit_engine.py +++ b/tests/runtime/test_livekit_engine.py @@ -2184,7 +2184,7 @@ def test_target_transcription_feeds_simulator_llm_context() -> None: # reads the AGENT's chat context, so the simulator LLM never saw a single # target turn and answered every one with "I can't hear you". session = _FakeReplySession() - captured: list[str] = [] + captured: list[dict] = [] asyncio.run( livekit._forward_target_transcription( @@ -2199,14 +2199,16 @@ def test_target_transcription_feeds_simulator_llm_context() -> None: # The pipeline persists the turn into both contexts itself; a manual # history add here would double the transcript entry. assert session.history_adds == [] - assert captured == ["How can I help you today?"] + assert len(captured) == 1 + assert captured[0]["content"] == "How can I help you today?" + assert captured[0]["started_speaking_at"] <= captured[0]["stopped_speaking_at"] def test_target_transcription_after_end_records_without_reply() -> None: session = _FakeReplySession() ended = asyncio.Event() ended.set() - captured: list[str] = [] + captured: list[dict] = [] asyncio.run( livekit._forward_target_transcription( @@ -2219,12 +2221,12 @@ def test_target_transcription_after_end_records_without_reply() -> None: assert session.reply_inputs == [] assert session.history_adds == [("user", "Goodbye!")] - assert captured == ["Goodbye!"] + assert [c["content"] for c in captured] == ["Goodbye!"] def test_target_transcription_falls_back_to_history_when_session_closing() -> None: session = _FakeReplySession(reply_error=RuntimeError("scheduling is paused")) - captured: list[str] = [] + captured: list[dict] = [] asyncio.run( livekit._forward_target_transcription( @@ -2236,7 +2238,68 @@ def test_target_transcription_falls_back_to_history_when_session_closing() -> No ) assert session.history_adds == [("user", "One last thing.")] - assert captured == ["One last thing."] + assert [c["content"] for c in captured] == ["One last thing."] + + +def test_merge_patches_target_turn_missing_stop_timing() -> None: + # Native target turns reach history via generate_reply(user_input=) with no + # audio metrics, so stop == start (zero duration) -> bot WPM/latency dead. + messages = [ + {"role": "user", "content": "hi", "started_speaking_at": 1.0, + "stopped_speaking_at": 2.0}, + {"role": "assistant", "content": "hello there how can i help", + "started_speaking_at": 3.0, "stopped_speaking_at": 3.0}, + ] + captured = [{"content": "hello there how can i help", + "started_speaking_at": 2.5, "stopped_speaking_at": 5.0}] + merged = livekit._merge_captured_target_turns(messages, captured) + assert len(merged) == 2 # patched in place, not appended + agent = next(m for m in merged if m["role"] == "assistant") + assert agent["started_speaking_at"] == 2.5 + assert agent["stopped_speaking_at"] == 5.0 + + +def test_merge_aggregates_partial_target_emissions() -> None: + # One turn arrives as several partial/extended emissions -> min-start/max-stop. + messages = [ + {"role": "assistant", "content": "can we look at options for that", + "started_speaking_at": 10.0, "stopped_speaking_at": 10.0}, + ] + captured = [ + {"content": "can we look", "started_speaking_at": 9.0, + "stopped_speaking_at": 9.5}, + {"content": "can we look at options for that", + "started_speaking_at": 9.2, "stopped_speaking_at": 12.0}, + ] + merged = livekit._merge_captured_target_turns(messages, captured) + assert merged[0]["started_speaking_at"] == 9.0 + assert merged[0]["stopped_speaking_at"] == 12.0 + + +def test_merge_does_not_override_real_target_timing() -> None: + messages = [ + {"role": "assistant", "content": "genuine turn", + "started_speaking_at": 4.0, "stopped_speaking_at": 6.0}, + ] + captured = [{"content": "genuine turn", "started_speaking_at": 1.0, + "stopped_speaking_at": 99.0}] + merged = livekit._merge_captured_target_turns(messages, captured) + assert merged[0]["started_speaking_at"] == 4.0 + assert merged[0]["stopped_speaking_at"] == 6.0 + + +def test_merge_appends_trailing_target_turn_with_real_timing() -> None: + messages = [{"role": "user", "content": "bye", "started_speaking_at": 5.0, + "stopped_speaking_at": 6.0}] + captured = [{"content": "take care now", "started_speaking_at": 7.0, + "stopped_speaking_at": 8.0}] + merged = livekit._merge_captured_target_turns(messages, captured) + assert len(merged) == 2 + trailing = merged[-1] + assert trailing["role"] == "assistant" + assert trailing["content"] == "take care now" + assert trailing["started_speaking_at"] == 7.0 + assert trailing["stopped_speaking_at"] == 8.0 class _FakeEventRoom: