""" State Directory Service (R11). Provides portable, safe state directory paths for logs, caches, and persistence. """ import logging import os import sys from typing import Optional logger = logging.getLogger("ComfyUI-OpenClaw.services.state_dir") # Environment variables to override state directory STATE_DIR_ENV = "OPENCLAW_STATE_DIR" LEGACY_STATE_DIR_ENV = "MOLTBOT_STATE_DIR" # Default subdirectory names under user data STATE_DIR_NAME = "comfyui-openclaw" LEGACY_STATE_DIR_NAME = "comfyui-moltbot" def _get_user_data_dir(subdir: Optional[str] = None) -> str: """ Get the platform-appropriate user data directory. Returns: Path to user-writable application data directory. """ subdir = subdir or STATE_DIR_NAME if sys.platform == "win32": # Windows: %LOCALAPPDATA%\{subdir} base = os.environ.get("LOCALAPPDATA", os.path.expanduser("~")) return os.path.join(base, subdir) elif sys.platform == "darwin": # macOS: ~/Library/Application Support/{subdir} return os.path.join( os.path.expanduser("~"), "Library", "Application Support", subdir ) else: # Linux/Unix: ~/.local/share/{subdir} (XDG_DATA_HOME) base = os.environ.get( "XDG_DATA_HOME", os.path.join(os.path.expanduser("~"), ".local", "share") ) return os.path.join(base, subdir) def _resolve_state_dir_path() -> str: """Resolve the canonical state-dir path without creating it.""" env_dir = os.environ.get(STATE_DIR_ENV) or os.environ.get(LEGACY_STATE_DIR_ENV) if env_dir: return os.path.abspath(env_dir) new_dir = _get_user_data_dir(STATE_DIR_NAME) legacy_dir = _get_user_data_dir(LEGACY_STATE_DIR_NAME) return ( legacy_dir if (os.path.exists(legacy_dir) and not os.path.exists(new_dir)) else new_dir ) def peek_state_dir() -> str: """Return the canonical state-dir path without creating directories.""" return _resolve_state_dir_path() def get_state_dir() -> str: """ Get the canonical state directory for all writable data. Priority: 1. OPENCLAW_STATE_DIR environment variable (explicit override) 2. (legacy) MOLTBOT_STATE_DIR environment variable (explicit override) 2. Platform-appropriate user data directory The directory is created if it doesn't exist. Returns: Absolute path to state directory. """ state_dir = _resolve_state_dir_path() # Ensure directory exists with appropriate permissions if not os.path.exists(state_dir): try: os.makedirs(state_dir, mode=0o700, exist_ok=True) logger.info(f"Created state directory: {state_dir}") except OSError as e: logger.error(f"Failed to create state directory: {e}") # Fallback to temp directory import tempfile state_dir = os.path.join(tempfile.gettempdir(), STATE_DIR_NAME) os.makedirs(state_dir, mode=0o700, exist_ok=True) logger.warning(f"Using fallback state directory: {state_dir}") return state_dir def peek_log_path() -> str: """Return the canonical log-file path without creating directories.""" state_dir = peek_state_dir() new_path = os.path.join(state_dir, "openclaw.log") legacy_path = os.path.join(state_dir, "moltbot.log") return ( legacy_path if (os.path.exists(legacy_path) and not os.path.exists(new_path)) else new_path ) def get_log_path() -> str: """Get the path for the log file.""" state_dir = get_state_dir() new_path = os.path.join(state_dir, "openclaw.log") legacy_path = os.path.join(state_dir, "moltbot.log") return ( legacy_path if (os.path.exists(legacy_path) and not os.path.exists(new_path)) else new_path ) def get_cache_dir() -> str: """Get the path for cache files.""" cache_dir = os.path.join(get_state_dir(), "cache") os.makedirs(cache_dir, mode=0o700, exist_ok=True) return cache_dir def get_db_path() -> str: """Get the path for the persistence database (future use).""" state_dir = get_state_dir() new_path = os.path.join(state_dir, "openclaw.db") legacy_path = os.path.join(state_dir, "moltbot.db") return ( legacy_path if (os.path.exists(legacy_path) and not os.path.exists(new_path)) else new_path ) # Module-level convenience for backwards compatibility STATE_DIR = None def ensure_state_dir() -> str: """ Ensure state directory exists and return its path. Caches the result for performance. """ global STATE_DIR if STATE_DIR is None: STATE_DIR = get_state_dir() return STATE_DIR