# OPENCLAW-WRAP: This module will be wrapped around OpenClaw. Preserve public API.

"""User identity resolution for Rex AI Assistant.

Provides session-scoped active user selection for scenarios where voice
speaker recognition is unavailable or uncertain.

The active user can be set via:
1. Explicit ``--user <id>`` flag on any command
2. ``rex identify --user <id>`` or interactive ``rex identify``
3. ``runtime.active_user`` in ``config/rex_config.json``
4. ``runtime.user_id`` in ``config/rex_config.json`` (legacy fallback)

Session state is stored in a temporary file under the OS-appropriate
app data directory.  It is cleared on ``rex identify --clear`` or when
the session file is deleted.
"""

from __future__ import annotations

import json
import logging
import os
import re
import tempfile
from datetime import UTC, datetime
from pathlib import Path

from .config import settings
from .runtime_paths import memory_dir as get_memory_dir

logger = logging.getLogger(__name__)

_USER_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
_WINDOWS_RESERVED_DEVICE_NAMES = {
    "con",
    "prn",
    "aux",
    "nul",
    "clock$",
    *(f"com{number}" for number in range(1, 10)),
    *(f"lpt{number}" for number in range(1, 10)),
}


def validate_user_id(user_id: str) -> str:
    """Validate and return a filesystem-safe user identifier.

    User IDs are used as directory names under ``Memory/``. Restricting the
    accepted characters and length prevents traversal through values such as
    ``..`` while retaining common IDs containing letters, digits, dots,
    underscores, and hyphens.

    Raises:
        ValueError: If *user_id* is empty, reserved, or filesystem-unsafe.
    """
    if not isinstance(user_id, str) or not _USER_ID_PATTERN.fullmatch(user_id):
        raise ValueError(f"Invalid user_id: {user_id!r}")
    if user_id in {".", ".."}:
        raise ValueError(f"Invalid user_id: {user_id!r}")
    # Windows strips trailing spaces/periods and treats the portion before an
    # extension as a device name. Apply that rule everywhere so a profile is
    # never portable on one OS but unsafe on another.
    normalized = user_id.rstrip(" .")
    device_stem = normalized.split(".", 1)[0].rstrip(" .").casefold()
    if device_stem in _WINDOWS_RESERVED_DEVICE_NAMES:
        raise ValueError(f"Invalid user_id: {user_id!r}")
    return user_id


def _validated_candidate(user_id: object, *, source: str) -> str | None:
    """Return a valid candidate ID or fail closed for persisted configuration."""
    if not isinstance(user_id, str) or not user_id:
        return None
    try:
        return validate_user_id(user_id)
    except ValueError:
        logger.warning("Ignoring invalid %s user ID: %r", source, user_id)
        return None


def _session_state_path() -> Path:
    """Return the path to the session state file."""
    if os.name == "nt":
        base = Path(os.environ.get("LOCALAPPDATA", str(Path.home() / "AppData" / "Local")))
    else:
        base = Path(
            os.environ.get("XDG_RUNTIME_DIR", os.environ.get("TMPDIR", "/tmp"))  # noqa: S108
        )
    return base / "rex-ai" / "session.json"


def _known_user_ids(*, memory_dir: Path | None = None) -> list[str]:
    """Discover known user IDs from canonical or test Memory storage."""
    root = memory_dir if memory_dir is not None else get_memory_dir()
    if not root.is_dir():
        return []
    users: list[str] = []
    for entry in sorted(root.iterdir()):
        if entry.is_dir() and (entry / "core.json").exists():
            if _validated_candidate(entry.name, source="Memory directory"):
                users.append(entry.name)
    return users


def _load_session() -> dict:
    """Load the current session state from disk."""
    path = _session_state_path()
    if path.exists():
        try:
            modified_at = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC)
            age = datetime.now(UTC) - modified_at
            if age.total_seconds() > settings.session_ttl_hours * 3600:
                path.unlink(missing_ok=True)
                return {}
            return json.loads(path.read_text(encoding="utf-8"))  # type: ignore[no-any-return]
        except json.JSONDecodeError as e:
            logger.warning("Corrupted session file %s, resetting: %s", path, e)
            return {}
        except Exception:
            return {}
    return {}


def _save_session(data: dict) -> None:
    """Persist session state to disk."""
    path = _session_state_path()
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
    except Exception as exc:
        logger.warning("Failed to save session state: %s", exc)


def get_session_user() -> str | None:
    """Return the active user from session state, or None."""
    session = _load_session()
    return _validated_candidate(session.get("active_user"), source="session")


def set_session_user(user_id: str) -> None:
    """Set the active user in session state."""
    user_id = validate_user_id(user_id)
    session = _load_session()
    session["active_user"] = user_id
    _save_session(session)
    logger.info("Session active user set to: %s", user_id)


def clear_session_user() -> None:
    """Clear the active user from session state."""
    session = _load_session()
    session.pop("active_user", None)
    _save_session(session)


def resolve_active_user(
    explicit_user: str | None = None,
    *,
    config: dict | None = None,
) -> str | None:
    """Resolve the active user through the priority chain.

    Priority:
    1. ``explicit_user`` (from ``--user`` flag)
    2. Session state (from ``rex identify``)
    3. ``runtime.active_user`` from config
    4. ``runtime.user_id`` from config

    Returns:
        User ID string, or ``None`` if no user could be resolved.
    """
    # 1. Explicit flag. Invalid explicit selection is a caller error and must
    # not silently fall back to another person's session.
    if explicit_user:
        return validate_user_id(explicit_user)

    # 2. Session state. Invalid persisted state fails closed.
    session_user = _load_session().get("active_user")
    if session_user:
        return _validated_candidate(session_user, source="session")

    # 3-4. Config values. Invalid configured identity fails closed rather than
    # unexpectedly selecting the legacy fallback.
    if config:
        runtime = config.get("runtime", {})
        active = runtime.get("active_user")
        if active:
            return _validated_candidate(active, source="runtime.active_user")
        uid = runtime.get("user_id")
        if uid and uid != "default":
            return _validated_candidate(uid, source="runtime.user_id")

    return None


def require_active_user(
    explicit_user: str | None = None,
    *,
    config: dict | None = None,
    action: str = "this command",
) -> str:
    """Resolve active user or raise an informative error.

    Args:
        explicit_user: User from ``--user`` flag.
        config: Config dict.
        action: Description of the action for error messages.

    Returns:
        Resolved user ID.

    Raises:
        SystemExit: If no user could be resolved.
    """
    user = resolve_active_user(explicit_user, config=config)
    if user:
        return user

    known = _known_user_ids()
    msg = f"Error: No active user for {action}.\n"
    if known:
        msg += f"Known users: {', '.join(known)}\n"
    msg += "Set one with: rex identify --user <id>\n"
    msg += "Or interactively: rex identify"
    raise SystemExit(msg)


def resolve_entrypoint_user_id(
    settings_obj: object | None = None,
    *,
    explicit_user: str | None = None,
) -> str:
    """Return the profile a first-party single-user entrypoint binds to.

    ``Assistant`` no longer invents an identity when none is supplied (issue
    #303), so entrypoints that intentionally serve one configured profile
    resolve it here — outside the assistant — and pass it explicitly to
    ``Assistant(user_id=...)``.

    Priority:
    1. ``explicit_user`` (e.g. a ``--user`` flag) — validated, caller errors
       do not fall back to another profile.
    2. The session/config active user (``rex identify`` chain).
    3. ``settings_obj.user_id`` (``runtime.user_id``) — validated.
    4. The explicit ``"default"`` profile.

    The final ``"default"`` is a deliberate selection of the profile named
    ``default`` by a trusted first-party entrypoint, not an automatic
    assistant-side fallback.

    Raises:
        ValueError: If an explicit or configured user ID fails validation.
    """
    if explicit_user:
        return validate_user_id(explicit_user)

    session_user = resolve_active_user()
    if session_user:
        return session_user

    configured = getattr(settings_obj, "user_id", None) if settings_obj is not None else None
    if isinstance(configured, str) and configured:
        return validate_user_id(configured)

    return "default"


def list_known_users(*, memory_dir: Path | None = None) -> list[dict]:
    """Return known users from canonical or test Memory storage."""
    root = memory_dir if memory_dir is not None else get_memory_dir()
    users: list[dict] = []
    if not root.is_dir():
        return users
    for entry in sorted(root.iterdir()):
        core = entry / "core.json"
        if not entry.is_dir() or not core.exists():
            continue
        if not _validated_candidate(entry.name, source="Memory directory"):
            continue
        try:
            data = json.loads(core.read_text(encoding="utf-8"))
            if not isinstance(data, dict):
                raise ValueError("profile JSON is not an object")
            raw_name = data.get("name", entry.name)
            raw_role = data.get("role", "")
            name = raw_name if isinstance(raw_name, str) and raw_name else entry.name
            role = raw_role if isinstance(raw_role, str) else ""
            users.append({"id": entry.name, "name": name, "role": role})
        except Exception:
            users.append({"id": entry.name, "name": entry.name, "role": ""})
    return users


def _atomic_write_json(path: Path, data: dict) -> None:
    """Atomically write one JSON object beside its destination."""
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp",
            delete=False,
        ) as handle:
            temporary = Path(handle.name)
            json.dump(data, handle, indent=2, ensure_ascii=False)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        temporary = None
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def create_user_profile(
    user_id: str,
    name: str,
    role: str = "",
    preferences: dict | None = None,
    *,
    memory_dir: Path | None = None,
    overwrite: bool = False,
) -> Path:
    """Create a user profile in canonical or test Memory storage."""
    user_id = validate_user_id(user_id)
    base = memory_dir if memory_dir is not None else get_memory_dir()
    core_path = base / user_id / "core.json"
    if core_path.exists() and not overwrite:
        raise FileExistsError(f"User profile already exists: {core_path}")
    now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
    data: dict = {
        "name": name,
        "role": role,
        "user": user_id,
        "preferences": preferences or {},
        "created_at": now,
        "last_updated": now,
    }
    _atomic_write_json(core_path, data)
    logger.info("Created user profile for %s", user_id)
    return core_path


def get_user_profile(
    user_id: str,
    *,
    memory_dir: Path | None = None,
) -> dict | None:
    """Load a user profile object from canonical or test Memory storage."""
    user_id = validate_user_id(user_id)
    base = memory_dir if memory_dir is not None else get_memory_dir()
    core_path = base / user_id / "core.json"
    if not core_path.exists():
        return None
    try:
        data = json.loads(core_path.read_text(encoding="utf-8"))
        if not isinstance(data, dict):
            logger.warning("Profile for %s is not a JSON object", user_id)
            return None
        return data
    except Exception as exc:
        logger.warning("Failed to load profile for %s: %s", user_id, exc)
        return None


def update_user_preferences(
    user_id: str,
    preferences: dict,
    *,
    memory_dir: Path | None = None,
) -> bool:
    """Merge preferences into an existing profile with atomic persistence."""
    user_id = validate_user_id(user_id)
    base = memory_dir if memory_dir is not None else get_memory_dir()
    core_path = base / user_id / "core.json"
    if not core_path.exists():
        return False
    try:
        data = json.loads(core_path.read_text(encoding="utf-8"))
        if not isinstance(data, dict):
            return False
        existing_prefs = data.get("preferences", {})
        if not isinstance(existing_prefs, dict):
            existing_prefs = {}
        merged = dict(existing_prefs)
        merged.update(preferences)
        data["preferences"] = merged
        data["last_updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
        _atomic_write_json(core_path, data)
        logger.info("Updated preferences for user %s", user_id)
        return True
    except Exception as exc:
        logger.warning("Failed to update preferences for %s: %s", user_id, exc)
        return False


__all__ = [
    "clear_session_user",
    "create_user_profile",
    "get_session_user",
    "get_user_profile",
    "list_known_users",
    "require_active_user",
    "resolve_active_user",
    "resolve_entrypoint_user_id",
    "set_session_user",
    "update_user_preferences",
    "validate_user_id",
]