Improve error handling across multiple components

livekit-agents/livekit/agents/inference/tts.py

@@ -538,6 +538,9 @@ async def _recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
                 if data.get("type") == "session.created":
                     pass
                 elif data.get("type") == "output_audio":
+                    if "audio" not in data:
+                        logger.warning("output_audio message missing 'audio' field")
+                        continue
                     b64data = base64.b64decode(data["audio"])
                     output_emitter.push(b64data)
                 elif data.get("type") == "done":

livekit-agents/livekit/agents/ipc/channel.py

@@ -22,13 +22,20 @@ def read(self, b: io.BytesIO) -> None: ...
 
 
 def _read_message(data: bytes, messages: MessagesDict) -> Message:
-    bio = io.BytesIO(data)
-    msg_id = read_int(bio)
-    msg = messages[msg_id]()
-    if isinstance(msg, DataMessage):
-        msg.read(bio)
+    try:
+        bio = io.BytesIO(data)
+        msg_id = read_int(bio)
+
+        if msg_id not in messages:
+            raise ValueError(f"Unknown message ID: {msg_id}")
+
+        msg = messages[msg_id]()
+        if isinstance(msg, DataMessage):
+            msg.read(bio)
 
-    return msg
+        return msg
+    except Exception as e:
+        raise ValueError(f"Failed to parse message: {e}") from e
 
 
 def _write_message(msg: Message) -> bytes:

livekit-agents/livekit/agents/llm/chat_context.py

@@ -590,6 +590,9 @@ async def _summarize(
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> ChatContext:
+        if "items" not in data:
+            raise ValueError("Missing 'items' key in chat context data")
+
         item_adapter = TypeAdapter(list[ChatItem])
         items = item_adapter.validate_python(data["items"])
         return cls(items)

livekit-agents/livekit/agents/tts/tts.py

@@ -907,8 +907,11 @@ def dump_segment() -> None:
                 if self._streaming
                 else f"lk_dump/{self._label}_{self._request_id}_{ts}.wav"
             )
-            with open(fname, "wb") as f:
-                f.write(rtc.combine_audio_frames(debug_frames).to_wav_bytes())
+            try:
+                with open(fname, "wb") as f:
+                    f.write(rtc.combine_audio_frames(debug_frames).to_wav_bytes())
+            except (OSError, PermissionError) as e:
+                logger.warning(f"Failed to write debug audio file {fname}: {e}")
 
             debug_frames.clear()
 

livekit-agents/livekit/agents/utils/aio/duplex_unix.py

@@ -53,9 +53,11 @@ async def send_bytes(self, data: bytes) -> None:
 
     async def aclose(self) -> None:
         try:
-            self._writer.close()
-            await self._writer.wait_closed()
-            self._sock.close()
+            if hasattr(self, "_writer") and self._writer is not None:
+                self._writer.close()
+                await self._writer.wait_closed()
+            if hasattr(self, "_sock") and self._sock is not None:
+                self._sock.close()
         except OSError as e:
             raise DuplexClosed() from e
 

livekit-agents/livekit/agents/utils/audio.py

@@ -174,15 +174,24 @@ async def audio_frames_from_file(
     decoder = AudioStreamDecoder(sample_rate=sample_rate, num_channels=num_channels)
 
     async def file_reader() -> None:
-        async with aiofiles.open(file_path, mode="rb") as f:
-            while True:
-                chunk = await f.read(4096)
-                if not chunk:
-                    break
-
-                decoder.push(chunk)
-
-        decoder.end_input()
+        try:
+            async with aiofiles.open(file_path, mode="rb") as f:
+                while True:
+                    try:
+                        chunk = await f.read(4096)
+                        if not chunk:
+                            break
+                        decoder.push(chunk)
+                    except OSError as e:
+                        logger.error(f"File read error: {e}")
+                        break
+        except Exception as e:
+            logger.error(f"Failed to open audio file {file_path}: {e}")
+        finally:
+            try:
+                decoder.end_input()
+            except Exception:
+                pass  # Decoder might already be closed
 
     reader_task = asyncio.create_task(file_reader())
 

livekit-agents/livekit/agents/utils/codecs/decoder.py

@@ -256,6 +256,9 @@ def _decode_wav_loop(self) -> None:
                     data += part
                     remaining -= len(part)
                 if chunk_id == b"fmt ":
+                    if len(data) < 16:
+                        raise ValueError("Invalid WAV file: fmt chunk too small")
+
                     audio_format, wave_channels, wave_rate, _, _, bits_per_sample = struct.unpack(
                         "<HHIIHH", data[:16]
                     )

livekit-agents/livekit/agents/utils/http_server.py

@@ -37,5 +37,6 @@ async def start(self) -> None:
 
     async def aclose(self) -> None:
         async with self._lock:
-            self._server.close()
-            await self._server.wait_closed()
+            if hasattr(self, "_server") and self._server is not None:
+                self._server.close()
+                await self._server.wait_closed()

livekit-agents/livekit/agents/utils/hw/cpu.py

@@ -71,19 +71,27 @@ def _read_cpu_max(self) -> tuple[str, int]:
         try:
             with open("/sys/fs/cgroup/cpu.max") as f:
                 data = f.read().strip().split()
-            quota = data[0]
-            period = int(data[1])
-        except FileNotFoundError:
+            if len(data) >= 2:
+                quota = data[0]
+                period = int(data[1])
+            else:
+                raise ValueError("Invalid cpu.max format")
+        except (FileNotFoundError, PermissionError, ValueError, OSError):
             quota = "max"
             period = 100000
         return quota, period
 
     def _read_cpu_usage(self) -> int:
-        with open("/sys/fs/cgroup/cpu.stat") as f:
-            for line in f:
-                if line.startswith("usage_usec"):
-                    return int(line.split()[1])
-        raise RuntimeError("Failed to read CPU usage")
+        try:
+            with open("/sys/fs/cgroup/cpu.stat") as f:
+                for line in f:
+                    if line.startswith("usage_usec"):
+                        parts = line.split()
+                        if len(parts) >= 2:
+                            return int(parts[1])
+        except (FileNotFoundError, PermissionError, ValueError, OSError) as e:
+            raise RuntimeError(f"Failed to read CPU usage: {e}") from e
+        raise RuntimeError("Failed to read CPU usage: usage_usec not found")
 
 
 class CGroupV1CPUMonitor(CPUMonitor):
@@ -136,9 +144,7 @@ def _read_first_int(self, paths: list[str]) -> Optional[int]:
             try:
                 with open(p) as f:
                     return int(f.read().strip())
-            except FileNotFoundError:
-                continue
-            except ValueError:
+            except (FileNotFoundError, PermissionError, ValueError, OSError):
                 continue
         return None