fix(simulate): harden hosted voice-sim reliability

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}"
         )

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,

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

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: