""" State persistence layer for gsd-openclaw. Provides atomic state saves with backup rotation for crash-safe session checkpointing. """ from __future__ import annotations import dataclasses import json import os import tempfile from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional @dataclass class SessionState: """Represents the persistent state of a GSD session. Attributes: schema_version: Version of the state schema (default "1.0") session_id: Unique identifier for this session project_name: Name of the project tmux_session_name: Name of the tmux session tmux_pane_id: ID of the tmux pane created_at: ISO format timestamp when session was created last_active_at: ISO format timestamp of last activity command_count: Number of commands executed in this session last_output_snapshot: Last captured output from the session display_name: User-friendly display name for the session """ schema_version: str = "1.0" session_id: str = "" project_name: str = "" tmux_session_name: str = "" tmux_pane_id: str = "" created_at: str = "" last_active_at: str = "" command_count: int = 0 last_output_snapshot: Optional[str] = None display_name: Optional[str] = None def to_dict(self) -> dict[str, Any]: """Convert SessionState to a dictionary. Returns: Dictionary representation of the session state. """ return dataclasses.asdict(self) @classmethod def from_dict(cls, data: dict[str, Any]) -> SessionState: """Create SessionState from a dictionary. Unknown fields in the dictionary are ignored. Args: data: Dictionary containing session state data. Returns: SessionState 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) # Placeholder functions - to be implemented in Task 2 and 3 def save_state(state: SessionState, state_path: Path) -> None: """Save session state to file atomically with backup rotation. This function implements crash-safe state persistence: 1. Creates parent directories if they don't exist 2. Rotates existing state file to backup (state.json -> state.json.bak) 3. Writes to a temporary file first 4. Atomically renames temp file to final path The atomic rename ensures that: - Readers never see a partially written file - Crash during write leaves either old or new file intact Args: state: The SessionState to save. state_path: Path where state.json should be written. Raises: OSError: If file operations fail (permissions, disk full, etc.) """ state_path = Path(state_path) backup_path = state_path.with_suffix('.json.bak') # 1. Create parent directories if missing state_path.parent.mkdir(parents=True, exist_ok=True) # 2. Rotate backup: if state.json exists, move it to state.json.bak if state_path.exists(): # Remove old backup if exists if backup_path.exists(): backup_path.unlink() # Rename current state to backup state_path.rename(backup_path) # 3. Write to temp file in same directory (for atomic rename) fd, temp_path = tempfile.mkstemp(dir=state_path.parent, suffix='.tmp') try: with os.fdopen(fd, 'w', encoding='utf-8') as f: json.dump(state.to_dict(), f, indent=2, ensure_ascii=False) # 4. Atomic replace os.replace(temp_path, state_path) except Exception: # Clean up temp file on failure if os.path.exists(temp_path): os.unlink(temp_path) raise def load_state(state_path: Path) -> Optional[SessionState]: """Load session state from file with validation and backup fallback. This function implements robust state loading: 1. If state.json doesn't exist, try state.json.bak 2. If neither exists, return None 3. Read and parse JSON with utf-8 encoding 4. Validate schema version (must be "1.0") 5. Validate required fields (session_id, project_name, tmux_session_name) 6. On JSONDecodeError or ValueError, fall back to backup Args: state_path: Path to state.json file. Returns: SessionState if loaded successfully, None if no state file exists. Raises: ValueError: If schema version is unsupported or required fields are missing (only raised if backup also fails). """ state_path = Path(state_path) backup_path = state_path.with_suffix('.json.bak') def _validate_and_load(path: Path) -> SessionState: """Internal: Load and validate state from a specific path.""" with open(path, 'r', encoding='utf-8') as f: data = json.load(f) # Validate schema version schema_version = data.get('schema_version', '') if schema_version != '1.0': raise ValueError(f"Unsupported schema version: {schema_version}") # Validate required fields required_fields = ['session_id', 'project_name', 'tmux_session_name'] for field in required_fields: if field not in data or not data[field]: raise ValueError(f"Missing required field: {field}") return SessionState.from_dict(data) # Try main file first, then backup paths_to_try = [] if state_path.exists(): paths_to_try.append(state_path) if backup_path.exists() and backup_path != state_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 _validate_and_load(path) except json.JSONDecodeError as e: last_error = e continue # Try next path (backup) except ValueError as e: last_error = e continue # Try next path (backup) # If we get here, all paths failed if last_error: raise last_error return None class StateManager: """Manager class for state persistence operations. Provides a clean interface for state persistence without requiring callers to manage paths directly. Encapsulates workspace path resolution and delegates to save_state/load_state functions. Attributes: WORKSPACES_ROOT: Default root directory for all workspaces. project_name: Name of the project. workspaces_root: Root directory for workspaces (configurable for testing). workspace_path: Full path to the project's workspace. state_path: Full path to the project's state.json file. """ WORKSPACES_ROOT = Path.home() / "gsd-workspaces" def __init__(self, project_name: str, workspaces_root: Optional[Path] = None): """Initialize StateManager for a project. Args: project_name: Name of the project (used for workspace directory). workspaces_root: Optional custom root for workspaces (useful for testing). Defaults to ~/gsd-workspaces. """ self.project_name = project_name self.workspaces_root = Path(workspaces_root) if workspaces_root else self.WORKSPACES_ROOT self.workspace_path = self.workspaces_root / project_name self.state_path = self.workspace_path / ".gsd" / "state.json" def save(self, state: SessionState) -> None: """Save session state to the project's state file. Args: state: The SessionState to save. Raises: OSError: If file operations fail. """ save_state(state, self.state_path) def load(self) -> Optional[SessionState]: """Load session state from the project's state file. Returns: SessionState if loaded successfully, None if no state file exists. Raises: ValueError: If schema version is unsupported or required fields are missing. """ return load_state(self.state_path) def get_state_path(self) -> Path: """Get the path to the project's state file. Returns: Path to state.json file. """ return self.state_path