fix: batch of 5 small contributor fixes — PortAudio, SafeWriter, IMAP, thread lock, prefill

gateway/platforms/email.py

@@ -230,7 +230,7 @@ async def connect(self) -> bool:
             # Mark all existing messages as seen so we only process new ones
             imap.select("INBOX")
             status, data = imap.uid("search", None, "ALL")
-            if status == "OK" and data[0]:
+            if status == "OK" and data and data[0]:
                 for uid in data[0].split():
                     self._seen_uids.add(uid)
             imap.logout()
@@ -295,7 +295,7 @@ def _fetch_new_messages(self) -> List[Dict[str, Any]]:
             imap.select("INBOX")
 
             status, data = imap.uid("search", None, "UNSEEN")
-            if status != "OK" or not data[0]:
+            if status != "OK" or not data or not data[0]:
                 imap.logout()
                 return results
 

hermes_state.py

@@ -855,23 +855,25 @@ def search_sessions(
 
     def session_count(self, source: str = None) -> int:
         """Count sessions, optionally filtered by source."""
-        if source:
-            cursor = self._conn.execute(
-                "SELECT COUNT(*) FROM sessions WHERE source = ?", (source,)
-            )
-        else:
-            cursor = self._conn.execute("SELECT COUNT(*) FROM sessions")
-        return cursor.fetchone()[0]
+        with self._lock:
+            if source:
+                cursor = self._conn.execute(
+                    "SELECT COUNT(*) FROM sessions WHERE source = ?", (source,)
+                )
+            else:
+                cursor = self._conn.execute("SELECT COUNT(*) FROM sessions")
+            return cursor.fetchone()[0]
 
     def message_count(self, session_id: str = None) -> int:
         """Count messages, optionally for a specific session."""
-        if session_id:
-            cursor = self._conn.execute(
-                "SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,)
-            )
-        else:
-            cursor = self._conn.execute("SELECT COUNT(*) FROM messages")
-        return cursor.fetchone()[0]
+        with self._lock:
+            if session_id:
+                cursor = self._conn.execute(
+                    "SELECT COUNT(*) FROM messages WHERE session_id = ?", (session_id,)
+                )
+            else:
+                cursor = self._conn.execute("SELECT COUNT(*) FROM messages")
+            return cursor.fetchone()[0]
 
     # =========================================================================
     # Export and cleanup

run_agent.py

@@ -108,7 +108,7 @@
 
 
 class _SafeWriter:
-    """Transparent stdio wrapper that catches OSError from broken pipes.
+    """Transparent stdio wrapper that catches OSError/ValueError from broken pipes.
 
     When hermes-agent runs as a systemd service, Docker container, or headless
     daemon, the stdout/stderr pipe can become unavailable (idle timeout, buffer
@@ -117,8 +117,13 @@ class _SafeWriter:
     run_conversation() — especially via double-fault when an except handler
     also tries to print.
 
+    Additionally, when subagents run in ThreadPoolExecutor threads, the shared
+    stdout handle can close between thread teardown and cleanup, raising
+    ``ValueError: I/O operation on closed file`` instead of OSError.
+
     This wrapper delegates all writes to the underlying stream and silently
-    catches OSError. It is transparent when the wrapped stream is healthy.
+    catches both OSError and ValueError. It is transparent when the wrapped
+    stream is healthy.
     """
 
     __slots__ = ("_inner",)
@@ -129,13 +134,13 @@ def __init__(self, inner):
     def write(self, data):
         try:
             return self._inner.write(data)
-        except OSError:
+        except (OSError, ValueError):
             return len(data) if isinstance(data, str) else 0
 
     def flush(self):
         try:
             self._inner.flush()
-        except OSError:
+        except (OSError, ValueError):
             pass
 
     def fileno(self):
@@ -144,7 +149,7 @@ def fileno(self):
     def isatty(self):
         try:
             return self._inner.isatty()
-        except OSError:
+        except (OSError, ValueError):
             return False
 
     def __getattr__(self, name):
@@ -2438,7 +2443,18 @@ def _sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any
                 "Pre-call sanitizer: added %d stub tool result(s)",
                 len(missing_results),
             )
-
+        # 3. Strip trailing empty assistant messages to prevent prefill rejection.
+        # These can leak from Responses API reasoning-only turns (Codex/MiniMax)
+        # where an empty assistant message is required by the Responses API but
+        # must NOT be sent to Chat Completions or Anthropic Messages API providers.
+        while (
+            messages
+            and messages[-1].get("role") == "assistant"
+            and not (messages[-1].get("content") or "").strip()
+            and not messages[-1].get("tool_calls")
+        ):
+            logger.debug("Pre-call sanitizer: removed trailing empty assistant message")
+            messages = messages[:-1]
         return messages
 
     @staticmethod

tools/voice_mode.py

@@ -81,8 +81,15 @@ def detect_audio_environment() -> dict:
                 warnings.append("No audio input/output devices detected")
         except Exception:
             warnings.append("Audio subsystem error (PortAudio cannot query devices)")
-    except (ImportError, OSError):
+    except ImportError:
         warnings.append("Audio libraries not installed (pip install sounddevice numpy)")
+    except OSError:
+        warnings.append(
+            "PortAudio system library not found -- install it first:\n"
+            "  Linux:  sudo apt-get install libportaudio2\n"
+            "  macOS:  brew install portaudio\n"
+            "Then retry /voice on."
+        )
 
     return {
         "available": len(warnings) == 0,