Class Logger

src/mcp_client/logger.py:28–957

class Logger:
    # Match scripts/dashboard.py palette
    _PAL_ACCENT = "#8bb5ff"
    _PAL_GREEN = "#8fd19e"
    _PAL_YELLOW = "#e3c07b"
    _PAL_RED = "#f28b82"
    _PAL_ORANGE = "#ffb072"
    _PAL_PURPLE = "#c9a0ff"
    _PAL_CYAN = "#7bd9ff"

    def __init__(
        self,
        mcp_client: MCPClient,
        log_file: str,
        to_stdout: bool = True,
        *,
        host_log_path: Path | None = None,
        event_log_path: str | None = None,
        host_event_log_path: Path | None = None,
        deduplicate: bool = False,
        append_mode: bool = False,
    ):
        """
        Initialize the logger.

        Args:
            log_file (str): Path to the log file.
            to_stdout (bool): If True, also print to stdout.
            host_log_path (Path | None): Optional local path to write logs outside MCP.
            event_log_path (str | None): Optional log path for structured events.
            host_event_log_path (Path | None): Optional local path for structured events.
            deduplicate (bool): If True, skip consecutive duplicate messages.
            append_mode (bool): If True, append to existing log files instead of
                overwriting them. Useful when re-creating the logger during resume
                to preserve previously logged events.
        """
        self.mcp_client = mcp_client
        self.log_path = log_file  # path under MCP fs_write
        self.to_stdout = to_stdout
        self.host_log_path = Path(host_log_path) if host_log_path else None
        self.event_log_path = event_log_path or self.log_path
        self.host_event_log_path = (
            Path(host_event_log_path) if host_event_log_path else self.host_log_path
        )
        self.deduplicate = deduplicate
        self.mcp_available = True
        self._last_message = None

        # Thread lock for thread-safe logging in parallel execution
        self._lock = threading.Lock()

        # Pretty console rendering (optional)
        self.pretty = (
            self._env_bool("CS_PRETTY", default=True)
            and self.to_stdout
            and Console is not None
        )
        self._console = (