kb-skills/gsd-openclaw/src/gsd_openclaw/user_context.py

"""
Per-user context management for gsd-openclaw.

Provides UserContextManager for tracking user project context across chat sessions,
supporting the /gsd:use command requirement.
"""
from __future__ import annotations

import dataclasses
import json
import os
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional


@dataclass
class UserContext:
    """Represents the persistent context for a user.

    Tracks the user's current project, recent projects, and preferences
    for persistence across chat sessions.

    Attributes:
        schema_version: Version of the context schema (default "1.0")
        user_id: Unique identifier for the user
        current_project: The user's currently active project
        recent_projects: List of recently used projects (max 5)
        last_active_at: ISO format timestamp of last activity
        preferences: Optional user preferences dictionary
    """
    schema_version: str = "1.0"
    user_id: str = ""
    current_project: str = ""
    recent_projects: List[str] = field(default_factory=list)
    last_active_at: str = ""
    preferences: Dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> Dict[str, Any]:
        """Convert UserContext to a dictionary.

        Returns:
            Dictionary representation of the user context.
        """
        return dataclasses.asdict(self)

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> UserContext:
        """Create UserContext from a dictionary.

        Unknown fields in the dictionary are ignored.

        Args:
            data: Dictionary containing user context data.

        Returns:
            UserContext instance.
        """
        # Get field names from the dataclass
        field_names = {f.name for f in dataclasses.fields(cls)}
        # Filter to only known fields
        filtered_data = {k: v for k, v in data.items() if k in field_names}
        return cls(**filtered_data)


class UserContextManager:
    """Manager class for per-user context persistence.

    Provides a clean interface for loading and saving user context,
    with atomic writes and multi-user isolation.

    Attributes:
        USERS_ROOT: Default root directory for all user contexts.
        user_id: Unique identifier for the user.
        users_root: Root directory for user contexts (configurable for testing).
        context_path: Full path to the user's context.json file.
    """

    USERS_ROOT = Path.home() / "gsd-workspaces" / ".users"

    def __init__(self, user_id: str, users_root: Optional[Path] = None):
        """Initialize UserContextManager for a user.

        Args:
            user_id: Unique identifier for the user.
            users_root: Optional custom root for user contexts (useful for testing).
                        Defaults to ~/gsd-workspaces/.users.
        """
        self.user_id = user_id
        self.users_root = Path(users_root) if users_root else self.USERS_ROOT
        self._context_path: Optional[Path] = None

    @property
    def context_path(self) -> Path:
        """Get the path to the user's context file.

        Returns:
            Path to context.json file.
        """
        return self.users_root / self.user_id / "context.json"

    def load(self) -> Optional[UserContext]:
        """Load user context from file with backup fallback.

        Returns:
            UserContext if loaded successfully, None if no context file exists.
        """
        context_path = self.context_path
        backup_path = context_path.with_suffix('.json.bak')

        def _load_from_path(path: Path) -> UserContext:
            """Internal: Load context from a specific path."""
            with open(path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            return UserContext.from_dict(data)

        # Try main file first, then backup
        paths_to_try = []
        if context_path.exists():
            paths_to_try.append(context_path)
        if backup_path.exists() and backup_path != context_path:
            paths_to_try.append(backup_path)

        if not paths_to_try:
            return None

        last_error = None
        for path in paths_to_try:
            try:
                return _load_from_path(path)
            except json.JSONDecodeError as e:
                last_error = e
                continue  # Try next path (backup)
            except (KeyError, TypeError) as e:
                last_error = e
                continue  # Try next path (backup)

        # If we get here, all paths failed - return None instead of raising
        return None

    def save(self, context: UserContext) -> None:
        """Save user context to file atomically with backup rotation.

        This function implements crash-safe context persistence:
        1. Creates parent directories if they don't exist
        2. Rotates existing context file to backup (context.json -> context.json.bak)
        3. Writes to a temporary file first
        4. Atomically renames temp file to final path

        Args:
            context: The UserContext to save.

        Raises:
            OSError: If file operations fail (permissions, disk full, etc.)
        """
        context_path = self.context_path
        backup_path = context_path.with_suffix('.json.bak')

        # 1. Create parent directories if missing
        context_path.parent.mkdir(parents=True, exist_ok=True)

        # 2. Rotate backup: if context.json exists, move it to context.json.bak
        if context_path.exists():
            # Remove old backup if exists
            if backup_path.exists():
                backup_path.unlink()
            # Rename current context to backup
            context_path.rename(backup_path)

        # 3. Write to temp file in same directory (for atomic rename)
        fd, temp_path = tempfile.mkstemp(dir=context_path.parent, suffix='.tmp')
        try:
            with os.fdopen(fd, 'w', encoding='utf-8') as f:
                json.dump(context.to_dict(), f, indent=2, ensure_ascii=False)
            # 4. Atomic replace
            os.replace(temp_path, context_path)
        except Exception:
            # Clean up temp file on failure
            if os.path.exists(temp_path):
                os.unlink(temp_path)
            raise

    def get_current_project(self) -> Optional[str]:
        """Get the user's current project.

        Returns:
            Current project name or None if no context exists.
        """
        context = self.load()
        if context is None or not context.current_project:
            return None
        return context.current_project

    def set_current_project(self, project: str) -> None:
        """Set the user's current project.

        Creates default context if none exists. Updates current_project,
        adds to recent_projects (max 5), and updates last_active_at.

        Args:
            project: Name of the project to set as current.
        """
        context = self.load()
        if context is None:
            context = self.create_default_context()

        # Update recent projects
        recent = context.recent_projects.copy()
        # Remove project if already in list (will be added at front)
        if project in recent:
            recent.remove(project)
        # Add to front of list
        recent.insert(0, project)
        # Keep only last 5
        context.recent_projects = recent[:5]

        # Update current project and timestamp
        context.current_project = project
        context.last_active_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

        self.save(context)

    def get_recent_projects(self) -> List[str]:
        """Get the user's recent projects list.

        Returns:
            List of recent project names (may be empty).
        """
        context = self.load()
        if context is None:
            return []
        return context.recent_projects

    def create_default_context(self) -> UserContext:
        """Create a new default context for the user.

        Returns:
            New UserContext with default values.
        """
        return UserContext(
            schema_version="1.0",
            user_id=self.user_id,
            current_project="",
            recent_projects=[],
            last_active_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            preferences={}
        )