Thread safety fixes for session and message counts

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

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

  merged to main

  files: gateway/platforms/email.py, hermes_state.py, run_agent.py, tools/voice_mode.py

  Show the diff

  • 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,