fix(logging): thread-safe queue state + bounded hard-exit drain + record copy

Self-review (3-agent + codex) findings on the async QueueListener change:

1. (HIGH) The os._exit shutdown backstop called flush_log_queue(), whose
   stop() joins the listener thread unbounded. If that thread is wedged on
   the rotation lock — the exact failure this change survives — shutdown
   re-freezes. Add drain_log_queue(timeout): stop-only, bounded via a
   throwaway joiner thread. Also release PID/runtime locks BEFORE the drain
   so a slow drain can't strand them.

2. (MED) _log_queue/_queue_listener/_queued_file_handlers were read-modify-
   written without a lock across register/stop/flush/reset; a gateway-init
   race with a plugin/CLI path could leave two live listeners. Guard all
   four globals with a single _queue_state_lock.

3. (MED) _NonFormattingQueueHandler.prepare() enqueued the same LogRecord a
   synchronous handler on the emitting thread may still format/mutate.
   Return copy.copy(record) (preserves msg/args/exc_info for deferred
   RedactingFormatter) to remove the cross-thread mutation race.

E2E-verified: bounded drain returns in ~500ms on a permanently-wedged
listener; 4x20 concurrent flushes single-listener no-crash; args still
format and secrets still redact through the copied record.

gateway/run.py

@@ -20239,36 +20239,41 @@ def _exit_after_graceful_shutdown(exit_code: int) -> None:
     released — so this is a no-op on the normal shutdown path and the actual
     cleanup on the early-exit paths.
 
-    Logging IS flushed here: the rotating file handlers are driven by an
+    Logging IS drained here: the rotating file handlers are driven by an
     async ``QueueListener`` on a dedicated thread (see
     ``hermes_logging._register_queued_handler``), so records emitted right
     before shutdown may still be sitting in the in-memory queue. ``os._exit``
     below bypasses ``atexit``, so the ``atexit``-registered listener drain
-    never runs on this path — we must drain explicitly here or lose the last
+    never runs on this path — we drain explicitly (bounded, via
-    log lines (including the shutdown reason on the early-exit paths). Stdio
+    ``drain_log_queue``) or lose the last log lines (including the shutdown
-    is flushed too.
+    reason on the early-exit paths). Stdio is flushed too.
     """
     for stream in (sys.stdout, sys.stderr):
         try:
             stream.flush()
         except Exception:
             pass
-    # Drain the async log queue: os._exit bypasses atexit, so the listener's
+    # Release PID + runtime lock BEFORE the log drain: the drain is bounded but
-    # atexit drain won't fire. flush_log_queue() no-ops when logging never
+    # could still take up to its timeout on a wedged disk, and these locks must
-    # initialized a queue (e.g. very early aborts), so this is always safe.
+    # never be stranded. os._exit skips atexit, and the early SystemExit exit
-    try:
+    # paths never run _stop_impl, so release here (idempotent).
-        from hermes_logging import flush_log_queue
-        flush_log_queue()
-    except Exception:
-        pass
-    # Guaranteed cleanup chokepoint: os._exit skips atexit, and the early
-    # SystemExit exit paths never run _stop_impl, so release here (idempotent).
     try:
         from gateway.status import remove_pid_file, release_gateway_runtime_lock
         remove_pid_file()
         release_gateway_runtime_lock()
     except Exception:
         pass
+    # Drain the async log queue: os._exit bypasses atexit, so the listener's
+    # atexit drain won't fire. Use drain_log_queue() (bounded, no restart), NOT
+    # flush_log_queue(): if the listener is wedged on the rotation lock — the
+    # exact failure this async-logging change survives — an unbounded stop()
+    # join would re-freeze the shutdown. drain_log_queue() no-ops when logging
+    # never initialized a queue (very early aborts), so this is always safe.
+    try:
+        from hermes_logging import drain_log_queue
+        drain_log_queue(timeout=1.0)
+    except Exception:
+        pass
     os._exit(exit_code)
 
 

hermes_logging.py

@@ -28,6 +28,7 @@ Session context:
 """
 
 import atexit
+import copy
 import io
 import logging
 import os
@@ -562,6 +563,13 @@ _log_queue: "Optional[queue.SimpleQueue]" = None
 _queue_listener: Optional[QueueListener] = None
 _queued_file_handlers: list = []
 _queue_atexit_registered = False
+# Guards every read-modify-write of the four globals above. setup_logging()
+# holds no lock and its _logging_initialized guard runs AFTER handler
+# registration, so _register_queued_handler() can run concurrently with a
+# flush/reset from another thread (gateway init racing a plugin/CLI path).
+# Without this, two threads can interleave listener.stop()/reassign/start()
+# and leave the queue with two live listeners or an orphaned worker thread.
+_queue_state_lock = threading.Lock()
 
 
 class _NonFormattingQueueHandler(QueueHandler):
@@ -569,16 +577,23 @@ class _NonFormattingQueueHandler(QueueHandler):
 
     Stdlib ``prepare()`` formats the record and drops ``args``/``exc_info`` so it
     can be pickled to another process.  Our queue is in-process, so we skip that
-    and pass the raw record through — the target file handlers must apply their
+    and hand the target file handlers an unformatted record — they apply their
     own ``RedactingFormatter`` and component filters on the listener thread.
+
+    We return a **shallow copy** rather than the original record: the same
+    record is still owned by the emitting thread (and any synchronous handler
+    on it, e.g. a ``StreamHandler``), which may format/mutate ``record.message``
+    while our listener thread reads it. Copying preserves ``msg``/``args``/
+    ``exc_info`` for the deferred format while removing the cross-thread
+    mutation race on a shared object.
     """
 
     def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
-        return record
+        return copy.copy(record)
 
 
-def _stop_queue_listener() -> None:
+def _stop_queue_listener_locked() -> None:
-    """Flush and stop the background log listener (idempotent)."""
+    """Stop the listener assuming ``_queue_state_lock`` is already held."""
     global _queue_listener
     listener, _queue_listener = _queue_listener, None
     if listener is not None:
@@ -588,47 +603,92 @@ def _stop_queue_listener() -> None:
             pass
 
 
+def _stop_queue_listener() -> None:
+    """Flush and stop the background log listener (idempotent, thread-safe).
+
+    This is the atexit hook, so it must acquire the state lock itself.
+    """
+    with _queue_state_lock:
+        _stop_queue_listener_locked()
+
+
 def _register_queued_handler(handler: logging.Handler) -> None:
     """Route *handler* through the shared async queue instead of attaching it to
     *root* directly, so emitting threads never block on file I/O or the
     cross-process rotation lock.  The ``QueueListener`` applies each handler's
     own level and filters on its worker thread."""
     global _log_queue, _queue_listener, _queue_atexit_registered
-    if _log_queue is None:
+    with _queue_state_lock:
-        _log_queue = queue.SimpleQueue()
+        if _log_queue is None:
-        qh = _NonFormattingQueueHandler(_log_queue)
+            _log_queue = queue.SimpleQueue()
-        qh._hermes_queue = True  # type: ignore[attr-defined]
+            qh = _NonFormattingQueueHandler(_log_queue)
-        # Always funnel through the root logger so records from any logger
+            qh._hermes_queue = True  # type: ignore[attr-defined]
-        # (production passes root here; callers may pass a child) reach the
+            # Always funnel through the root logger so records from any logger
-        # queue via propagation.
+            # (production passes root here; callers may pass a child) reach the
-        logging.getLogger().addHandler(qh)
+            # queue via propagation.
-    _queued_file_handlers.append(handler)
+            logging.getLogger().addHandler(qh)
-    # Rebuild the listener with the full target set.  This only happens while
+        _queued_file_handlers.append(handler)
-    # init_logging() adds handlers (2-3 times, queue empty), so stop() returns
+        # Rebuild the listener with the full target set.  This only happens
-    # immediately.
+        # while init_logging() adds handlers (2-3 times, queue empty), so
-    if _queue_listener is not None:
+        # stop() returns immediately.
-        _queue_listener.stop()
+        if _queue_listener is not None:
-    _queue_listener = QueueListener(
+            _queue_listener.stop()
-        _log_queue, *_queued_file_handlers, respect_handler_level=True
+        _queue_listener = QueueListener(
-    )
+            _log_queue, *_queued_file_handlers, respect_handler_level=True
-    _queue_listener.start()
+        )
-    if not _queue_atexit_registered:
+        _queue_listener.start()
-        # Runs before logging.shutdown (registered earlier at import time), so
+        if not _queue_atexit_registered:
-        # the listener stops before its file handlers are closed.
+            # Runs before logging.shutdown (registered earlier at import time),
-        atexit.register(_stop_queue_listener)
+            # so the listener stops before its file handlers are closed.
-        _queue_atexit_registered = True
+            atexit.register(_stop_queue_listener)
+            _queue_atexit_registered = True
 
 
 def flush_log_queue() -> None:
     """Block until all queued records have been written, then resume.
 
     Draining is done by stopping the listener (which processes every pending
-    record before joining) and restarting it.  Used at shutdown and by tests
+    record before joining) and restarting it.  Used by tests that read a log
-    that read a log file right after emitting to it."""
+    file right after emitting to it.
+
+    NOTE: ``stop()`` joins the worker thread, so this blocks until the queue
+    is empty. Do NOT call this on a hard-exit path where the listener may be
+    wedged on the rotation lock — use ``drain_log_queue()`` there instead,
+    which bounds the wait.
+    """
+    with _queue_state_lock:
+        listener = _queue_listener
+        if listener is not None:
+            listener.stop()
+            listener.start()
+
+
+def drain_log_queue(timeout: float = 1.0) -> None:
+    """Best-effort, time-bounded drain for hard-exit paths (no restart).
+
+    Unlike ``flush_log_queue()``, this stops the listener WITHOUT restarting it
+    (the process is about to exit) and bounds the drain: if the listener's
+    worker thread is wedged on the cross-process rotation lock — the very
+    failure this async-logging change exists to survive — an unbounded
+    ``stop()``/join would re-freeze the shutdown path. We run ``stop()`` on a
+    throwaway thread and only wait ``timeout`` seconds for it; if it hasn't
+    drained by then we abandon the last few records and let ``os._exit``
+    proceed. Availability beats the last log line when the disk is already
+    wedged.
+    """
     listener = _queue_listener
-    if listener is not None:
+    if listener is None:
-        listener.stop()
+        return
-        listener.start()
+
+    def _drain() -> None:
+        try:
+            listener.stop()
+        except Exception:
+            pass
+
+    t = threading.Thread(target=_drain, name="hermes-log-drain", daemon=True)
+    t.start()
+    t.join(timeout)
 
 
 def rotating_file_handlers() -> list:
@@ -643,18 +703,19 @@ def rotating_file_handlers() -> list:
 def _reset_queued_handlers() -> None:
     """Tear down the async logging queue + listener (test-isolation helper)."""
     global _log_queue
-    _stop_queue_listener()
+    with _queue_state_lock:
-    root = logging.getLogger()
+        _stop_queue_listener_locked()
-    for h in list(root.handlers):
+        root = logging.getLogger()
-        if getattr(h, "_hermes_queue", False):
+        for h in list(root.handlers):
-            root.removeHandler(h)
+            if getattr(h, "_hermes_queue", False):
-    for h in list(_queued_file_handlers):
+                root.removeHandler(h)
-        try:
+        for h in list(_queued_file_handlers):
-            h.close()
+            try:
-        except Exception:
+                h.close()
-            pass
+            except Exception:
-    _queued_file_handlers.clear()
+                pass
-    _log_queue = None
+        _queued_file_handlers.clear()
+        _log_queue = None
 
 
 def _add_rotating_handler(