workspace_path.py

"""Workspace path detection mirroring src/utils/workspace-path.ts"""

from __future__ import annotations

import os
import re
import sys
import subprocess
import threading

from .path_helpers import expand_tilde_path

# Module-level override set via POST /api/set-workspace.
# CLI ``--base-dir`` passes resolve_workspace_path(override=...) instead.
# Reads and writes are serialized by _workspace_path_lock so threaded WSGI
# workers (gunicorn --threads, waitress, etc.) always see the latest override
# from another thread and resolve_workspace_path's snapshot+expand stays consistent.
_workspace_path_lock = threading.Lock()
_workspace_path_override: str | None = None


def set_workspace_path_override(path: str | None) -> None:
    """Set the thread-safe module-level workspace path override.

    Args:
        path: Canonical workspace storage root from POST /api/set-workspace, or
            ``None`` to clear the override.
    """
    global _workspace_path_override
    with _workspace_path_lock:
        _workspace_path_override = path


def get_workspace_path_override() -> str | None:
    """Return the current module-level workspace path override, if any.

    Returns:
        Canonical path set via :func:`set_workspace_path_override`, else ``None``.
    """
    with _workspace_path_lock:
        return _workspace_path_override


def is_multi_worker_process_deployment() -> bool:
    """Return True when multiple WSGI worker processes may serve requests.

    Used by POST /api/set-workspace to fail loud instead of updating only one
    process. Single-process and multi-threaded deployments return False.

    Operators can force True/False with ``CURSOR_BROWSER_MULTI_WORKER``. Otherwise
    ``WEB_CONCURRENCY``, ``GUNICORN_WORKERS``, or ``--workers`` / ``-w`` in
    ``GUNICORN_CMD_ARGS`` are consulted when present.
    """
    flag = os.environ.get("CURSOR_BROWSER_MULTI_WORKER", "").strip().lower()
    if flag in ("1", "true", "yes"):
        return True
    if flag in ("0", "false", "no"):
        return False
    for key in ("WEB_CONCURRENCY", "GUNICORN_WORKERS"):
        raw = os.environ.get(key, "").strip()
        if raw.isdigit():
            count = int(raw)
            if count > 1:
                return True
    cmd_args = os.environ.get("GUNICORN_CMD_ARGS", "")
    if cmd_args:
        for pattern in (
            r"(?:--workers|-w)\s+(\d+)",
            r"(?:--workers|-w)=(\d+)",
        ):
            for match in re.finditer(pattern, cmd_args):
                if int(match.group(1)) > 1:
                    return True
    return False


def get_default_workspace_path() -> str:
    """Detect the default Cursor workspace storage path based on OS."""
    home = os.path.expanduser("~")
    release = os.uname().release.lower() if hasattr(os, "uname") else ""
    is_wsl = "microsoft" in release or "wsl" in release
    is_remote = bool(
        os.environ.get("SSH_CONNECTION")
        or os.environ.get("SSH_CLIENT")
        or os.environ.get("SSH_TTY")
    )

    if is_wsl:
        username = os.getenv("USER", "")
        try:
            output = subprocess.check_output(
                ["cmd.exe", "/c", "echo", "%USERNAME%"],
                text=True,
                stderr=subprocess.DEVNULL,
            )
            username = output.strip()
        except Exception:
            pass
        return f"/mnt/c/Users/{username}/AppData/Roaming/Cursor/User/workspaceStorage"

    if sys.platform == "win32":
        return os.path.join(home, "AppData", "Roaming", "Cursor", "User", "workspaceStorage")
    elif sys.platform == "darwin":
        return os.path.join(home, "Library", "Application Support", "Cursor", "User", "workspaceStorage")
    elif sys.platform == "linux":
        if is_remote:
            return os.path.join(home, ".cursor-server", "data", "User", "workspaceStorage")
        return os.path.join(home, ".config", "Cursor", "User", "workspaceStorage")
    else:
        return os.path.join(home, "workspaceStorage")


def resolve_workspace_path(*, override: str | None = None) -> str:
    """Return the effective workspace path (call override > module > env > default).

    *override* is for one-shot callers (e.g. CLI ``--base-dir``) and does not
    mutate ``WORKSPACE_PATH``. Module override comes from POST /api/set-workspace
    (validated). ``WORKSPACE_PATH`` is only tilde-expanded — trusted-operator
    escape hatch, not the same checks as the API (issue #15).
    """
    if override:
        return expand_tilde_path(override)
    with _workspace_path_lock:
        module_override = _workspace_path_override
    if module_override:
        return expand_tilde_path(module_override)
    env_path = os.environ.get("WORKSPACE_PATH", "").strip()
    if env_path:
        return expand_tilde_path(env_path)
    return get_default_workspace_path()


def get_cli_chats_path() -> str:
    """Return the Cursor CLI chats directory (~/.cursor/chats).

    This is where the ``agent`` CLI stores chat sessions, independent of
    platform and completely separate from the IDE workspace storage.

    Override with the ``CLI_CHATS_PATH`` environment variable (useful in tests).
    """
    env_path = os.environ.get("CLI_CHATS_PATH", "").strip()
    if env_path:
        return expand_tilde_path(env_path)
    return os.path.join(os.path.expanduser("~"), ".cursor", "chats")