hermes-webui/api/workspace.py

"""
Hermes Web UI -- Workspace and file system helpers.

Workspace lists and last-used workspace are stored per-profile so each
profile has its own workspace configuration.  State files live at
``{profile_home}/webui_state/workspaces.json`` and
``{profile_home}/webui_state/last_workspace.txt``.  The global STATE_DIR
paths are used as fallback when no profile module is available.
"""
import hashlib
import json
import logging
import os
import posixpath
import re
import secrets
import shutil
import stat
import subprocess
import sys
import concurrent.futures
import contextlib
import contextvars
import threading
import time
from collections.abc import Callable
from pathlib import Path, PurePosixPath

logger = logging.getLogger(__name__)

_ESCAPE_AUTH_TTL_SECONDS = 300
_ESCAPE_AUTH_LOCK = threading.Lock()
_ESCAPE_AUTH_TOKENS: dict[str, dict[str, str | int | float]] = {}

from api.config import (
    WORKSPACES_FILE as _GLOBAL_WS_FILE,
    LAST_WORKSPACE_FILE as _GLOBAL_LW_FILE,
    DEFAULT_WORKSPACE as _BOOT_DEFAULT_WORKSPACE,
    MAX_FILE_BYTES, IMAGE_EXTS, MD_EXTS
)
from api.subprocess_utils import windows_hide_flags


# ── Profile-aware path resolution ───────────────────────────────────────────

# Logical profile-name grammar — mirrors api.profiles._PROFILE_ID_RE. Kept as
# a local copy so workspace-layer validation does not import profiles at module
# load time (profiles may not be importable in every embedding context).
_PROFILE_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9_-]{0,63}$')


def _profile_state_dir(profile: str | Path | None = None) -> Path:
    """Return the webui_state directory for the active or given profile.

    For the default profile, returns the global STATE_DIR (respects
    HERMES_WEBUI_STATE_DIR env var for test isolation).
    For named profiles, returns {profile_home}/webui_state/.
    """
    try:
        from api.profiles import get_active_profile_name, get_active_hermes_home
        if profile is not None:
            # Literal-"default" STATE routing (#7168 re-gate round 7): the
            # default profile's workspace state always lives in the global
            # state files, even when isolated mode pins the default home at
            # <base>/profiles/default. The round-6 resolver change made
            # _resolve_profile_home_param("default") return that pinned home,
            # so the canonical-home check below sent explicit
            # profile="default" state reads/writes to {pinned}/webui_state/
            # while ambient calls kept using the global dir — splitting saved
            # workspaces between two authorities. Config and workspace PATH
            # resolution still use the pinned home via
            # _resolve_profile_home_param; only this state-file tier stays
            # global for the logical string "default".
            if isinstance(profile, str) and profile.strip() == 'default':
                return _GLOBAL_WS_FILE.parent
            profile_home = _resolve_profile_home_param(profile)
            if not _is_default_profile_home(profile_home):
                d = profile_home / 'webui_state'
                d.mkdir(parents=True, exist_ok=True)
                return d
            return _GLOBAL_WS_FILE.parent

        name = get_active_profile_name()
        if name and name != 'default':
            d = get_active_hermes_home() / 'webui_state'
            d.mkdir(parents=True, exist_ok=True)
            return d
    except ImportError:
        logger.debug("Failed to import profiles module, using global state dir")
    return _GLOBAL_WS_FILE.parent


def _workspaces_file(profile: str | Path | None = None) -> Path:
    """Return the workspaces.json path for the active or given profile."""
    return _profile_state_dir(profile=profile) / 'workspaces.json'


def _last_workspace_file(profile: str | Path | None = None) -> Path:
    """Return the last_workspace.txt path for the active or given profile."""
    return _profile_state_dir(profile=profile) / 'last_workspace.txt'


def _workspaces_file_for_profile(profile: str | Path | None = None) -> Path | None:
    """Profile-scoped workspaces.json path, or None for an INVALID profile.

    ``None`` is the fail-closed contract (#7168 re-gate round 4): a malformed
    profile name must not be clamped onto the default/global state files, so
    callers treat it as "no readable/writable profile-local state".
    """
    try:
        return _workspaces_file(profile=profile) if profile is not None else _workspaces_file()
    except TypeError:
        return _workspaces_file()
    except ValueError:
        logger.debug("Ignoring invalid profile name %r for workspaces file", profile)
        return None


def _last_workspace_file_for_profile(profile: str | Path | None = None) -> Path | None:
    """Profile-scoped last_workspace.txt path, or None for an INVALID profile."""
    try:
        return _last_workspace_file(profile=profile) if profile is not None else _last_workspace_file()
    except TypeError:
        return _last_workspace_file()
    except ValueError:
        logger.debug("Ignoring invalid profile name %r for last-workspace file", profile)
        return None


def _expanduser_path(path: str | Path) -> Path:
    """Return *path* after shell-style home expansion.

    ``Path.expanduser()`` on Windows does not consistently honor a monkeypatched
    ``HOME`` in tests, which makes host-native replay diverge from the repo's
    portability expectations. Use ``os.path.expanduser`` as the expansion source,
    then wrap it back into ``Path``.
    """
    raw = str(path)
    if raw.startswith('~'):
        home = (
            os.environ.get('HOME')
            or os.environ.get('USERPROFILE')
            or (
                (os.environ.get('HOMEDRIVE') or '') + (os.environ.get('HOMEPATH') or '')
            )
            or str(Path.home())
        )
        if raw in ('~', '~/', '~\\'):
            return Path(home)
        if raw.startswith('~/') or raw.startswith('~\\'):
            return Path(home) / raw[2:]
        # NOTE: ``~user`` / ``~root`` forms are intentionally NOT expanded here.
        # Master deliberately does not block ``/root`` (#510/#521 — Hermes commonly
        # runs as root, where ``/root`` is the legitimate home and is allowed via
        # the home carve-out). Expanding ``~root`` -> ``/root`` for a NON-root
        # deployment would let it register root's home; leaving the literal form
        # (resolved relative to cwd, then rejected if it escapes) is the safer
        # behavior and matches the current security model.
    return Path(raw)


def _resolve_path(path: str | Path, profile: str | Path | None = None) -> Path:
    """Resolve *path* after env-aware home expansion, preserving remote POSIX paths without raising."""
    remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile)
    if remote_candidate is not None:
        return remote_candidate
    return _safe_resolve(_expanduser_path(path))


def _home_path() -> Path:
    """Return the current effective home directory with env-aware expansion."""
    return _resolve_path("~")


def _as_posix_path(path: str | Path | None) -> PurePosixPath | None:
    if path in (None, ""):
        return None
    raw = _strip_surrounding_quotes(str(path)).strip().replace('\\', '/')
    # Reject embedded null bytes here rather than letting them survive normpath
    # and crash later at .resolve() with an uncaught ValueError (surfaces as a
    # 500). Fail-closed: treat as an invalid path.
    if '\x00' in raw:
        return None
    if not raw.startswith('/'):
        return None
    return PurePosixPath(posixpath.normpath(raw))


def _posix_is_within(path: PurePosixPath, root: PurePosixPath) -> bool:
    try:
        path.relative_to(root)
        return True
    except ValueError:
        return False


def _normalize_posix_path(path: str | Path | None) -> str | None:
    candidate = _as_posix_path(path)
    if candidate is None:
        return None
    return candidate.as_posix()


def _is_remote_terminal_backend(terminal_cfg: dict | None) -> bool:
    """Return True when the active terminal backend runs outside this WebUI host."""
    if not isinstance(terminal_cfg, dict):
        return False
    backend = str(terminal_cfg.get('backend') or '').strip().lower()
    return backend not in ('', 'local')


# Call-scoped cache for `_resolve_profile_home_param`'s filesystem
# `.resolve()` call. The webui's session list builds one `Session` per CLI
# session row (`_load_cli_sessions_uncached`), and `Session.__init__` calls
# `_resolve_profile_home_param(profile)` once per row -- with only one
# profile in play (the common case) that is the SAME path resolved hundreds
# of times per request. Individually a resolve() is ~1ms, but at a few
# hundred sessions this compounded into multi-second sidebar hangs under
# host load (confirmed via repeated "Slow WebUI request still running"
# warnings pinned to this call site).
#
# IMPORTANT: this is deliberately NOT a process-lifetime cache. An earlier
# revision of this fix cached forever, on the premise that a profile
# argument always resolves to the same path for the life of the process --
# but `Path.resolve()` is a filesystem call, not a pure function of
# startup constants: a profile-home symlink can be retargeted while the
# webui process keeps running for days/weeks, and `_safe_resolve()` below
# deliberately falls back to returning the UNRESOLVED input after a
# transient `OSError`/`RuntimeError`/`ValueError`, which a process-lifetime
# cache would then serve forever. Both are real correctness bugs (caught in
# PR #7636 review), not style nitpicks.
#
# Instead the cache is scoped to exactly ONE invocation of the hot loop
# (`_load_cli_sessions_uncached`, wrapped below in `profile_home_resolve_cache_scope`)
# via a `ContextVar` that holds a fresh dict only while that single
# sidebar-list build is running, and is `None` everywhere else. Every other
# call site -- and every request that isn't actively inside that one
# build -- resolves fresh every time, exactly like pre-PR behavior, so a
# symlink retarget or a transient resolve error is never masked past the
# single call that would have paid for the redundant resolves anyway.
_PROFILE_HOME_RESOLVE_SCOPE: "contextvars.ContextVar[dict[Path, Path] | None]" = (
    contextvars.ContextVar("_PROFILE_HOME_RESOLVE_SCOPE", default=None)
)


@contextlib.contextmanager
def profile_home_resolve_cache_scope():
    """Enable memoization of `_resolve_profile_home_param` for one call.

    Use as a decorator (or a `with` block) around the single hot-path
    operation that resolves the SAME profile argument hundreds of times in a
    tight loop -- currently `_load_cli_sessions_uncached` building the
    CLI/cron session list. While active, repeated resolves of the same
    profile argument are served from a dict that lives ONLY for the
    duration of the call. Outside of it -- including before/after, and any
    call site other than the one wrapped -- `_resolve_profile_home_param`
    resolves fresh every time, with no caching, matching pre-PR behavior
    exactly. Nested/reentrant use restores the previous (outer) value on
    exit rather than clobbering it, so this is safe to nest.
    """
    token = _PROFILE_HOME_RESOLVE_SCOPE.set({})
    try:
        yield
    finally:
        _PROFILE_HOME_RESOLVE_SCOPE.reset(token)


def _cached_safe_resolve_profile_home(pre_resolve: Path) -> Path:
    """`_safe_resolve()` for profile-home paths.

    Memoized only while a `profile_home_resolve_cache_scope()` is active
    (see module doc above); otherwise resolves fresh every call, matching
    pre-PR behavior.
    """
    cache = _PROFILE_HOME_RESOLVE_SCOPE.get()
    if cache is None:
        return _safe_resolve(pre_resolve)
    cached = cache.get(pre_resolve)
    if cached is not None:
        return cached
    resolved = _safe_resolve(pre_resolve)
    cache[pre_resolve] = resolved
    return resolved


def _resolve_profile_home_param(profile: str | Path | None) -> Path:
    """Resolve a profile parameter (name string, directory path string, or Path) to a profile home Path.

    Logical profile names are validated strictly (#7168 re-gate round 4): a
    name that fails the profile-id grammar raises ValueError instead of being
    silently clamped onto the default home — the old clamp let a malformed
    name such as ``"bad name"`` read/write the DEFAULT profile's state.
    Round 5 tightens the grammar gate: a STRING profile value is strictly a
    logical profile id and is NEVER treated as a path-shaped home — the old
    ``"/" in raw`` branch resolved any slash-bearing string directly, so a
    malformed value like ``"../evil"`` bypassed validation entirely and could
    read/overwrite an arbitrary ``webui_state/last_workspace.txt``
    (#7168 re-gate round 5, path traversal on the profile-isolation boundary).
    Round 6 closes the last isolation hole in this resolver: the logical
    string ``"default"`` used to short-circuit to ``_DEFAULT_HERMES_HOME``
    before reaching ``get_hermes_home_for_profile()``, bypassing the
    isolated-mode clamp in ``api.profiles._resolve_profile_home_for_name``
    (#7168 re-gate round 6). In an isolated deployment pinned at
    ``<base>/profiles/default``, a session created with ``profile="default"``
    therefore resolved workspace/config from the BASE root home instead of
    the pinned one. The name now flows through the same delegated path as
    every other logical id; literal-default routing to the global state
    files is retained in ``_profile_state_dir``/``get_last_workspace`` via
    canonical ``_is_default_profile_home`` identity.
    An explicit home directory is expressed as a ``Path`` object (callers such
    as streaming's legacy ``_profile_home`` fallback wrap their home strings
    in ``Path``); Path values are honored and canonicalized so identity
    comparisons never depend on lexical spelling (e.g. a symlink alias of the
    default home must compare equal to it).
    """
    if profile is None or str(profile).strip() == "":
        from api.profiles import get_active_hermes_home
        return get_active_hermes_home()

    raw = str(profile).strip()

    if isinstance(profile, Path):
        return _cached_safe_resolve_profile_home(profile.expanduser())

    # Strings are LOGICAL PROFILE IDS ONLY — no path-shaped strings, ever.
    if not _PROFILE_NAME_RE.fullmatch(raw):
        raise ValueError(f"invalid profile name: {raw!r}")

    from api.profiles import get_hermes_home_for_profile
    return _cached_safe_resolve_profile_home(get_hermes_home_for_profile(raw))


def _is_default_profile_home(profile_home: Path) -> bool:
    """Canonical identity check against the root/default Hermes home.

    Compares resolved paths so a symlink alias of _DEFAULT_HERMES_HOME is
    recognized as the default profile rather than treated as a foreign,
    lexically-different directory (#7168 re-gate round 4).
    """
    try:
        from api.profiles import _DEFAULT_HERMES_HOME
        return _safe_resolve(profile_home) == _safe_resolve(_DEFAULT_HERMES_HOME)
    except Exception:
        return False


def _remote_terminal_cwd(profile: str | Path | None = None) -> str | None:
    """Return target-side terminal cwd for a remote profile, without local stat()."""
    try:
        from api.config import get_config_for_profile_home

        profile_home = _resolve_profile_home_param(profile)
        terminal_cfg = get_config_for_profile_home(profile_home).get('terminal', {})

        if not _is_remote_terminal_backend(terminal_cfg):
            return None
        cwd = str(terminal_cfg.get('cwd') or '').strip()
        if not cwd or cwd == '.':
            return None
        return cwd
    except Exception:
        logger.debug("Failed to read remote terminal cwd", exc_info=True)
        return None


def _remote_terminal_workspace_candidate(path: str | Path, profile: str | Path | None = None) -> Path | None:
    """Return a non-stat'ed target-side Path when it is under terminal.cwd for the given profile.

    Remote workspace paths live on the target host (e.g., remote SSH/Docker
    backend). For valid target-side POSIX paths under ``terminal.cwd``, the
    normalized target path is preserved as a ``Path`` object without invoking
    local host-filesystem resolution (avoiding host-specific firmlink rewriting
    such as macOS synthetic ``/home`` -> ``/System/Volumes/Data/home``).
    """
    try:
        cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd()
    except TypeError:
        cwd = _remote_terminal_cwd()
    if not cwd:
        return None
    raw = _strip_surrounding_quotes(str(path)).strip()
    if not raw:
        return None
    if '\x00' in raw or '\x00' in cwd:
        return None
    normalized_raw = _normalize_posix_path(raw)
    normalized_cwd = _normalize_posix_path(cwd)
    if normalized_raw is not None and normalized_cwd is not None:
        posix_candidate = PurePosixPath(normalized_raw)
        posix_base = PurePosixPath(normalized_cwd)
        if _is_blocked_workspace_path(Path(normalized_raw), normalized_raw) or _is_blocked_workspace_path(Path(normalized_cwd), normalized_cwd):
            return None
        if posix_candidate == posix_base or _posix_is_within(posix_candidate, posix_base):
            return Path(normalized_raw)
        return None
    candidate = _safe_resolve(_expanduser_path(raw))
    base = _safe_resolve(_expanduser_path(cwd))
    if _is_blocked_workspace_path(candidate, raw) or _is_blocked_workspace_path(base, cwd):
        return None
    if candidate == base or _is_within(candidate, base):
        return candidate
    return None


def _profile_default_workspace(profile: str | Path | None = None) -> str:
    """Read the profile's default workspace from its config.yaml.

    Checks keys in priority order:
      1. 'workspace'         — explicit webui workspace key
      2. 'default_workspace' — alternate explicit key
      3. 'terminal.cwd'      — hermes-agent terminal working dir (most common)

    For remote/SSH terminal profiles, ``terminal.cwd`` lives on the target
    machine, not on the WebUI server. In that case return it without a
    server-local existence check so WebUI can send the correct workspace hint
    to the agent/tool backend.

    Falls back to the live DEFAULT_WORKSPACE from api.config.
    """
    try:
        from api.config import get_config_for_profile_home
        profile_home = _resolve_profile_home_param(profile)
        cfg = get_config_for_profile_home(profile_home)
        terminal_cfg = cfg.get('terminal', {})
        remote_terminal = _is_remote_terminal_backend(terminal_cfg)
        # Explicit webui workspace keys first
        for key in ('workspace', 'default_workspace'):
            ws = cfg.get(key)
            if ws:
                if remote_terminal:
                    return str(ws).strip()
                p = _resolve_path(str(ws), profile=profile)
                if remote_terminal or p.is_dir():
                    return str(p)
        # Fall through to terminal.cwd — the agent's configured working directory
        if isinstance(terminal_cfg, dict):
            cwd = terminal_cfg.get('cwd', '')
            if cwd and str(cwd) not in ('.', ''):
                if remote_terminal:
                    return str(cwd).strip()
                p = _resolve_path(str(cwd), profile=profile)
                if remote_terminal or p.is_dir():
                    return str(p)
    except (ImportError, Exception):
        logger.debug("Failed to load profile default workspace config")
    try:
        from api.config import DEFAULT_WORKSPACE as _LIVE_DEFAULT_WORKSPACE

        return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE, profile=profile))
    except Exception:
        return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE, profile=profile))


# ── Public API ──────────────────────────────────────────────────────────────

def _clean_workspace_list(workspaces: list, profile: str | Path | None = None) -> list:
    """Sanitize a workspace list:
    - Preserve target-side remote terminal workspace paths (SSH/Docker) without
      resolving them against the local WebUI host filesystem.
    - Preserve saved paths even when they are currently missing or inaccessible;
      picker state must not be destroyed by a transient stat/permission failure.
    - Remove entries whose paths live inside another profile's directory
      (e.g. ~/.hermes/profiles/X/... should not appear on a different profile).
    - Rename any entry whose name is literally 'default' to 'Home' (avoids
      confusion with the 'default' profile name).
    Returns the cleaned list (may be empty).
    """
    hermes_profiles = (_home_path() / '.hermes' / 'profiles').resolve()
    result = []
    for w in workspaces:
        path = w.get('path', '')
        name = w.get('name', '')
        if not path:
            continue
        remote_cand = _remote_terminal_workspace_candidate(path, profile=profile)
        if remote_cand is not None:
            p = remote_cand
        else:
            p = _safe_resolve(_expanduser_path(path))
        # Skip paths inside a DIFFERENT profile's directory (cross-profile leak).
        # Allow paths inside the CURRENT profile's own directory (e.g. test workspaces
        # created under ~/.hermes/profiles/webui/webui-mvp-test/).
        try:
            p.relative_to(hermes_profiles)
            # p is under ~/.hermes/profiles/ — only skip if it's under a DIFFERENT profile
            try:
                from api.profiles import get_active_hermes_home
                if profile is not None:
                    # Explicit profile wins: the list belongs to that profile,
                    # so "own" is defined by the profile parameter, never by the
                    # ambient home (loading profile A's list under ambient B must
                    # not silently drop A's own workspaces).
                    own_profile_dir = _resolve_profile_home_param(profile).resolve()
                else:
                    own_profile_dir = get_active_hermes_home().resolve()
                p.relative_to(own_profile_dir)
                # p is under our own profile dir — keep it
            except (ValueError, Exception):
                continue  # under profiles/ but not our own — cross-profile leak, skip
        except ValueError:
            pass  # not under profiles/ at all — keep it
        # Rename confusing 'default' label to 'Home'
        if name.lower() == 'default':
            name = 'Home'
        result.append({'path': str(p), 'name': name})
    return result


def _workspace_access_error(candidate: Path, *, missing_label: str = "Path does not exist") -> str | None:
    """Return a user-facing validation error for an unusable workspace path.

    ``Path.exists()`` can collapse permission/stat failures into a generic falsey
    result on some Python/OS combinations, which produced misleading "does not
    exist" messages for macOS/TCC-denied directories.  Probe with ``stat()`` so
    missing paths, non-directories, and permission-denied paths can be reported
    separately.
    """
    try:
        st = candidate.stat()
    except FileNotFoundError:
        return f"{missing_label}: {candidate}"
    except ValueError as exc:
        # Embedded null byte (or similar invalid path) — .stat() raises ValueError,
        # not OSError. Report as an access error rather than letting it surface as
        # an uncaught 500.
        return f"Cannot access path: {candidate!r}. Invalid path ({exc})."
    except PermissionError as exc:
        return (
            f"Cannot access path: {candidate}. The server process could not inspect "
            f"this directory ({exc}). On macOS, grant Full Disk Access or Files and "
            f"Folders permission to the Hermes/WebUI app or server process, then try again."
        )
    except OSError as exc:
        return f"Cannot access path: {candidate}. The server process could not inspect this path ({exc})."
    if not stat.S_ISDIR(st.st_mode):
        return f"Path is not a directory: {candidate}"
    return None


def _migrate_global_workspaces() -> list:
    """Read the legacy global workspaces.json, clean it, and return the result.

    This is the migration path for users upgrading from a pre-profile version:
    their global file may contain cross-profile entries, test artifacts, and
    stale paths accumulated over time.  We clean it in-place and rewrite it.
    """
    if not _GLOBAL_WS_FILE.exists():
        return []
    try:
        raw = json.loads(_GLOBAL_WS_FILE.read_text(encoding='utf-8'))
        cleaned = _clean_workspace_list(raw)
        if len(cleaned) != len(raw):
            # Rewrite the cleaned version so future reads are already clean
            _GLOBAL_WS_FILE.write_text(
                json.dumps(cleaned, ensure_ascii=False, indent=2), encoding='utf-8'
            )
        return cleaned
    except Exception:
        return []


def load_workspaces(profile: str | Path | None = None) -> list:
    ws_file = _workspaces_file_for_profile(profile)
    if ws_file is not None and ws_file.exists():
        try:
            raw = json.loads(ws_file.read_text(encoding='utf-8'))
            cleaned = _clean_workspace_list(raw, profile=profile)
            if len(cleaned) != len(raw):
                # Persist the cleaned version so stale entries don't keep reappearing
                try:
                    ws_file.write_text(
                        json.dumps(cleaned, ensure_ascii=False, indent=2), encoding='utf-8'
                    )
                except Exception:
                    logger.debug("Failed to persist cleaned workspace list")
            return cleaned or [{'path': _profile_default_workspace(profile=profile), 'name': 'Home'}]
        except Exception:
            logger.debug("Failed to load workspaces from %s", ws_file)
    # No profile-local file yet.
    # For the DEFAULT profile: migrate from the legacy global file (one-time cleanup).
    # For NAMED profiles: always start clean with just their own workspace.
    try:
        from api.profiles import get_active_profile_name
        if profile is not None:
            profile_home = _resolve_profile_home_param(profile)
            is_default = _is_default_profile_home(profile_home)
        else:
            is_default = get_active_profile_name() in ('default', None)
    except ImportError:
        is_default = True
    if is_default:
        migrated = _migrate_global_workspaces()
        if migrated:
            return migrated
    # Fresh start: single entry from the profile's configured workspace, labeled "Home"
    return [{'path': _profile_default_workspace(profile=profile), 'name': 'Home'}]


def save_workspaces(workspaces: list, profile: str | Path | None = None) -> None:
    ws_file = _workspaces_file_for_profile(profile)
    if ws_file is None:
        # Fail-closed: an invalid profile name must not write any state file
        # (it would land in the default profile's directory via the old clamp).
        raise ValueError(f"cannot save workspaces for invalid profile {profile!r}")
    ws_file.parent.mkdir(parents=True, exist_ok=True)
    ws_file.write_text(json.dumps(workspaces, ensure_ascii=False, indent=2), encoding='utf-8')


def get_profile_default_workspace(profile: str | Path | None = None) -> str:
    """Resolve the ACTIVE PROFILE's default workspace, never the global file.

    Like get_last_workspace() but WITHOUT the global ``_GLOBAL_LW_FILE``
    fallback: for a named profile that has not yet written its own
    profile-scoped ``last_workspace.txt``, that global fallback would leak the
    *global* last-workspace instead of the profile's configured workspace —
    which is exactly the #5169 bug (the composer chip on a blank new-chat page
    showing the wrong/global workspace for a named profile). Used by
    ``GET /api/profile/active`` so a cold boot under a profile cookie reflects
    the profile's own configured working directory.

    Priority: profile-scoped ``last_workspace.txt`` -> profile ``config.yaml``
    ``workspace``/``default_workspace`` -> ``terminal.cwd`` -> process default.
    """
    try:
        remote_cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd()
    except TypeError:
        remote_cwd = _remote_terminal_cwd()

    def _valid(raw: str) -> str | None:
        if not raw:
            return None
        if remote_cwd:
            if _remote_terminal_workspace_candidate(raw, profile=profile) is not None:
                return raw
            return None
        if Path(raw).is_dir():
            return raw
        return None

    lw_file = _last_workspace_file_for_profile(profile)
    if lw_file is not None and lw_file.exists():
        try:
            p = _valid(lw_file.read_text(encoding='utf-8').strip())
            if p:
                return p
        except Exception:
            logger.debug("Failed to read profile last workspace from %s", lw_file)
    return _profile_default_workspace(profile=profile)


def get_last_workspace(profile: str | Path | None = None) -> str:
    try:
        remote_cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd()
    except TypeError:
        remote_cwd = _remote_terminal_cwd()

    def valid_last_workspace(raw: str) -> str | None:
        if not raw:
            return None
        if remote_cwd:
            # For remote/SSH profiles, last_workspace is target-side state. Do
            # not accept stale server-local paths merely because they exist on
            # the WebUI host; require the value to stay under terminal.cwd.
            if _remote_terminal_workspace_candidate(raw, profile=profile) is not None:
                return raw
            return None
        if Path(raw).is_dir():
            return raw
        return None

    lw_file = _last_workspace_file_for_profile(profile)
    if lw_file is not None and lw_file.exists():
        try:
            p = valid_last_workspace(lw_file.read_text(encoding='utf-8').strip())
            if p:
                return p
        except Exception:
            logger.debug("Failed to read last workspace from %s", lw_file)
    # Fallback: try global file — but ONLY for the root/default profile. A named
    # profile must never inherit another profile's last-workspace binding through
    # the legacy global state (#7168 re-gate round 3). Identity is canonical:
    # a symlink alias of the default home still counts as default, while a
    # malformed profile name fails validation and is denied the fallback
    # rather than being clamped onto the global file (#7168 re-gate round 4).
    _global_fallback_allowed = True  # ambient / no explicit profile: historical behavior
    if profile is not None:
        if str(profile).strip() == 'default':
            _global_fallback_allowed = True
        else:
            try:
                _global_fallback_allowed = _is_default_profile_home(
                    _resolve_profile_home_param(profile)
                )
            except Exception:
                # Conservative default: an unresolvable explicit profile is NOT the
                # root profile — deny the global fallback rather than leak.
                _global_fallback_allowed = False
    if _global_fallback_allowed and _GLOBAL_LW_FILE.exists():
        try:
            p = valid_last_workspace(_GLOBAL_LW_FILE.read_text(encoding='utf-8').strip())
            if p:
                return p
        except Exception:
            logger.debug("Failed to read global last workspace")
    return _profile_default_workspace(profile=profile)


def set_last_workspace(path: str, profile: str | Path | None = None) -> None:
    try:
        lw_file = _last_workspace_file_for_profile(profile)
        if lw_file is None:
            # Fail-closed: an invalid profile name must not write the default
            # profile's last_workspace.txt (#7168 re-gate round 4).
            logger.debug("Refusing to set last workspace for invalid profile %r", profile)
            return
        lw_file.parent.mkdir(parents=True, exist_ok=True)
        lw_file.write_text(str(path), encoding='utf-8')
    except Exception:
        logger.debug("Failed to set last workspace")


def _safe_resolve(p: Path) -> Path:
    """Path.resolve() that never raises — falls back to the input path on error."""
    try:
        return p.resolve()
    except (OSError, RuntimeError, ValueError):
        # ValueError covers embedded-null-byte paths, which .resolve() raises on
        # — fail-closed to the raw path so the downstream block-list gate rejects
        # it cleanly instead of surfacing a 500.
        return p


# Per-user temp directories that sit nominally under a "system" prefix but are
# actually user-writable scratch space.  Workspaces registered here (e.g. by
# pytest's ``tmp_path_factory`` on macOS, which uses ``/var/folders/<hash>/T/``)
# must remain accepted even though their parent (``/var``) is blocked.  These
# carve-outs apply to BOTH workspace registration and runtime file ops so a
# symlink target inside the carve-out is also reachable.
_USER_TMP_PREFIXES: tuple[Path, ...] = (
    Path('/var/folders'),         # macOS per-user tmp (literal form)
    Path('/private/var/folders'),  # macOS per-user tmp (resolved form)
    Path('/var/tmp'),               # Linux/macOS system-wide tmp (user-writable)
    Path('/private/var/tmp'),       # macOS resolved form
)


def _workspace_blocked_roots() -> tuple[Path, ...]:
    """System roots that must never be accepted as workspace candidates.

    Returns both the literal path and its symlink-resolved canonical form,
    deduped.  This matters on macOS where ``/etc``, ``/var``, and ``/tmp``
    are symlinks to ``/private/etc`` etc.  Without the resolved forms,
    callers that pass a ``.resolve()``-d candidate (every caller does)
    would compare ``/private/etc`` against literal ``Path('/etc')`` and the
    ``relative_to`` check would miss — letting ``/etc`` through as a
    registered workspace on macOS.

    Carve-outs for legitimate user-tmp paths nominally under these roots
    (e.g. ``/var/folders/.../T/`` on macOS) are handled by
    :func:`_is_blocked_system_path`, not by exclusion from this list.
    """
    _raw = (
        # Linux / macOS
        '/etc',
        '/usr',
        '/var',
        '/bin',
        '/sbin',
        '/boot',
        '/proc',
        '/sys',
        '/dev',
        '/lib',
        '/lib64',
        '/opt/homebrew',
        '/System',
        '/Library',
    )
    _seen: set[Path] = set()
    _out: list[Path] = []
    for _p in _raw:
        for _form in (Path(_p), _safe_resolve(Path(_p))):
            if _form not in _seen:
                _seen.add(_form)
                _out.append(_form)
    return tuple(_out)


def _is_blocked_posix_workspace_path(raw_path: str | Path | None) -> bool:
    """Detect blocked POSIX-style system roots even on non-POSIX hosts."""
    candidate = _as_posix_path(raw_path)
    if candidate is None:
        return False
    if candidate == PurePosixPath('/'):
        return True
    carveouts = (
        PurePosixPath('/var/folders'),
        PurePosixPath('/private/var/folders'),
        PurePosixPath('/var/tmp'),
        PurePosixPath('/private/var/tmp'),
    )
    for tmp in carveouts:
        if _posix_is_within(candidate, tmp):
            return False
    blocked_roots = (
        PurePosixPath('/etc'),
        PurePosixPath('/usr'),
        PurePosixPath('/var'),
        PurePosixPath('/bin'),
        PurePosixPath('/sbin'),
        PurePosixPath('/boot'),
        PurePosixPath('/proc'),
        PurePosixPath('/sys'),
        PurePosixPath('/dev'),
        PurePosixPath('/lib'),
        PurePosixPath('/lib64'),
        PurePosixPath('/opt/homebrew'),
        PurePosixPath('/System'),
        PurePosixPath('/Library'),
        PurePosixPath('/private/etc'),
        PurePosixPath('/private/var'),
    )
    for blocked in blocked_roots:
        if _posix_is_within(candidate, blocked):
            return True
    return False


def _is_blocked_system_path(candidate: Path) -> bool:
    """Return True if *candidate* falls under a blocked system root.

    Honours :data:`_USER_TMP_PREFIXES` carve-outs so per-user tmp directories
    nominally under ``/var`` (``/var/folders`` on macOS, ``/var/tmp`` on
    Linux/macOS) remain valid workspace candidates and reachable file targets.
    """
    for tmp in _USER_TMP_PREFIXES:
        if _is_within(candidate, tmp):
            return False
    for blocked in _workspace_blocked_roots():
        if _is_within(candidate, blocked):
            return True
    return False


def _workspace_blocked_resolved_subtrees() -> tuple[Path, ...]:
    roots = list(_workspace_blocked_roots()) + [Path('/private/etc')]
    resolved: list[Path] = []
    for root in roots:
        try:
            p = root.expanduser().resolve()
        except Exception:
            p = root
        if p not in resolved:
            resolved.append(p)
    return tuple(resolved)


def _workspace_blocked_exact_roots() -> tuple[Path, ...]:
    roots = [Path('/'), Path('/private/var')]
    for root in _workspace_blocked_roots():
        try:
            roots.append(root.expanduser().resolve())
        except Exception:
            roots.append(root)
    unique: list[Path] = []
    for root in roots:
        if root not in unique:
            unique.append(root)
    return tuple(unique)


def _is_blocked_workspace_path(candidate: Path, raw_path: str | Path | None = None) -> bool:
    """Return True when candidate points at a known OS/system directory.

    Compare both the original spelling and the resolved path.  This closes the
    macOS /etc -> /private/etc bypass without globally banning temporary pytest
    paths under /private/var/folders.
    """
    raw = None
    if raw_path not in (None, ""):
        try:
            normalized_posix = _normalize_posix_path(raw_path)
            raw = Path(normalized_posix) if normalized_posix is not None else _expanduser_path(raw_path)
        except Exception:
            raw = None

    posix_probe = raw_path if raw_path not in (None, "") else candidate.as_posix()
    if _is_blocked_posix_workspace_path(posix_probe):
        return True

    exact = _workspace_blocked_exact_roots()
    if candidate in exact or (raw is not None and raw in _workspace_blocked_roots()):
        return True

    for tmp in _USER_TMP_PREFIXES:
        if _is_within(candidate, tmp) or (raw is not None and _is_within(raw, tmp)):
            return False

    # Raw paths under literal roots (e.g. /etc/ssh, /var/db) are always blocked.
    if raw is not None:
        for blocked in _workspace_blocked_roots():
            if _is_within(raw, blocked):
                return True

    # Resolved subtree checks catch symlink aliases such as /private/etc.  The
    # macOS temp root /private/var/folders is intentionally allowed for pytest
    # and per-user temporary workspaces; other direct /private/var system data
    # such as /private/var/db and /private/var/log remains blocked.
    allowed_private_var = (Path('/private/var/folders'), Path('/private/var/tmp'))
    for blocked in _workspace_blocked_resolved_subtrees():
        if blocked == Path('/private/var'):
            if candidate == blocked:
                return True
            if any(_is_within(candidate, allowed) for allowed in allowed_private_var):
                continue
            if _is_within(candidate, blocked):
                return True
            continue
        if _is_within(candidate, blocked):
            return True
    return False


def _is_within(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
        return True
    except ValueError:
        return False


def _trusted_workspace_roots(profile: str | Path | None = None) -> list[Path]:
    """Return the host directories workspace suggestions may traverse.

    Saved-workspace roots follow the same trust rule as
    :func:`resolve_trusted_workspace`: with an explicit *profile*, only
    workspaces saved under THAT profile widen the boundary (plus the ambient
    home / boot-default carve-outs); ``None`` keeps the historical ambient /
    global saved-list behaviour.
    """
    roots: list[Path] = []

    def add(candidate: str | Path | None) -> None:
        if candidate in (None, ""):
            return
        try:
            p = _resolve_path(candidate)
        except Exception:
            return
        if not p.exists() or not p.is_dir():
            return
        if _is_blocked_workspace_path(p, candidate):
            return
        if p not in roots:
            roots.append(p)

    add(_home_path())
    add(_BOOT_DEFAULT_WORKSPACE)
    for w in load_workspaces(profile=profile):
        add(w.get("path"))
    roots.sort(key=lambda p: len(str(p)))
    return roots


def list_workspace_suggestions(
    prefix: str = "", limit: int = 12, profile: str | Path | None = None
) -> list[str]:
    """Return workspace path suggestions under trusted roots only.

    Suggestions are limited to directories under one of:
      - Path.home()
      - the boot default workspace
      - already-saved workspace roots (scoped to *profile* when given)

    Arbitrary system prefixes return an empty list rather than an error so the
    UI can safely autocomplete while the user types.
    """
    roots = _trusted_workspace_roots(profile=profile)
    if not roots:
        return []

    raw = (prefix or "").strip()
    if not raw:
        return [str(p) for p in roots[:limit]]

    if raw.startswith("~"):
        target = _expanduser_path(raw)
    elif Path(raw).is_absolute():
        target = Path(raw)
    else:
        target = _home_path() / raw

    try:
        match_target = _resolve_path(target)
    except Exception:
        match_target = target

    normalized = str(match_target)
    normalized_lower = normalized.lower()
    preserve_tilde = raw.startswith("~")
    home_root: Path | None = None
    if preserve_tilde:
        try:
            home_root = _home_path()
        except Exception:
            home_root = None
    suggestions: list[str] = []

    def format_suggestion(path: Path) -> str:
        if preserve_tilde and home_root is not None:
            try:
                rel = path.resolve().relative_to(home_root)
                if str(rel) == ".":
                    return "~"
                return "~/" + rel.as_posix()
            except (OSError, ValueError):
                pass
        return str(path)

    def add(path: Path) -> None:
        value = format_suggestion(path)
        if value not in suggestions:
            suggestions.append(value)

    # If the user is typing a partial trusted root like /Users/xuef..., suggest
    # the matching trusted roots without scanning arbitrary system parents.
    for root in roots:
        if str(root).lower().startswith(normalized_lower):
            add(root)

    in_root = [
        root
        for root in roots
        if normalized == str(root) or normalized.startswith(str(root) + os.sep)
    ]
    if not in_root:
        return suggestions[:limit]

    anchor_root = max(in_root, key=lambda p: len(str(p)))
    ends_with_sep = raw.endswith(os.sep) or raw.endswith('/')
    parent = target if ends_with_sep else target.parent
    leaf = '' if ends_with_sep else target.name
    show_hidden = leaf.startswith('.')

    try:
        parent_resolved = _resolve_path(parent)
    except Exception:
        return suggestions[:limit]

    if not parent_resolved.exists() or not parent_resolved.is_dir():
        return suggestions[:limit]
    if not _is_within(parent_resolved, anchor_root):
        return suggestions[:limit]

    leaf_lower = leaf.lower()
    try:
        children = sorted(parent_resolved.iterdir(), key=lambda p: p.name.lower())
    except OSError:
        return suggestions[:limit]

    for child in children:
        if not child.is_dir():
            continue
        if child.name.startswith('.') and not show_hidden:
            continue
        if leaf_lower and not child.name.lower().startswith(leaf_lower):
            continue
        add(child.resolve())
        if len(suggestions) >= limit:
            break
    return suggestions[:limit]


def resolve_trusted_workspace(path: str | Path | None = None, profile: str | Path | None = None) -> Path:
    """Resolve and validate a workspace path.

    A path is trusted if it satisfies at least one of:
      (A) It is under the user's home directory (Path.home()).
          Works cross-platform: ~/... on Linux/macOS, C:\\Users\\... on Windows.
      (B) It is already in the profile's saved workspace list.
          This covers self-hosted deployments where workspaces live outside home
          (e.g. /data/projects, /opt/workspace) — once a workspace is saved by
          an admin, it can be reused without re-validation.

    Additionally enforced regardless of (A)/(B):
      1. The path must exist.
      2. The path must be a directory.
      3. The path must not be a known system root (/etc, /usr, /var, /bin, /sbin,
         /boot, /proc, /sys, /dev, /root on Linux/macOS; Windows system dirs).
         This prevents even admin-saved workspaces from pointing at OS internals.

    None/empty path falls back to the boot-time DEFAULT_WORKSPACE, which is always
    trusted (it was validated at server startup).
    """
    if path in (None, ""):
        return _resolve_path(_BOOT_DEFAULT_WORKSPACE, profile) if profile is not None else _resolve_path(_BOOT_DEFAULT_WORKSPACE)

    candidate = _resolve_path(path, profile) if profile is not None else _resolve_path(path)

    access_error = _workspace_access_error(candidate)
    remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile)
    if access_error:
        # For remote terminal profiles, workspace paths belong to the target
        # machine. Allow paths under terminal.cwd so session switching can
        # update the workspace hint even though this WebUI host cannot stat
        # the target-side path.
        if remote_candidate is None:
            raise ValueError(access_error)

    if remote_candidate is not None:
        return remote_candidate

    # (A) Trusted if under the user's home directory — cross-platform via Path.home()
    # Must be checked before system roots to allow symlinks like /var/home.
    _home = _home_path()
    if _home != Path("/"):
        try:
            candidate.relative_to(_home)
            return candidate
        except ValueError:
            pass

    if _is_blocked_workspace_path(candidate, path):
        raise ValueError(f"Path points to a system directory: {candidate}")

    # (B) Trusted if already in the saved workspace list — covers non-home installs
    try:
        saved = load_workspaces(profile=profile)
        saved_paths = {_resolve_path(w["path"], profile) for w in saved if w.get("path")}
        if candidate in saved_paths:
            return candidate
    except TypeError:
        saved = load_workspaces()
        saved_paths = {_resolve_path(w["path"]) for w in saved if w.get("path")}
        if candidate in saved_paths:
            return candidate
    except Exception:
        pass

    # (C) Trusted if it is equal to or under the boot-time DEFAULT_WORKSPACE.
    #     In Docker deployments HERMES_WEBUI_DEFAULT_WORKSPACE is often set to a
    #     volume mount outside the user's home (e.g. /data/workspace).  That path
    #     was already validated at server startup, so any sub-path of it is safe
    #     without requiring the user to add it to the workspace list manually.
    try:
        boot_default = _resolve_path(_BOOT_DEFAULT_WORKSPACE)
        candidate.relative_to(boot_default)
        return candidate
    except ValueError:
        pass

    raise ValueError(
        f"Path is outside the user home directory, not in the saved workspace "
        f"list, and not under the default workspace: {candidate}. "
        f"Add it via Settings → Workspaces first."
    )


def resolve_implicit_workspace_with_recovery(
    candidate: str | Path | None,
    fallback: str | Path | None | Callable[[], str | Path | None],
    profile: str | Path | None = None,
) -> tuple[Path, bool]:
    """Resolve an implicit workspace, recovering only a genuinely missing path.

    The fallback still passes through :func:`resolve_trusted_workspace`. Existing
    but untrusted, inaccessible, or non-directory candidates are not recovery
    cases: their original validation error is preserved so fallback cannot widen
    the workspace trust boundary. When *profile* is given, both the trust
    resolution and the recovery fallback are scoped to that profile.
    """
    try:
        if profile is not None:
            try:
                return resolve_trusted_workspace(candidate, profile=profile), False
            except TypeError:
                pass
        return resolve_trusted_workspace(candidate), False
    except ValueError as original_error:
        if candidate in (None, ""):
            raise original_error from None
        # Remote terminal workspaces live on the target host. Classify the
        # backend independently of terminal.cwd: remote backends may omit cwd,
        # set it to an empty string, or use ".". A failed host-local stat can
        # never prove target-side deletion. Config-read uncertainty also fails
        # closed by preserving the original validation error.
        try:
            from api.config import get_config, get_config_for_profile_home

            if profile is not None:
                # Classify the backend from THIS profile's own config, never the
                # ambient one: a remote profile without terminal.cwd must still be
                # recognized as remote when loaded under a different ambient home.
                terminal_cfg = get_config_for_profile_home(
                    _resolve_profile_home_param(profile)
                ).get("terminal", {})
            else:
                terminal_cfg = get_config().get("terminal", {})
        except Exception:
            logger.debug("Failed to classify terminal backend for workspace recovery", exc_info=True)
            raise original_error from None
        if _is_remote_terminal_backend(terminal_cfg):
            raise original_error from None
        try:
            local_candidate = (
                _resolve_path(candidate, profile=profile)
                if profile is not None
                else _resolve_path(candidate)
            )
            local_candidate.stat()
        except FileNotFoundError:
            def _profile_bound_fallback():
                if profile is not None and callable(fallback):
                    # Profile-bound getter FIRST (#7168 re-gate round 3): the
                    # production call sites pass profile-aware getters such as
                    # get_last_workspace, so binding the explicit profile must
                    # take precedence over any zero-argument compatibility call,
                    # which would read ambient/global state.
                    try:
                        return fallback(profile)
                    except TypeError:
                        pass
                return fallback() if callable(fallback) else fallback

            fallback_value = _profile_bound_fallback()
            if profile is not None:
                try:
                    return resolve_trusted_workspace(fallback_value, profile=profile), True
                except TypeError:
                    pass
            return resolve_trusted_workspace(fallback_value), True
        except (OSError, RuntimeError, ValueError):
            raise original_error from None
        raise original_error from None




def _strip_surrounding_quotes(path: str) -> str:
    """Strip a single pair of surrounding single or double quotes from a path string.

    macOS Finder's "Copy as Pathname" (Cmd+Option+C) returns paths wrapped in
    single quotes, e.g. ``'/Users/x/Documents/foo'``. Other shells and OS file
    managers do similar things with double quotes. Users routinely paste these
    quoted strings into the Add Space input expecting them to "just work" —
    the only reason they didn't was a missing strip.

    Only paired quotes are stripped (matching opener and closer). One-sided quotes
    are preserved on the slim chance a path legitimately contains a literal quote
    character.
    """
    s = path.strip()
    if len(s) >= 2 and s[0] == s[-1] and s[0] in ("'", '"'):
        return s[1:-1]
    return s


def validate_workspace_to_add(path: str, profile: str | Path | None = None) -> Path:
    """Validate a path for *adding* to the workspace list (less restrictive than resolve_trusted_workspace).

    When a user explicitly adds a new workspace path, we trust their intent — they
    have console or filesystem access to that path and are consciously registering it.
    We only block: non-existent paths, non-directories, and known system roots.

    The stricter ``resolve_trusted_workspace`` is used when *using* an existing workspace
    (file reads/writes) to prevent path traversal after the list is built.

    Surrounding quotes (single or double) are stripped before validation —
    macOS Finder's "Copy as Pathname" wraps paths in single quotes by default,
    and users routinely paste those into the Add Space input.
    """
    path = _strip_surrounding_quotes(path)
    candidate = _resolve_path(path, profile) if profile is not None else _resolve_path(path)

    access_error = _workspace_access_error(candidate)
    remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile)
    if access_error:
        # Remote terminal profiles validate workspace existence on the target
        # machine, not on the WebUI server. Permit target-side paths under
        # terminal.cwd.
        if remote_candidate is None:
            raise ValueError(access_error)

    if remote_candidate is not None:
        return remote_candidate

    # Home directory is always trusted regardless of where it lives on disk
    # (e.g. /var/home/... on systemd-homed Fedora/RHEL).
    _home = _home_path()
    if _home != Path("/") and _is_within(candidate, _home):
        return candidate

    if _is_blocked_workspace_path(candidate, path):
        raise ValueError(f"Path points to a system directory: {candidate}")

    return candidate

def safe_resolve_ws(root: Path, requested: str) -> Path:
    """Resolve a relative path inside a workspace root, raising ValueError on traversal.

    Both raw ``..`` traversal and symlink escapes are blocked.  Workspace file
    APIs can be reached by browser UI actions and agent/tool calls, so a symlink
    inside the workspace must not expand the trusted workspace boundary to an
    arbitrary host path.
    """
    root_resolved = root.resolve()
    resolved = (root / requested).resolve()
    try:
        resolved.relative_to(root_resolved)
    except ValueError:
        raise ValueError(f"Path traversal blocked: {requested}")
    return resolved


# ── Race-safe (TOCTOU) anchored open ─────────────────────────────────────────
# safe_resolve_ws() validates a path, but if callers then re-open by pathname a
# symlink swapped in AFTER the check could still escape the workspace. To close
# that window we open the (already symlink-resolved) target component-by-component
# from the workspace root using openat (dir_fd) + O_NOFOLLOW: every component must
# be a real, non-symlink entry, so a component swapped to a symlink mid-flight is
# refused. Legit in-workspace symlinks still work because safe_resolve_ws() has
# already collapsed them to their real in-workspace target, and we walk that real
# (symlink-free) path. Portable: uses os.supports_dir_fd where available (Linux,
# macOS); on platforms without dir_fd support (Windows — where creating symlinks
# also requires admin) we fall back to a plain pathname open, matching the prior
# behaviour with no regression.

_DIR_FD_OK = os.open in getattr(os, "supports_dir_fd", set())
_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
_O_DIRECTORY = getattr(os, "O_DIRECTORY", 0)
_O_BINARY = getattr(os, "O_BINARY", 0)


def open_anchored_fd(workspace: Path, target: Path, *, want_dir: bool) -> int:
    """Open ``target`` race-safely and return an owned file descriptor.

    ``target`` must be the symlink-resolved path returned by safe_resolve_ws()
    (i.e. already verified to live under the workspace). Raises FileNotFoundError
    if a component is missing / wrong-type, or ValueError if a component was
    swapped to a symlink (escape attempt). Caller owns and must close the fd.
    """
    root_resolved = workspace.resolve()
    # Relative, symlink-free component list (resolve() already collapsed any links).
    try:
        rel_parts = target.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {target}") from None

    if not _DIR_FD_OK:
        # Windows / no openat: fall back to a plain pathname open. No new race
        # protection, but no regression vs the prior path-based behaviour, and
        # symlink creation needs admin on Windows anyway.
        flags = (
            os.O_RDONLY
            | (_O_DIRECTORY if want_dir else _O_BINARY)
            | _O_NOFOLLOW
        )
        try:
            return os.open(str(target), flags)
        except OSError:
            raise FileNotFoundError(f"Not found: {target}") from None

    # Open the (trusted) workspace root. root_resolved is canonical (resolve()
    # collapsed any symlinks to REACH it, e.g. macOS /tmp -> /private/tmp), so its
    # final component is legitimately a real directory — O_NOFOLLOW here only fires
    # if the root itself was raced into a symlink after resolve() (escape attempt).
    fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW)
    try:
        for i, part in enumerate(rel_parts):
            is_last = i == len(rel_parts) - 1
            want_directory = (not is_last) or want_dir
            flags = (
                os.O_RDONLY
                | _O_NOFOLLOW
                | (_O_DIRECTORY if want_directory else _O_BINARY)
            )
            try:
                nfd = os.open(part, flags, dir_fd=fd)
            except OSError:
                # ELOOP (component is a symlink — swapped in) or missing/wrong type.
                raise FileNotFoundError(f"Not found: {target}") from None
            os.close(fd)
            fd = nfd
        return fd
    except BaseException:
        try:
            os.close(fd)
        except OSError:
            pass
        raise


def open_anchored_create_fd(root: Path, dest: Path) -> int:
    """Create ``dest`` for exclusive writing race-safely, anchored under ``root``.

    Walks from ``root`` via openat + O_NOFOLLOW (creating missing intermediate
    directories with mkdir(dir_fd=...)), then creates the leaf with
    O_CREAT|O_EXCL|O_NOFOLLOW so a symlink raced into any component cannot
    redirect the write outside ``root``. ``dest`` must be the resolved path and
    must not already exist (callers dedup first). Raises ValueError if ``dest``
    is not under ``root``, FileExistsError if it exists, FileNotFoundError if a
    component was swapped to a symlink. Caller owns and must close the returned
    write fd. On platforms without dir_fd support (Windows) falls back to a plain
    exclusive create — no new race protection but no regression.
    """
    root_resolved = root.resolve()
    try:
        rel_parts = dest.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {dest}") from None
    if not rel_parts:
        raise ValueError(f"Invalid destination: {dest}")

    if not _DIR_FD_OK:
        # Windows / no openat: create parent dirs then exclusively create the leaf.
        dest.parent.mkdir(parents=True, exist_ok=True)
        return os.open(str(dest), os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW, 0o644)

    fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW)
    try:
        for part in rel_parts[:-1]:
            try:
                nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd)
            except FileNotFoundError:
                os.mkdir(part, 0o755, dir_fd=fd)
                nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd)
            except OSError:
                # ELOOP — component swapped to a symlink (escape attempt).
                raise FileNotFoundError(f"Not found: {dest}") from None
            os.close(fd)
            fd = nfd
        return os.open(
            rel_parts[-1],
            os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW,
            0o644,
            dir_fd=fd,
        )
    finally:
        try:
            os.close(fd)
        except OSError:
            pass


def make_anchored_dir(root: Path, dest: Path) -> None:
    """Create directory ``dest`` (and any missing parents) race-safely under ``root``.

    Walks from ``root`` via openat + O_NOFOLLOW, creating each missing component
    with mkdir(dir_fd=...), so a symlink raced into any component cannot make the
    server create directories outside ``root``. Idempotent (existing dirs are
    fine). Raises ValueError if ``dest`` is not under ``root``, FileNotFoundError
    if a component was swapped to a symlink. On platforms without dir_fd support
    (Windows) falls back to a plain Path.mkdir — no regression.
    """
    root_resolved = root.resolve()
    dest_resolved = dest.resolve()
    if dest_resolved == root_resolved:
        return
    try:
        rel_parts = dest_resolved.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {dest}") from None

    if not _DIR_FD_OK:
        dest.mkdir(parents=True, exist_ok=True)
        return

    fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW)
    try:
        for part in rel_parts:
            try:
                nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd)
            except FileNotFoundError:
                os.mkdir(part, 0o755, dir_fd=fd)
                nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd)
            except OSError:
                # ELOOP — component swapped to a symlink (escape attempt).
                raise FileNotFoundError(f"Not found: {dest}") from None
            os.close(fd)
            fd = nfd
    finally:
        try:
            os.close(fd)
        except OSError:
            pass


def open_anchored_write_fd(root: Path, target: Path) -> int:
    """Open existing ``target`` for truncating writes anchored under ``root``."""
    root_resolved = root.resolve()
    target_resolved = target.resolve()
    try:
        rel_parts = target_resolved.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {target}") from None
    if not rel_parts:
        raise ValueError(f"Invalid target: {target}")

    flags = os.O_WRONLY | os.O_TRUNC | _O_NOFOLLOW
    if not _DIR_FD_OK:
        return os.open(str(target_resolved), flags)

    parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True)
    try:
        return os.open(rel_parts[-1], flags, dir_fd=parent_fd)
    finally:
        os.close(parent_fd)


def unlink_anchored(root: Path, target: Path) -> None:
    """Unlink an existing file anchored under ``root``."""
    root_resolved = root.resolve()
    target_resolved = target.resolve()
    try:
        rel_parts = target_resolved.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {target}") from None
    if not rel_parts:
        raise ValueError(f"Invalid target: {target}")

    if not _DIR_FD_OK:
        target_resolved.unlink()
        return

    parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True)
    try:
        os.unlink(rel_parts[-1], dir_fd=parent_fd)
    finally:
        os.close(parent_fd)


def rmtree_anchored(root: Path, target: Path) -> None:
    """Remove a directory tree anchored under ``root`` without following symlink swaps."""
    root_resolved = root.resolve()
    target_resolved = target.resolve()
    try:
        rel_parts = target_resolved.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {target}") from None
    if not rel_parts:
        raise ValueError(f"Invalid target: {target}")

    if not _DIR_FD_OK:
        shutil.rmtree(target_resolved)
        return

    parent_fd = open_anchored_fd(root_resolved, target_resolved.parent, want_dir=True)
    try:
        shutil.rmtree(rel_parts[-1], dir_fd=parent_fd)
    finally:
        os.close(parent_fd)


def rename_anchored(root: Path, source: Path, dest: Path) -> None:
    """Rename ``source`` to ``dest`` using anchored parent directory fds."""
    root_resolved = root.resolve()
    source_resolved = source.resolve()
    dest_parent_resolved = dest.parent.resolve()
    try:
        source_parts = source_resolved.relative_to(root_resolved).parts
    except ValueError:
        raise ValueError(f"Path traversal blocked: {source}") from None
    try:
        dest_parent_resolved.relative_to(root_resolved)
    except ValueError:
        raise ValueError(f"Path traversal blocked: {dest}") from None
    if not source_parts:
        raise ValueError(f"Invalid source: {source}")
    dest_leaf = dest.name
    if not dest_leaf:
        raise ValueError(f"Invalid destination: {dest}")

    if not _DIR_FD_OK:
        source_resolved.rename(dest)
        return

    src_parent_fd = open_anchored_fd(root_resolved, source_resolved.parent, want_dir=True)
    try:
        dst_parent_fd = open_anchored_fd(root_resolved, dest_parent_resolved, want_dir=True)
        try:
            try:
                os.stat(dest_leaf, dir_fd=dst_parent_fd, follow_symlinks=False)
                raise FileExistsError(dest_leaf)
            except FileNotFoundError:
                pass
            os.rename(
                source_parts[-1],
                dest_leaf,
                src_dir_fd=src_parent_fd,
                dst_dir_fd=dst_parent_fd,
            )
        finally:
            os.close(dst_parent_fd)
    finally:
        os.close(src_parent_fd)


def _birthtime_ns(lst) -> int | None:
    """Return creation time in ns, or None when the platform lacks birthtime."""
    value = getattr(lst, 'st_birthtime_ns', None)
    if value is not None:
        return value
    value = getattr(lst, 'st_birthtime', None)
    if value is not None:
        return int(value * 1_000_000_000)
    if sys.platform == 'win32':
        return getattr(lst, 'st_ctime_ns', None)
    return None


def _browser_timestamp_ns(value) -> str | None:
    if value is None:
        return None
    try:
        return str(int(value))
    except (TypeError, ValueError):
        return None


def serialize_workspace_entries_for_browser(entries: list[dict] | None) -> list[dict]:
    payload = []
    for entry in entries or []:
        item = dict(entry or {})
        if 'mtime_ns' in item:
            item['mtime_ns'] = _browser_timestamp_ns(item.get('mtime_ns'))
        if 'birthtime_ns' in item:
            item['birthtime_ns'] = _browser_timestamp_ns(item.get('birthtime_ns'))
        payload.append(item)
    return payload


def list_dir(workspace: Path, rel: str='.'):
    target = safe_resolve_ws(workspace, rel)
    if not target.is_dir():
        raise FileNotFoundError(f"Not a directory: {rel}")
    ws_resolved = workspace.resolve()
    target_resolved = target.resolve()
    entries = []

    def _process(name, is_symlink, raw_link, lstat_result, reachable):
        """Append one directory entry. ``raw_link`` is the os.readlink() result
        for symlinks (else None); ``lstat_result`` is an os.stat_result obtained
        with follow_symlinks=False (else None); ``reachable`` is False when a
        follow_symlinks=True stat raised (broken target or symlink loop)."""
        if is_symlink:
            # Keep the transport rank aligned with _sort_key_de/_sort_key_p.
            workspace_sort_rank = 0
            if raw_link is None:
                return
            # A symlink whose follow-stat raised (ELOOP / broken target) can never
            # be opened — filter it. This catches mutual/self loops portably across
            # Python versions where Path.resolve() loop handling differs (3.11
            # raises RuntimeError, 3.13 can return a path), so do not rely on
            # resolve() raising for cycle detection.
            if not reachable:
                return
            try:
                link_target = (target_resolved / raw_link).resolve()
            except (OSError, RuntimeError):
                return
            # Cycle detection: skip if symlink points back to current dir or root.
            if link_target == target_resolved or link_target == ws_resolved:
                return
            try:
                target_resolved.relative_to(link_target)
                return  # target is under link_target — ancestor → cycle
            except ValueError:
                pass
            # Tag symlinks whose resolved target escapes the workspace root.
            # Previously silently dropped; now emitted with target_outside_workspace=True
            # so the workspace tree can show the link exists (display-only — the
            # read/list gate in safe_resolve_ws / open_anchored_fd still blocks
            # navigation through it).
            target_outside_workspace = False
            try:
                link_target.relative_to(ws_resolved)
            except ValueError:
                target_outside_workspace = True
            if _is_blocked_system_path(link_target):
                return
            display_path = name
            if rel and rel != '.':
                display_path = rel + '/' + display_path
            mtime_ns = lstat_result.st_mtime_ns if lstat_result is not None else None
            if target_outside_workspace:
                # #4581 hardening: a display-only escape-target symlink must NOT
                # disclose where it points. Emit ONLY display-safe fields — never
                # the resolved outside path, target-derived is_dir, or target size
                # (the row exists to show the link is present; navigation/read
                # through it stays blocked by safe_resolve_ws/open_anchored_fd).
                entry = {
                    'name': name,
                    'path': display_path,
                    'type': 'symlink',
                    'is_dir': False,
                    'workspace_sort_rank': workspace_sort_rank,
                    'target_outside_workspace': True,
                    'mtime_ns': mtime_ns,
                    'birthtime_ns': _birthtime_ns(lstat_result) if lstat_result is not None else None,
                }
                entries.append(entry)
            else:
                is_dir = link_target.is_dir()
                entry = {
                    'name': name,
                    'path': display_path,
                    'type': 'symlink',
                    'target': str(link_target),
                    'is_dir': is_dir,
                    'workspace_sort_rank': workspace_sort_rank,
                    'target_outside_workspace': False,
                    'mtime_ns': mtime_ns,
                    'birthtime_ns': _birthtime_ns(lstat_result) if lstat_result is not None else None,
                }
                if not is_dir:
                    try:
                        entry['size'] = link_target.stat().st_size
                    except OSError:
                        entry['size'] = None
                entries.append(entry)
        else:
            entry_path = name
            if rel and rel != '.':
                entry_path = rel + '/' + name
            if lstat_result is not None:
                is_file = stat.S_ISREG(lstat_result.st_mode)
                workspace_sort_rank = 2 if is_file else 1
                size = lstat_result.st_size if is_file else None
                mtime_ns = lstat_result.st_mtime_ns
                is_dir_entry = stat.S_ISDIR(lstat_result.st_mode)
            else:
                size = None
                mtime_ns = None
                is_dir_entry = False
                workspace_sort_rank = 1
            entries.append({
                'name': name,
                'path': entry_path,
                'type': 'dir' if is_dir_entry else 'file',
                'size': size,
                'mtime_ns': mtime_ns,
                'birthtime_ns': _birthtime_ns(lstat_result) if lstat_result is not None else None,
                'workspace_sort_rank': workspace_sort_rank,
            })

    if _DIR_FD_OK:
        # #3398 TOCTOU hardening (Linux/macOS): open the directory via an anchored
        # openat-walk (O_NOFOLLOW on every component) and enumerate via the verified
        # fd (os.scandir(fd) + fd-relative fstatat/readlinkat), so a path component
        # swapped to an escaping symlink after safe_resolve_ws() cannot redirect the
        # listing.
        def _sort_key_de(de):
            try:
                is_link = de.is_symlink()
            except OSError:
                is_link = False
            is_file = False
            if not is_link:
                try:
                    is_file = de.is_file()
                except OSError:
                    pass
            return (not is_link, is_file, de.name.lower())

        dir_fd = open_anchored_fd(workspace, target, want_dir=True)
        try:
            st = os.fstat(dir_fd)
            if not stat.S_ISDIR(st.st_mode):
                raise FileNotFoundError(f"Not a directory: {rel}")
            with os.scandir(dir_fd) as scan:
                scandir_entries = sorted(scan, key=_sort_key_de)
            for de in scandir_entries:
                name = de.name
                is_symlink = de.is_symlink()
                raw_link = None
                if is_symlink:
                    try:
                        raw_link = os.readlink(name, dir_fd=dir_fd)
                    except OSError:
                        raw_link = None
                try:
                    lst = os.stat(name, dir_fd=dir_fd, follow_symlinks=False)
                except OSError:
                    lst = None
                # reachable: follow-stat succeeds (filters ELOOP/broken symlinks).
                reachable = True
                if is_symlink:
                    try:
                        os.stat(name, dir_fd=dir_fd, follow_symlinks=True)
                    except OSError:
                        reachable = False
                _process(name, is_symlink, raw_link, lst, reachable)
                if len(entries) >= 200:
                    break
        finally:
            try:
                os.close(dir_fd)
            except OSError:
                pass
    else:
        # Portability fallback (Windows / no dir_fd): path-based enumeration after
        # safe_resolve_ws(). No anchored-fd race protection on these platforms, but
        # no regression vs the prior behaviour (creating symlinks on Windows needs
        # admin anyway), and safe_resolve_ws() still blocks the static escape.
        def _sort_key_p(p: Path):
            is_link = p.is_symlink()
            is_file = False
            if not is_link:
                try:
                    is_file = p.is_file()
                except OSError:
                    pass
            return (not is_link, is_file, p.name.lower())

        for item in sorted(target.iterdir(), key=_sort_key_p):
            name = item.name
            is_symlink = item.is_symlink()
            raw_link = None
            if is_symlink:
                try:
                    raw_link = os.readlink(str(item))
                except OSError:
                    raw_link = None
            try:
                lst = item.lstat()
            except OSError:
                lst = None
            # reachable: follow-stat succeeds (filters ELOOP/broken symlinks).
            reachable = True
            if is_symlink:
                try:
                    os.stat(str(item), follow_symlinks=True)
                except OSError:
                    reachable = False
            _process(name, is_symlink, raw_link, lst, reachable)
            if len(entries) >= 200:
                break
    return entries


def dir_signature(workspace: Path, rel: str = '.', entries: list[dict] | None = None) -> str:
    """Return a cheap, stable signature for a listed workspace directory.

    The signature is based only on bounded directory-entry metadata already used
    by the workspace tree: names, displayed paths, entry type, file sizes,
    mtimes, and symlink targets. It intentionally does not read file contents.
    """
    if entries is None:
        entries = list_dir(workspace, rel)
    payload = []
    for entry in entries:
        payload.append({
            'name': entry.get('name'),
            'path': entry.get('path'),
            'type': entry.get('type'),
            'is_dir': entry.get('is_dir'),
            'size': entry.get('size'),
            'mtime_ns': entry.get('mtime_ns'),
            'target': entry.get('target'),
            'target_outside_workspace': entry.get('target_outside_workspace'),
        })
    raw = json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
    return hashlib.sha256(raw.encode('utf-8')).hexdigest()


def read_file_content(workspace: Path, rel: str) -> dict:
    target = safe_resolve_ws(workspace, rel)
    if not target.is_file():
        raise FileNotFoundError(f"Not a file: {rel}")
    # #3398 TOCTOU hardening: open the resolved file via an anchored openat-walk
    # (O_NOFOLLOW on every component) so a path swapped to an escaping symlink
    # after safe_resolve_ws() cannot be followed, then read from the fd (not the
    # pathname) so the bytes returned are guaranteed to be the verified file.
    fd = open_anchored_fd(workspace, target, want_dir=False)
    with os.fdopen(fd, 'rb', closefd=True) as fh:
        st = os.fstat(fh.fileno())
        if not stat.S_ISREG(st.st_mode):
            raise FileNotFoundError(f"Not a file: {rel}")
        if st.st_size > MAX_FILE_BYTES:
            raise ValueError(f"File too large ({st.st_size} bytes, max {MAX_FILE_BYTES})")
        raw = fh.read(MAX_FILE_BYTES + 1)
    if Path(str(rel)).suffix.lower() in {".docx", ".xlsx", ".pptx"}:
        from api.office_documents import preview_office_document

        return preview_office_document(rel, raw)
    content = raw.decode('utf-8', errors='replace')
    return {'path': rel, 'content': content, 'size': len(raw), 'lines': content.count('\n') + 1}


def _normalize_workspace_rel_path(rel: str | Path) -> str:
    raw = _strip_surrounding_quotes(str(rel or "")).strip().replace("\\", "/")
    if not raw or raw == ".":
        return "."
    norm = posixpath.normpath(raw)
    if not norm or norm == ".":
        return "."
    if norm == ".." or norm.startswith("../") or norm.startswith("/"):
        raise ValueError(f"Path traversal blocked: {rel}")
    return norm


def _escape_virtual_path(root: str, rel: str) -> str:
    root_norm = _normalize_workspace_rel_path(root)
    rel_norm = _normalize_workspace_rel_path(rel)
    if root_norm == ".":
        return rel_norm
    if rel_norm == ".":
        return root_norm
    return f"{root_norm}/{rel_norm}"


def _escape_surface_target(workspace: Path, rel: str) -> tuple[Path, Path]:
    workspace_root = workspace.resolve()
    surface_rel = _normalize_workspace_rel_path(rel)
    surface_posix = PurePosixPath(surface_rel)
    parent_rel = str(surface_posix.parent)
    if parent_rel in ("", "."):
        parent_path = workspace_root
    else:
        parent_path = safe_resolve_ws(workspace_root, parent_rel)
    surface_path = parent_path / surface_posix.name
    if not surface_path.is_symlink():
        raise ValueError(f"Path is not an escape-target symlink: {rel}")
    target = surface_path.resolve()
    if not target.exists():
        raise ValueError(f"Path is no longer reachable: {rel}")
    try:
        target.relative_to(workspace_root)
    except ValueError:
        pass
    else:
        raise ValueError(f"Path does not escape workspace: {rel}")
    if _is_blocked_system_path(target):
        raise ValueError(f"Path points to a system directory: {target}")
    return surface_path, target


def _escape_authorized_root(target: Path) -> tuple[Path, str]:
    resolved_target = target.resolve()
    if resolved_target.is_dir():
        return resolved_target, "."
    return resolved_target.parent, resolved_target.name


class EscapeAuthorizationExpiredError(ValueError):
    pass


def _escape_prune_tokens(now: float | None = None) -> None:
    cutoff = time.time() if now is None else now
    expired = [token for token, record in _ESCAPE_AUTH_TOKENS.items() if float(record.get("expires_at") or 0.0) <= cutoff]
    for token in expired:
        _ESCAPE_AUTH_TOKENS.pop(token, None)


def authorize_escape_target(workspace: Path, session_id: str, rel: str) -> dict:
    """Mint a short-lived browser-only grant for one surfaced escape-target symlink."""
    workspace_root = workspace.resolve()
    _surface_path, target = _escape_surface_target(workspace_root, rel)
    external_root, external_entry_rel = _escape_authorized_root(target)
    token = secrets.token_urlsafe(24)
    expires_at = time.time() + _ESCAPE_AUTH_TTL_SECONDS
    record = {
        "session_id": str(session_id or ""),
        "workspace_root": str(workspace_root),
        "surface_path": _normalize_workspace_rel_path(rel),
        "external_root": str(external_root),
        "external_entry_rel": external_entry_rel,
        "surface_target": str(target),
        "expires_at": expires_at,
    }
    with _ESCAPE_AUTH_LOCK:
        _escape_prune_tokens()
        _ESCAPE_AUTH_TOKENS[token] = record
    return {
        "token": token,
        "path": record["surface_path"],
        "is_dir": target.is_dir(),
        "expires_at": expires_at,
        "expires_in": _ESCAPE_AUTH_TTL_SECONDS,
        "read_only": True,
    }


def _escape_authorization_record(workspace: Path, session_id: str, token: str) -> dict:
    workspace_root = str(workspace.resolve())
    token = str(token or "").strip()
    if not token:
        raise ValueError("Escape authorization token is required")
    now = time.time()
    with _ESCAPE_AUTH_LOCK:
        _escape_prune_tokens(now)
        record = dict(_ESCAPE_AUTH_TOKENS.get(token) or {})
    if not record:
        raise EscapeAuthorizationExpiredError("Escape authorization expired")
    if str(record.get("session_id") or "") != str(session_id or ""):
        raise EscapeAuthorizationExpiredError("Escape authorization expired")
    if str(record.get("workspace_root") or "") != workspace_root:
        raise EscapeAuthorizationExpiredError("Escape authorization expired")
    surface_path = _normalize_workspace_rel_path(record.get("surface_path") or ".")
    surface_target = str(record.get("surface_target") or "")
    try:
        _surface, current_target = _escape_surface_target(Path(workspace_root), surface_path)
    except ValueError:
        raise EscapeAuthorizationExpiredError("Escape authorization expired") from None
    if str(current_target.resolve()) != surface_target:
        raise EscapeAuthorizationExpiredError("Escape authorization expired")
    if not current_target.exists() or _is_blocked_system_path(current_target):
        raise EscapeAuthorizationExpiredError("Escape authorization expired")
    return record


def resolve_authorized_escape_request(workspace: Path, session_id: str, token: str, rel: str) -> dict:
    record = _escape_authorization_record(workspace, session_id, token)
    surface_path = _normalize_workspace_rel_path(record["surface_path"])
    request_path = _normalize_workspace_rel_path(rel)
    try:
        requested_rel = str(PurePosixPath(request_path).relative_to(PurePosixPath(surface_path)))
    except ValueError:
        raise ValueError(f"Path traversal blocked: {rel}") from None
    if not requested_rel:
        requested_rel = "."
    external_root = Path(str(record["external_root"]))
    external_entry_rel = _normalize_workspace_rel_path(record.get("external_entry_rel") or ".")
    if external_entry_rel == ".":
        external_rel = requested_rel
    elif requested_rel == ".":
        external_rel = external_entry_rel
    else:
        external_rel = str(PurePosixPath(external_entry_rel) / PurePosixPath(requested_rel))
    target = external_root / external_rel
    return {
        "record": record,
        "workspace_root": Path(str(record["workspace_root"])),
        "surface_path": surface_path,
        "request_path": request_path,
        "external_rel": external_rel,
        "external_root": external_root,
        "target": target,
    }


def list_authorized_escape_dir(workspace: Path, session_id: str, token: str, rel: str) -> dict:
    resolved = resolve_authorized_escape_request(workspace, session_id, token, rel)
    external_root = resolved["external_root"]
    external_rel = resolved["external_rel"]
    entries = list_dir(external_root, external_rel)
    surface_path = resolved["surface_path"]
    external_root_resolved = external_root.resolve()
    for entry in entries:
        entry["path"] = _escape_virtual_path(surface_path, entry.get("path") or ".")
        entry["escape_read_only"] = True
        target = entry.get("target")
        if not target:
            continue
        try:
            target_path = Path(str(target)).resolve()
            target_rel = target_path.relative_to(external_root_resolved).as_posix()
        except Exception:
            entry.pop("target", None)
            continue
        entry["target"] = _escape_virtual_path(surface_path, target_rel)
    return {
        "path": resolved["request_path"],
        "entries": entries,
        "signature": dir_signature(external_root, external_rel, entries),
        "virtual_root": surface_path,
        "read_only": True,
    }


def read_authorized_escape_file_content(workspace: Path, session_id: str, token: str, rel: str) -> dict:
    resolved = resolve_authorized_escape_request(workspace, session_id, token, rel)
    payload = read_file_content(resolved["external_root"], resolved["external_rel"])
    payload["path"] = resolved["request_path"]
    payload["escape_read_only"] = True
    return payload


def raw_authorized_escape_target(workspace: Path, session_id: str, token: str, rel: str) -> tuple[Path, Path]:
    resolved = resolve_authorized_escape_request(workspace, session_id, token, rel)
    target = safe_resolve_ws(resolved["external_root"], resolved["external_rel"])
    return resolved["external_root"], target


# ── Git detection ──────────────────────────────────────────────────────────

def _run_git(args, cwd, timeout=3):
    """Run a git command and return stdout, or None on failure."""
    try:
        r = subprocess.run(
            ['git'] + args, cwd=str(cwd), capture_output=True,
            text=True, timeout=timeout,
            creationflags=windows_hide_flags(),
        )
        return r.stdout.strip() if r.returncode == 0 else None
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        return None


def git_info_for_workspace(workspace: Path) -> dict:
    """Return git info for a workspace directory, or None if not a git repo."""
    if not (workspace / '.git').exists():
        return None
    branch = _run_git(['rev-parse', '--abbrev-ref', 'HEAD'], workspace)
    if branch is None:
        return None
    # Run the remaining git commands in parallel via threads — they are
    # independent subprocess calls and together can take 50-200ms when run
    # serially.  Threading is safe here because each call blocks only on the
    # subprocess pipe, not on the GIL.
    def _ahead():
        r = _run_git(['rev-list', '--count', '@{u}..HEAD'], workspace)
        return int(r) if r and r.isdigit() else 0
    def _behind():
        r = _run_git(['rev-list', '--count', 'HEAD..@{u}'], workspace)
        return int(r) if r and r.isdigit() else 0
    def _status():
        out = _run_git(['status', '--porcelain'], workspace) or ''
        lines = [l for l in out.splitlines() if l]
        modified = sum(1 for l in lines if len(l) >= 2 and (l[0] in 'MAR' or l[1] in 'MAR'))
        untracked = sum(1 for l in lines if l.startswith('??'))
        return len(lines), modified, untracked
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        f_status = pool.submit(_status)
        f_ahead  = pool.submit(_ahead)
        f_behind = pool.submit(_behind)
        dirty, modified, untracked = f_status.result()
        ahead  = f_ahead.result()
        behind = f_behind.result()
    return {
        'branch': branch,
        'dirty': dirty,
        'modified': modified,
        'untracked': untracked,
        'ahead': ahead,
        'behind': behind,
        'is_git': True,
    }