Guard print/isatty against closed stdout streams

Fixes on record

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

• #955 fix: _SafeWriter for broken stdout + doctor MiniMax + vision logging

  merged to main

  files: tools/vision_tools.py

• #963 fix: guard all print() against OSError with _SafeWriter

  merged to main

  files: run_agent.py, tests/test_run_agent.py

• #1668 fix(core): guard print() calls in run_conversation() against OSError

  merged to main

  files: run_agent.py

  Show the diff

  • run_agent.py

    @@ -856,6 +856,19 @@ def __init__(
                 else:
                     print(f"📊 Context limit: {self.context_compressor.context_length:,} tokens (auto-compression disabled)")
         
    +    @staticmethod
    +    def _safe_print(*args, **kwargs):
    +        """Print that silently handles broken pipes / closed stdout.
    +
    +        In headless environments (systemd, Docker, nohup) stdout may become
    +        unavailable mid-session.  A raw ``print()`` raises ``OSError`` which
    +        can crash cron jobs and lose completed work.
    +        """
    +        try:
    +            print(*args, **kwargs)
    +        except OSError:
    +            pass
    +
         def _vprint(self, *args, force: bool = False, **kwargs):
             """Verbose print — suppressed when streaming TTS is active.
     
    @@ -864,7 +877,7 @@ def _vprint(self, *args, force: bool = False, **kwargs):
             """
             if not force and self._has_stream_consumers():
                 return
    -        print(*args, **kwargs)
    +        self._safe_print(*args, **kwargs)
     
         def _max_tokens_param(self, value: int) -> dict:
             """Return the correct max tokens kwarg for the current provider.
    @@ -4752,7 +4765,7 @@ def run_conversation(
             self._persist_user_message_idx = current_turn_user_idx
             
             if not self.quiet_mode:
    -            print(f"💬 Starting conversation: '{user_message[:60]}{'...' if len(user_message) > 60 else ''}'")
    +            self._safe_print(f"💬 Starting conversation: '{user_message[:60]}{'...' if len(user_message) > 60 else ''}'")
             
             # ── System prompt (cached per session for prefix caching) ──
             # Built once on first call, reused for all subsequent calls.
    @@ -4822,7 +4835,7 @@ def run_conversation(
                         f"{self.context_compressor.context_length:,}",
                     )
                     if not self.quiet_mode:
    -                    print(
    +                    self._safe_print(
                             f"📦 Preflight compression: ~{_preflight_tokens:,} tokens "
                             f">= {self.context_compressor.threshold_tokens:,} threshold"
                         )
    @@ -4862,13 +4875,13 @@ def run_conversation(
                 if self._interrupt_requested:
                     interrupted = True
                     if not self.quiet_mode:
    -                    print(f"\n⚡ Breaking out of tool loop due to interrupt...")
    +                    self._safe_print(f"\n⚡ Breaking out of tool loop due to interrupt...")
                     break
                 
                 api_call_count += 1
                 if not self.iteration_budget.consume():
                     if not self.quiet_mode:
    -                    print(f"\n⚠️  Session iteration budget exhausted ({self.iteration_budget.max_total} total across agent + subagents)")
    +                    self._safe_print(f"\n⚠️  Session iteration budget exhausted ({self.iteration_budget.max_total} total across agent + subagents)")
                     break
     
                 # Fire step_callback for gateway hooks (agent:step event)
    @@ -5287,7 +5300,7 @@ def _stop_spinner():
                             if self.context_compressor._context_probed:
                                 ctx = self.context_compressor.context_length
                                 save_context_length(self.model, self.base_url, ctx)
    -                            print(f"{self.log_prefix}💾 Cached context length: {ctx:,} tokens for {self.model}")
    +                            self._safe_print(f"{self.log_prefix}💾 Cached context length: {ctx:,} tokens for {self.model}")
                                 self.context_compressor._context_probed = False
     
                             self.session_prompt_tokens += prompt_tokens
    @@ -6129,12 +6142,15 @@ def _stop_spinner():
                         messages.append(final_msg)
                         
                         if not self.quiet_mode:
    -                        print(f"🎉 Conversation completed after {api_call_count} OpenAI-compatible API call(s)")
    +                        self._safe_print(f"🎉 Conversation completed after {api_call_count} OpenAI-compatible API call(s)")
                         break
                     
                 except Exception as e:
                     error_msg = f"Error during OpenAI-compatible API call #{api_call_count}: {str(e)}"
    -                print(f"❌ {error_msg}")
    +                try:
    +                    print(f"❌ {error_msg}")
    +                except OSError:
    +                    logger.error(error_msg)
                     
                     if self.verbose_logging:
                         logging.exception("Detailed error information:")

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

• #3843 fix(cli): handle closed stdout ValueError in safe print paths

  merged to main

  files: run_agent.py

  Show the diff

  • run_agent.py

    @@ -1285,7 +1285,7 @@ def _safe_print(self, *args, **kwargs):
             try:
                 fn = self._print_fn or print
                 fn(*args, **kwargs)
    -        except OSError:
    +        except (OSError, ValueError):
                 pass
     
         def _vprint(self, *args, force: bool = False, **kwargs):
    @@ -7909,7 +7909,7 @@ def _stop_spinner():
                     error_msg = f"Error during OpenAI-compatible API call #{api_call_count}: {str(e)}"
                     try:
                         print(f"❌ {error_msg}")
    -                except OSError:
    +                except (OSError, ValueError):
                         logger.error(error_msg)
                     
                     if self.verbose_logging: