"""Centralized logging setup for Hermes Agent.

Log files: agent.log (INFO+, everything), errors.log (WARNING+), gateway.log (INFO+,
gateway components; ``mode="gateway"``), gui.log (INFO+, dashboard/TUI-gateway;
``mode="gui"``). All are rotating files driven through one async queue and formatted
with ``RedactingFormatter`` so secrets never reach disk.
"""

import atexit
import contextlib
import copy
import io
import logging
import os
import queue
import sys
import threading
from logging.handlers import QueueHandler, QueueListener
from pathlib import Path
from time import monotonic as _monotonic
from typing import Optional, Sequence

from hermes_constants import (
    get_config_path, get_hermes_home, mkdir_under_hermes_home, named_profile_is_deleted,
)

# setup_logging() is idempotent: a second call is a no-op unless ``force=True``.
_logging_initialized = False

# True only when CLH was rejected at import because portalocker cannot take a
# lock on this Windows box; file handlers then use stdlib rotation (rollover
# disabled — see the module-header comment) and setup_logging() warns once.
_WINDOWS_CLH_FALLBACK = False
_WINDOWS_CLH_FALLBACK_REASON = ""
_fallback_warned = False


def _portalocker_probe() -> bool:
    """Return True when portalocker can actually take a lock on this box.

    concurrent-log-handler locks every write through portalocker, which on
    Windows instantiates Win32Locker and imports pywintypes. Bundled payloads
    have shipped with that import broken (the venv's .pth files were never
    processed), and portalocker 3.x gives no msvcrt fallback — every emit then
    dies with the ImportError, CLH retries 20x, and the suppressed "Cannot
    acquire lock" RuntimeError below hides it completely. Probe a scratch file
    once at import so we can fall back to stdlib rotation instead of silently
    dropping every record. No-op (True) off Windows, where stdlib is in use.
    """
    global _WINDOWS_CLH_FALLBACK_REASON
    if sys.platform != "win32":
        return True
    try:
        import portalocker
        import tempfile
    except Exception as exc:
        _WINDOWS_CLH_FALLBACK_REASON = repr(exc)
        return False
    fd, path = tempfile.mkstemp(prefix="hermes-portalocker-")
    try:
        with os.fdopen(fd, "r+b") as stream:
            portalocker.lock(stream, portalocker.LOCK_EX)
            portalocker.unlock(stream)
    except Exception as exc:
        _WINDOWS_CLH_FALLBACK_REASON = repr(exc)
        return False
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass
    return True


# Windows-ONLY swap (#44873): stdlib ``RotatingFileHandler.doRollover()`` calls
# ``os.rename()``, which fails with ``PermissionError [WinError 32]`` whenever
# another process holds an append handle on ``agent.log`` — essentially always
# in Hermes (TUI, gateway, hy_memory, MCP servers, CLI commands all log) —
# pinning the file at the size threshold and spamming stderr on every emit.
# ``concurrent-log-handler`` serializes rollover with a cross-process lock.
# POSIX keeps stdlib: renames of open files work, and managed mode (NixOS)
# relies on stdlib's exact ``_open()``/``doRollover()`` lifecycle for the
# 0660 chmod and eager file creation; CLH opens lazily and rotates differently.
if sys.platform == "win32":
    if _portalocker_probe():
        from concurrent_log_handler import (  # noqa: E402
            ConcurrentRotatingFileHandler as RotatingFileHandler,
        )
    else:
        # portalocker cannot take a lock on this box (typical cause: a sealed
        # bundle whose venv never processed pywin32.pth, so `import pywintypes`
        # fails and portalocker's Win32Locker has no msvcrt fallback). CLH
        # would silently drop every record through the suppressed lock-timeout
        # below; fall back to stdlib rotation instead. Rollover is disabled in
        # the fallback: multi-process appends make Windows renames fail with
        # WinError 32, the exact #44873 trap CLH exists to avoid.
        from logging.handlers import RotatingFileHandler  # noqa: E402

        _WINDOWS_CLH_FALLBACK = True
else:
    from logging.handlers import RotatingFileHandler  # noqa: E402

# Thread-local per-conversation session context.
_session_context = threading.local()

# ``%(session_tag)s`` exists on every LogRecord via _install_session_record_factory().
_LOG_FORMAT = "%(asctime)s %(levelname)s%(session_tag)s %(name)s: %(message)s"
_LOG_FORMAT_VERBOSE = "%(asctime)s - %(name)s - %(levelname)s%(session_tag)s - %(message)s"


# The stdout stream _line_buffer_piped_stdout() already reconfigured: setup_logging runs on every
# AIAgent build (per message in the gateway), so later calls skip the flush + reconfigure.
_line_buffered_stdout = None


def _line_buffer_piped_stdout() -> None:
    """Best-effort line buffering for a piped stdout (#92281).

    Python block-buffers stdout when it isn't a TTY, so an agent loop
    driving ``print()`` into a supervisor's pipe delivers its output in
    large delayed bursts — a headless run looks stalled for minutes while
    the work is actually progressing (only stderr, which we already give
    ``line_buffering=True`` in ``_safe_stderr()``, shows up on time).
    Reconfigure the interpreter's own stdout for line buffering when it is
    piped; interactive TTY stdout is already line-buffered. ACP reaches this
    too (its AIAgent calls ``setup_logging``) with stdout as its protocol
    channel — harmless, since line buffering only adds flushes at newlines.
    """
    global _line_buffered_stdout
    stream = sys.stdout
    if stream is None or stream is _line_buffered_stdout:
        return
    if getattr(stream, "line_buffering", False) is True:
        _line_buffered_stdout = stream
        return
    reconfigure = getattr(stream, "reconfigure", None)
    if reconfigure is None:
        return
    # A closed/detached stream raises ValueError / io.UnsupportedOperation (an OSError);
    # buffering is observability, not correctness — never crash setup_logging over it.
    with contextlib.suppress(OSError, ValueError):
        if not stream.isatty():
            reconfigure(line_buffering=True)
            _line_buffered_stdout = stream


def _safe_stderr():  # type: ignore[return]
    """Return a stderr stream that tolerates Unicode on all platforms.

    Wraps ``sys.stderr`` with ``errors='replace'`` so un-encodable characters become
    ``?`` instead of crashing the process.
    """
    stream = sys.stderr
    encoding = getattr(stream, "encoding", None) or "utf-8"
    if encoding.lower().replace("-", "") in ("utf8", "utf8surrogateescape"):
        return stream
    try:
        wrapped = io.TextIOWrapper(stream.buffer, encoding="utf-8", errors="replace", line_buffering=True)
        # Prevent the wrapper from closing the underlying buffer when garbage-collected.
        wrapped.close = lambda: None  # type: ignore[assignment]
        return wrapped
    except Exception:
        return stream  # best-effort: no buffer / wrapping failed -> original stream


def _is_windows_concurrent_log_lock_timeout(exc: BaseException | None) -> bool:
    """True for concurrent-log-handler's Windows lock timeout.

    Slash-command workers and the gateway share rotating files on Windows Desktop;
    when another process holds the rollover lock too long CLH raises this
    RuntimeError, which must not escape into Desktop chat output.
    """
    return (
        sys.platform == "win32"
        and isinstance(exc, RuntimeError)
        and "Cannot acquire lock after 20 attempts" in str(exc)
    )


_windows_lock_timeout_warned = False
_windows_lock_timeout_warn_lock = threading.Lock()


def _warn_windows_lock_timeout_once() -> None:
    """Report a suppressed CLH lock timeout exactly once per process.

    Every emit after the first failure raises the same RuntimeError, so the
    warning must be one-shot or it would spam errors.log as badly as the
    stderr noise it replaces. CLH does not chain the underlying cause (the
    RuntimeError is raised outside the except, from the retry loop's else
    clause), so the message cannot include it; a constantly repeating timeout
    means file logging is degraded — the startup portalocker probe in this
    module should have caught a dead portalocker and fallen back already.
    """
    global _windows_lock_timeout_warned
    with _windows_lock_timeout_warn_lock:
        if _windows_lock_timeout_warned:
            return
        _windows_lock_timeout_warned = True
    logging.getLogger("hermes_logging").warning(
        "concurrent-log-handler timed out acquiring the cross-process log "
        "lock; this and later records were dropped (the Desktop slash-worker "
        "surface stays clean, but file logging is degraded)."
    )


def _is_unavailable_log_stream(exc: BaseException | None) -> bool:
    """True when a file handler lost its backing stream during teardown or I/O."""
    return (
        (isinstance(exc, OSError) and exc.errno == 5)
        or (isinstance(exc, ValueError) and "closed file" in str(exc).lower())
    )


# Third-party loggers that are noisy at DEBUG/INFO level.
_NOISY_LOGGERS = (
    "openai", "openai._base_client", "httpx", "httpcore", "asyncio", "hpack", "hpack.hpack",
    "grpc", "modal", "urllib3", "urllib3.connectionpool", "websockets", "charset_normalizer",
    "markdown_it",
)


def _quiet_noisy_loggers() -> None:
    """Pin noisy third-party loggers at WARNING."""
    for name in _NOISY_LOGGERS:
        logging.getLogger(name).setLevel(logging.WARNING)


def set_session_context(session_id: str) -> None:
    """Set the session ID for the current thread."""
    _session_context.session_id = session_id


def clear_session_context() -> None:
    """Clear the session ID for the current thread."""
    _session_context.session_id = None


def _install_session_record_factory() -> None:
    """Replace the global LogRecord factory with one that adds ``session_tag``.

    Unlike a Filter, the record factory runs for EVERY record in the process (propagated
    and third-party-handled ones included), so ``%(session_tag)s`` never KeyErrors.
    Idempotent via a marker attribute.
    """
    current_factory = logging.getLogRecordFactory()
    if getattr(current_factory, "_hermes_session_injector", False):
        return

    def _session_record_factory(*args, **kwargs):
        record = current_factory(*args, **kwargs)
        sid = getattr(_session_context, "session_id", None)
        record.session_tag = f" [{sid}]" if sid else ""  # type: ignore[attr-defined]
        # QueueListener formats on its own thread, after the profile-scoped
        # ContextVar is gone; keep the resolved home on the record so a
        # multiplex desktop ticker can route to the job owner's files (#97489).
        try:
            record.hermes_home = str(get_hermes_home().resolve())  # type: ignore[attr-defined]
        except Exception:
            record.hermes_home = ""  # type: ignore[attr-defined]
        return record
    _session_record_factory._hermes_session_injector = True  # type: ignore[attr-defined]
    logging.setLogRecordFactory(_session_record_factory)


# Install on import so session_tag exists on all records even before setup_logging().
_install_session_record_factory()


class _ComponentFilter(logging.Filter):
    """Only pass records whose logger name starts with one of *prefixes*."""

    def __init__(self, prefixes: Sequence[str]) -> None:
        super().__init__()
        self._prefixes = tuple(prefixes)

    def filter(self, record: logging.LogRecord) -> bool:
        return record.name.startswith(self._prefixes)


# Logger name prefixes per component; used by _ComponentFilter and ``hermes logs --component``.
COMPONENT_PREFIXES = {
    # ``plugins.platforms``: messaging adapters that migrated out of
    # ``gateway/platforms/`` into bundled plugins (#41112) are still gateway
    # components and belong in gateway.log.
    "gateway": ("gateway", "hermes_plugins", "plugins.platforms"),
    "agent": ("agent", "run_agent", "model_tools", "batch_runner"),
    "tools": ("tools",),
    "cli": ("hermes_cli", "cli"),
    "cron": ("cron",),
    "gui": ("hermes_cli.web_server", "hermes_cli.pty_bridge", "hermes_cli.desktop", "tui_gateway", "uvicorn"),
}


def _known_log_homes() -> set[Path]:
    """Homes the queued file handlers already serve: static handlers by their file, routers by
    their default home plus every profile home they route. Caller holds ``_queue_state_lock``."""
    homes: set[Path] = set()
    for handler in _queued_file_handlers:
        if isinstance(handler, _ProfileRoutingFileHandler):
            homes.add(handler._default_home)
            homes.update(handler._profile_homes)
        elif isinstance(handler, RotatingFileHandler):
            try:
                homes.add(Path(handler.baseFilename).resolve().parent.parent)
            except (TypeError, ValueError, OSError):
                continue
    return homes


def _adopt_secondary_home(home: Path) -> bool:
    """Route *home*'s records to its own files when this process already logs for another home.
    Enables profile routing for the union of homes (or widens the live routers); False when
    *home* is the first home seen or is already served."""
    try:
        resolved = Path(home).expanduser().resolve()
    except (TypeError, ValueError, OSError):
        return False
    with _queue_state_lock:
        known = _known_log_homes()
    if not known or resolved in known:
        return False
    return enable_profile_log_routing([*sorted(known), resolved])


def setup_logging(
    *,
    hermes_home: Optional[Path] = None,
    log_level: Optional[str] = None,
    max_size_mb: Optional[int] = None,
    backup_count: Optional[int] = None,
    mode: Optional[str] = None,
    force: bool = False,
) -> Path:
    """Configure the Hermes logging subsystem; returns the ``logs/`` directory.

    Safe to call multiple times; the second call is a no-op unless *force*. Level and
    rotation defaults come from config.yaml ``logging.*``. ``mode="gateway"`` adds
    ``gateway.log`` and ``mode="gui"`` adds ``gui.log``.
    """
    global _logging_initialized
    global _fallback_warned
    home = hermes_home or get_hermes_home()
    log_dir = mkdir_under_hermes_home(home / "logs")

    # Stdout is block-buffered when piped (no TTY); line-buffer it so a
    # headless supervisor's log stream tracks the agent loop incrementally
    # (#92281). Runs before the initialized check so every entry mode that
    # reaches this function gets it; a no-op once this stdout is line-buffered.
    _line_buffer_piped_stdout()

    # A second Hermes home in a process that already logs for another one — a dashboard or
    # ``hermes serve`` backend building agents for several profiles, a multiplexed gateway —
    # gets routed by record home. Stacking another file handler here would hand it EVERY
    # profile's records (the handlers carry no home filter), and a duplicate writer on top of
    # an existing router.
    if _adopt_secondary_home(home):
        return log_dir
    cfg_level, cfg_max_size, cfg_backup = _read_logging_config()
    level_name = (log_level or cfg_level or "INFO").upper()
    level = getattr(logging, level_name, logging.INFO)
    max_bytes = (max_size_mb or cfg_max_size or 5) * 1024 * 1024
    backups = backup_count or cfg_backup or 3

    from agent.redact import RedactingFormatter  # lazy: circular at module load

    root = logging.getLogger()

    # (filename, level, max_bytes, backup_count, component) — a component gates
    # the file on ``mode`` and restricts it to that component's logger prefixes.
    handler_specs = (
        ("agent.log", level, max_bytes, backups, None),
        ("errors.log", logging.WARNING, 2 * 1024 * 1024, 2, None),
        ("gateway.log", logging.INFO, 5 * 1024 * 1024, 3, "gateway"),
        ("gui.log", logging.INFO, 10 * 1024 * 1024, 5, "gui"),
    )
    for filename, lvl, size, count, component in handler_specs:
        if component is not None and mode != component:
            continue
        _add_rotating_handler(
            log_dir / filename, level=lvl, max_bytes=size, backup_count=count,
            formatter=RedactingFormatter(_LOG_FORMAT),
            log_filter=_ComponentFilter(COMPONENT_PREFIXES[component]) if component else None,
        )

    if _WINDOWS_CLH_FALLBACK and not _fallback_warned:
        # One-shot, and the file handlers above are already live, so this lands
        # in errors.log/agent.log — the fallback must never be invisible again.
        _fallback_warned = True
        logging.getLogger("hermes_logging").warning(
            "concurrent-log-handler unavailable on this Windows install (%s); "
            "file logging fell back to stdlib rotation without rollover.",
            _WINDOWS_CLH_FALLBACK_REASON or "portalocker probe failed",
        )

    if _logging_initialized and not force:
        return log_dir

    # Root level must be low enough for the handlers to fire.
    if root.level == logging.NOTSET or root.level > level:
        root.setLevel(level)
    _quiet_noisy_loggers()
    _logging_initialized = True
    return log_dir


def setup_verbose_logging() -> None:
    """Enable DEBUG-level console logging for ``--verbose`` / ``-v`` mode."""
    from agent.redact import RedactingFormatter

    root = logging.getLogger()

    if any(getattr(h, "_hermes_verbose", False) for h in root.handlers):
        return
    handler = logging.StreamHandler(_safe_stderr())
    handler.setLevel(logging.DEBUG)
    handler.setFormatter(RedactingFormatter(_LOG_FORMAT_VERBOSE, datefmt="%H:%M:%S"))
    handler._hermes_verbose = True  # type: ignore[attr-defined]
    root.addHandler(handler)

    if root.level > logging.DEBUG:
        root.setLevel(logging.DEBUG)
    _quiet_noisy_loggers()
    # rex-deploy at INFO for sandbox status.
    logging.getLogger("rex-deploy").setLevel(logging.INFO)


def _quietly(fn) -> None:
    """Call *fn* (a ``close``/``stop`` bound method) swallowing errors — teardown must never raise."""
    try:
        fn()
    except Exception:
        pass


class _ManagedRotatingFileHandler(RotatingFileHandler):
    """RotatingFileHandler with managed-mode perms and external-rotation detection.

    In managed mode (NixOS) the setgid stateDir needs group-readable files, but
    ``_open()``/``doRollover()`` honor the umask (0644), so ``chmod 0660`` follows both.
    A rotating handler also holds an fd: if the file is rotated externally (logrotate,
    ``mv``) writes silently go to the old inode, so each emit compares the path's inode
    to the open stream's and reopens on mismatch (the ``WatchedFileHandler`` pattern).
    """

    def __init__(self, *args, **kwargs):
        from hermes_cli.config import is_managed
        self._managed = is_managed()
        self._unavailable_reported = False
        # Set by _ProfileRoutingFileHandler: a vanished logs/ dir means the profile was deleted,
        # so skip the write (the router re-routes the record) instead of a FileNotFoundError traceback.
        self._skip_write_when_dir_missing = False
        self.skipped_missing_dir = False
        super().__init__(*args, **kwargs)
        self._record_stream_stat()

    def _chmod_if_managed(self):
        if self._managed:
            try:
                os.chmod(self.baseFilename, 0o660)
            except OSError:
                pass

    def _record_stream_stat(self, st: Optional[os.stat_result] = None) -> None:
        """Snapshot dev/ino of ``baseFilename`` so emit() can detect external rotation."""
        try:
            st = st or os.stat(self.baseFilename)
            self._stat_dev, self._stat_ino = st.st_dev, st.st_ino
        except OSError:
            self._stat_dev, self._stat_ino = None, None

    def _reopen_stream(self, stat_result=None) -> None:
        """Close and reopen ``baseFilename`` (best-effort).

        On failure the stream is left ``None`` so the next emit bails rather than
        writing to a stale inode.
        """
        if self.stream is not None:
            _quietly(self.stream.close)
        self.stream = None  # type: ignore[assignment]
        try:
            self.stream = self._open()
        except Exception:
            return
        self._record_stream_stat(stat_result)

    def _reopen_if_externally_rotated(self) -> None:
        """Reopen when ``baseFilename`` was renamed, unlinked, or replaced by another inode.

        Silent + best-effort: any error falls back to the existing (possibly stale)
        stream so logging keeps working instead of dying on a stat failure.
        """
        try:
            st = os.stat(self.baseFilename)
        except FileNotFoundError:
            self._reopen_stream()  # rotated/unlinked underneath us: recreate at the path
            return
        except OSError:
            return  # transient — try again on the next emit

        if self._stat_dev is None or self._stat_ino is None:
            self._record_stream_stat(st)
        elif (st.st_dev, st.st_ino) != (self._stat_dev, self._stat_ino):
            self._reopen_stream(st)

    def emit(self, record: logging.LogRecord) -> None:
        self.skipped_missing_dir = False
        had_stream = self.stream is not None
        # The kernel caches inode metadata, so this stat is sub-microsecond on a hot file.
        file_exists = had_stream or os.path.exists(self.baseFilename)
        if file_exists:
            self._reopen_if_externally_rotated()
        # Only when the file is gone (or its reopen just failed), so a healthy emit pays no extra
        # stat. ``stream is None`` alone is not a signal: on Windows concurrent-log-handler closes
        # the stream after every write.
        if (self.stream is None and (had_stream or not file_exists) and self._skip_write_when_dir_missing
                and not os.path.isdir(os.path.dirname(self.baseFilename))):
            self.skipped_missing_dir = True
            return
        super().emit(record)
        # A record actually reached the file: only now has the destination recovered. Resetting
        # in _open() is wrong — open() succeeds on a device whose write/flush still raise EIO,
        # which re-armed the report and printed the path once per record.
        if self.stream is not None:
            self._unavailable_reported = False

    def handleError(self, record: logging.LogRecord) -> None:
        """Suppress the known Windows ``concurrent-log-handler`` lock timeout.

        CLH's ``emit()`` routes that RuntimeError here, so this is the single point to
        silence it before stdlib prints to stderr (which the Desktop slash-worker
        captures into chat output). Silencing is not silent: warn once through the
        logging system so a wedged lock is visible in the logs instead of a black hole.
        """
        exc = sys.exc_info()[1]
        if _is_windows_concurrent_log_lock_timeout(exc):
            _warn_windows_lock_timeout_once()
            return
        if _is_unavailable_log_stream(exc):
            # The QueueListener must not turn a failing log destination into a traceback for
            # every queued record. Name the path once, drop the stale stream; the next emit
            # reopens it if the destination has recovered.
            if not self._unavailable_reported:
                self._unavailable_reported = True
                _quietly(lambda: print(
                    f"hermes_logging: {self.baseFilename} unavailable ({exc}); "
                    "file logging paused until it recovers", file=_safe_stderr()))
            if self.stream is not None:
                _quietly(self.stream.close)
            self.stream = None  # type: ignore[assignment]
            return
        super().handleError(record)

    def _open(self):
        stream = super()._open()
        self._chmod_if_managed()
        return stream

    def doRollover(self):
        # The stdlib rollover opens a fresh baseFilename owned by whichever process crossed
        # maxBytes. With one rotating handler per profile that is usually the long-lived root
        # gateway, and a worker on another uid can never reopen its own agent.log (#120151).
        # Only root can hand the file back; an unprivileged process never changed the owner.
        try:
            previous = os.stat(self.baseFilename)
        except OSError:
            previous = None
        super().doRollover()
        if previous is not None:
            try:
                if getattr(os, "geteuid", lambda: -1)() == 0:
                    os.chown(self.baseFilename, previous.st_uid, previous.st_gid)
                os.chmod(self.baseFilename, previous.st_mode & 0o7777)
            except OSError:
                pass  # a log that cannot be chowned is still a working log
        self._chmod_if_managed()
        # Our own rollover writes a new baseFilename; refresh the snapshot so
        # the next emit doesn't mistake it for external rotation.
        self._record_stream_stat()


def _new_file_handler(
    path: Path, *, level: int, max_bytes: int, backup_count: int, formatter
) -> "_ManagedRotatingFileHandler":
    """Create the ``logs/`` directory and a configured ``_ManagedRotatingFileHandler``."""
    mkdir_under_hermes_home(path.parent)
    if _WINDOWS_CLH_FALLBACK:
        # stdlib fallback: no rollover, or the file pins at the size threshold
        # and every emit re-triggers the WinError 32 rename failure (#44873).
        max_bytes, backup_count = 0, 0
    handler = _ManagedRotatingFileHandler(
        str(path), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"
    )
    handler.setLevel(level)
    handler.setFormatter(formatter)
    return handler


# A routed profile home is re-checked for an out-of-band delete (missing dir or tombstone) at
# most this often. WHY: the check is two stats per record per router on the listener thread,
# nearly doubling the syscalls of every live-profile emit; a deleted home only needs catching
# within a couple of seconds. Inside that window a tombstoned (not yet removed) home still gets
# its records; an rmtree is caught at once by the write that finds the directory gone.
_PROFILE_LIVENESS_RECHECK_S = 2.0


class _ProfileRoutingFileHandler(logging.Handler):
    """Route queued records to the log file for their Hermes home.

    Used only behind the QueueListener, so its small routing lock never blocks an agent
    or dashboard event loop. Per-home handlers keep rotation, redaction and managed perms.
    """

    def __init__(self, existing: RotatingFileHandler, profile_homes: Sequence[Path]) -> None:
        """Take over *existing*'s path, level, rotation, formatter and filters."""
        super().__init__(level=existing.level)
        resolved = Path(existing.baseFilename).resolve()
        self.baseFilename = str(resolved)
        self._hermes_routed_log_path = resolved
        self._default_home = resolved.parent.parent.resolve()
        self._profile_homes = {Path(home).expanduser().resolve() for home in profile_homes}
        self._filename = resolved.name
        self._max_bytes = getattr(existing, "maxBytes", 0)
        self._backup_count = getattr(existing, "backupCount", 0)
        self._profile_handlers: dict[Path, _ManagedRotatingFileHandler] = {}
        self._profile_handlers_lock = threading.RLock()
        # Home -> monotonic time of its last liveness check (see _PROFILE_LIVENESS_RECHECK_S).
        self._liveness_checked_at: dict[Path, float] = {}
        self.setFormatter(existing.formatter)
        for log_filter in existing.filters:
            self.addFilter(log_filter)

    def _home_for_record(self, record: logging.LogRecord) -> Path:
        raw_home = getattr(record, "hermes_home", "")
        try:
            candidate = Path(raw_home).expanduser().resolve()
        except (TypeError, ValueError, OSError):
            candidate = self._default_home
        if candidate == self._default_home or candidate not in self._profile_homes:
            return self._default_home
        last = self._liveness_checked_at.get(candidate)
        if last is None or _monotonic() - last >= _PROFILE_LIVENESS_RECHECK_S:
            # First sighting always checks, so a home deleted before it ever logged is caught on
            # its first record; afterwards at most once per interval. An rmtree inside the
            # interval is caught by emit() on the record whose write finds the dir gone.
            if self._release_if_deleted(candidate):
                return self._default_home
        return candidate

    def _release_if_deleted(self, home: Path) -> bool:
        """Release *home* if it was deleted out of band; True when it was."""
        self._liveness_checked_at[home] = _monotonic()
        if home.is_dir() and not named_profile_is_deleted(home):
            return False
        # Deleted out of band (CLI ``hermes profile delete`` while this process runs; it
        # tombstones before rmtree). The startup snapshot still names it, so its handler would
        # retry the vanished logs/ path on every record (#103777). Release it the way an
        # in-process delete does: the stale fd is closed on this routed record and the home is
        # dropped from the routing set, so later records fall back to the default home. This
        # does not reach into another process: a CLI delete racing a live server's open fd
        # still relies on release_profile_log_handlers in-process and #130285.
        self.release_profile(home)
        return True

    def _handler_for_home(self, home: Path) -> _ManagedRotatingFileHandler:
        with self._profile_handlers_lock:
            if home not in self._profile_handlers:
                handler = _new_file_handler(
                    home / "logs" / self._filename, level=self.level, max_bytes=self._max_bytes,
                    backup_count=self._backup_count, formatter=self.formatter,
                )
                handler._skip_write_when_dir_missing = home != self._default_home
                self._profile_handlers[home] = handler
            return self._profile_handlers[home]

    def emit(self, record: logging.LogRecord) -> None:
        try:
            home = self._home_for_record(record)
            handler = self._handler_for_home(home)
            if home == self._default_home:
                handler.handle(record)
                return
            # Formatted here, on the listener thread, where the record's profile scope is gone: bind its home so
            # RedactingFormatter applies THAT profile's redact_secrets policy and vault values, not the launch's.
            from hermes_constants import reset_hermes_home_override, set_hermes_home_override
            token = set_hermes_home_override(str(home))
            try:
                handler.handle(record)
            finally:
                reset_hermes_home_override(token)
            # The write was skipped because logs/ is gone (an rmtree inside the liveness interval,
            # see _ManagedRotatingFileHandler.emit): re-check now, and keep the record in the
            # default log either way (a live profile whose logs/ was removed by hand included).
            if handler.skipped_missing_dir:
                self._release_if_deleted(home)
                self._handler_for_home(self._default_home).handle(record)
        except Exception:
            self.handleError(record)

    def close(self) -> None:
        with self._profile_handlers_lock:
            handlers = list(self._profile_handlers.values())
            self._profile_handlers.clear()
        for handler in handlers:
            _quietly(handler.close)
        super().close()

    def release_profile(self, home: Path) -> bool:
        """Close and forget the routed files belonging to a deleted profile."""
        with self._profile_handlers_lock:
            handler = self._profile_handlers.pop(home, None)
            self._profile_homes.discard(home)
            self._liveness_checked_at.pop(home, None)
        if handler is None:
            return False
        _quietly(handler.close)
        return True


# Asynchronous file logging: an ``emit`` can block on the cross-process
# rotation lock (module header); on an asyncio thread that stalls the loop and
# drops WebSocket clients. Every file handler is therefore driven by a single
# QueueListener thread; loggers only do a non-blocking enqueue.

_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 race a flush/reset from
# another thread (gateway init vs a plugin/CLI path); without this, two
# threads can interleave stop()/reassign/start() and leave two live listeners.
_queue_state_lock = threading.Lock()


class _NonFormattingQueueHandler(QueueHandler):
    """``QueueHandler`` for an in-process queue.

    Stdlib ``prepare()`` formats and strips ``args``/``exc_info`` for cross-process
    pickling; ours is in-process, so targets get the unformatted record and apply their
    own ``RedactingFormatter`` on the listener thread. A shallow copy is returned because
    the emitting thread's synchronous handlers may mutate ``record.message`` meanwhile.
    """

    def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
        return copy.copy(record)


def _stop_queue_listener() -> None:
    """Flush and stop the background log listener (idempotent; atexit hook, so it takes the lock)."""
    global _queue_listener
    with _queue_state_lock:
        listener, _queue_listener = _queue_listener, None
        if listener is not None:
            _quietly(listener.stop)


def _start_queue_listener_locked() -> None:
    """(Re)build + start a listener over the current handler set (``_queue_state_lock`` held).

    A running listener is stopped first; this only happens while handlers are being
    added (queue empty), so ``stop()`` returns immediately.
    """
    global _queue_listener
    if _queue_listener is not None:
        _queue_listener.stop()
    _queue_listener = QueueListener(_log_queue, *_queued_file_handlers, respect_handler_level=True)
    _queue_listener.start()


def _register_queued_handler(handler: logging.Handler) -> None:
    """Route *handler* through the shared async queue instead of attaching it to root.

    Emitting threads never block on file I/O or the rotation lock; the ``QueueListener``
    applies each handler's own level and filters on its worker thread.
    """
    global _log_queue, _queue_atexit_registered
    with _queue_state_lock:
        if _log_queue is None:
            _log_queue = queue.SimpleQueue()
            qh = _NonFormattingQueueHandler(_log_queue)
            qh._hermes_queue = True  # type: ignore[attr-defined]
            # Always on the root logger so records from any logger reach the queue.
            logging.getLogger().addHandler(qh)
        _queued_file_handlers.append(handler)
        _start_queue_listener_locked()
        if not _queue_atexit_registered:
            # Runs before logging.shutdown (registered earlier at import time),
            # so the listener stops before its file handlers are closed.
            atexit.register(_stop_queue_listener)
            _queue_atexit_registered = True


def flush_log_queue() -> None:
    """Block until all queued records have been written, then resume.

    Stops the listener (which processes every pending record before joining) and
    restarts it. ``stop()`` joins the worker thread — do NOT call this on a hard-exit
    path where the listener may be wedged on the rotation lock; use
    ``drain_log_queue()`` there, 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).

    If the listener's worker is wedged on the cross-process rotation lock — the very
    failure async logging exists to survive — an unbounded join would re-freeze shutdown.
    """
    listener = _queue_listener
    if listener is None:
        return
    t = threading.Thread(target=lambda: _quietly(listener.stop), name="hermes-log-drain", daemon=True)
    t.start()
    t.join(timeout)


def release_profile_log_handlers(profile_home: str | Path) -> int:
    """Release this process's routed log files below a profile before it is removed.

    The Desktop serve process can route records for several profiles through one
    ``QueueListener``. On Windows, each routed concurrent log handler keeps its
    lock file open, so closing only external profile resources still leaves
    ``logs/.__agent.lock`` and ``logs/.__errors.lock`` unavailable to rmtree.
    """
    global _queue_listener
    try:
        home = Path(profile_home).expanduser().resolve()
    except (TypeError, ValueError, OSError):
        return 0

    with _queue_state_lock:
        listener = _queue_listener
        if listener is not None:
            listener.stop()
            _queue_listener = None
        released = sum(
            handler.release_profile(home)
            for handler in _queued_file_handlers
            if isinstance(handler, _ProfileRoutingFileHandler)
        )
        if listener is not None:
            _start_queue_listener_locked()
    return released


def enable_profile_log_routing(profile_homes: Sequence[str | Path]) -> bool:
    """Make the queued file logs follow a desktop profile context.

    ``setup_logging`` binds handlers to one process home; the desktop dashboard's
    embedded cron ticker may run jobs for every profile, so its static file handlers
    are replaced with profile routers once the profile list is known. Returns ``True``
    when routing is (or already was) enabled; a single-profile caller is left untouched.
    """
    global _queue_listener
    homes: list[Path] = []
    for entry in profile_homes:
        try:
            resolved = Path(entry[1] if isinstance(entry, tuple) else entry).expanduser().resolve()
        except (TypeError, ValueError, OSError):
            continue
        if resolved not in homes:
            homes.append(resolved)
    if len(homes) < 2:
        return False

    with _queue_state_lock:
        if not _queued_file_handlers:
            return False
        routers = [h for h in _queued_file_handlers if isinstance(h, _ProfileRoutingFileHandler)]
        if routers:
            for handler in routers:
                with handler._profile_handlers_lock:
                    handler._profile_homes = handler._profile_homes.union(homes)
            return True
        listener = _queue_listener
        if listener is not None:
            listener.stop()
            _queue_listener = None
        replacement = []
        for existing in _queued_file_handlers:
            if isinstance(existing, RotatingFileHandler):
                replacement.append(_ProfileRoutingFileHandler(existing, homes))
                _quietly(existing.close)
            else:
                replacement.append(existing)
        _queued_file_handlers[:] = replacement
        if listener is not None:
            _start_queue_listener_locked()
        return True


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()
        for h in list(root.handlers):
            if getattr(h, "_hermes_queue", False):
                root.removeHandler(h)
        for h in list(_queued_file_handlers):
            _quietly(h.close)
        _queued_file_handlers.clear()
        _log_queue = None


def _add_rotating_handler(
    path: Path,
    *,
    level: int,
    max_bytes: int,
    backup_count: int,
    formatter: logging.Formatter,
    log_filter: Optional[logging.Filter] = None,
) -> None:
    """Register a queued ``RotatingFileHandler`` for *path*; idempotent per resolved path."""
    resolved = path.resolve()
    for existing in _queued_file_handlers:
        # Already attached directly, or already covered by the profile router — for its default
        # home or any profile home it routes (a bare handler beside it would take every record).
        if getattr(existing, "_hermes_routed_log_path", None) == resolved or (
            isinstance(existing, RotatingFileHandler)
            and Path(getattr(existing, "baseFilename", "")).resolve() == resolved
        ):
            return
        if isinstance(existing, _ProfileRoutingFileHandler) and existing._filename == resolved.name and (
            resolved.parent.parent == existing._default_home or resolved.parent.parent in existing._profile_homes
        ):
            return
    handler = _new_file_handler(
        path, level=level, max_bytes=max_bytes, backup_count=backup_count, formatter=formatter,
    )
    if log_filter is not None:
        handler.addFilter(log_filter)
    # Routing already on (a second home adopted earlier): a component log added now —
    # ``mode="gateway"`` after the fact — must route too, or it takes every home's records.
    routers = [h for h in _queued_file_handlers if isinstance(h, _ProfileRoutingFileHandler)]
    if routers:
        homes: set[Path] = set()
        for router in routers:
            homes.add(router._default_home)
            homes.update(router._profile_homes)
        routed = _ProfileRoutingFileHandler(handler, sorted(homes))
        _quietly(handler.close)
        handler = routed
    # Queue, not ``addHandler``: the rotation-lock wait never runs on the caller's thread.
    _register_queued_handler(handler)


def _read_logging_config():
    """Best-effort read of ``logging.*`` from config.yaml."""
    try:
        # Prefer the shared effective-config cache (managed overlay included, so an administrator
        # can pin logging.*) so this reuses hermes_cli.main's early parse (one config.yaml parse
        # per process); fall back to a direct parse for bare hermes_logging consumers.
        try:
            from hermes_cli.config_effective import load_user_config_effective
            cfg = load_user_config_effective(get_config_path())
        except Exception:
            from utils import fast_safe_load
            config_path = get_config_path()
            cfg = {}
            if config_path.exists():
                with open(config_path, "r", encoding="utf-8-sig") as f:
                    cfg = fast_safe_load(f) or {}
        if not cfg:
            return (None, None, None)
        log_cfg = cfg.get("logging", {})
        if isinstance(log_cfg, dict):
            return (log_cfg.get("level"), log_cfg.get("max_size_mb"), log_cfg.get("backup_count"))
    except Exception:
        pass
    return (None, None, None)