/state-management-patterns

name: state-management-patterns
description: "JSON persistence, atomic writes, file locking, crash recovery, and state versioning patterns. Use when implementing stateful libraries or features requiring persistent state. TRIGGER when: state persistence, atomic write, file locking, crash recovery, checkpoint. DO NOT TRIGGER when: stateless utilities, pure functions, config reads, documentation."
allowed-tools: [Read]

State Management Patterns Skill

Standardized state management and persistence patterns for the autonomous-dev plugin ecosystem. Ensures reliable, crash-resistant state persistence across Claude restarts and system failures.

When This Skill Activates

• Implementing state persistence
• Managing crash recovery
• Handling concurrent state access
• Versioning state schemas
• Tracking batch operations
• Managing user preferences
• Keywords: "state", "persistence", "JSON", "atomic", "crash recovery", "checkpoint"

---

Core Patterns

1. JSON Persistence with Atomic Writes

**Definition**: Store state in JSON files with atomic writes to prevent corruption on crash.

**Pattern**:

import json
from pathlib import Path
from typing import Dict, Any
import tempfile
import os

def save_state_atomic(state: Dict[str, Any], state_file: Path) -> None:
    """Save state with atomic write to prevent corruption.

    Args:
        state: State dictionary to persist
        state_file: Target state file path

    Security:
        - Atomic Write: Prevents partial writes on crash
        - Temp File: Write to temp, then rename (atomic operation)
        - Permissions: Preserves file permissions
    """
    # Write to temporary file first
    temp_fd, temp_path = tempfile.mkstemp(
        dir=state_file.parent,
        prefix=f".{state_file.name}.",
        suffix=".tmp"
    )

    try:
        # Write JSON to temp file
        with os.fdopen(temp_fd, 'w') as f:
            json.dump(state, f, indent=2)

        # Atomic rename (overwrites target)
        os.replace(temp_path, state_file)

    except Exception:
        # Clean up temp file on failure
        if Path(temp_path).exists():
            Path(temp_path).unlink()
        raise

**See**: `docs/json-persistence.md`, `examples/batch-state-example.py`

---

2. File Locking for Concurrent Access

**Definition**: Use file locks to prevent concurrent modification of state files.

**Pattern**:

import fcntl
import json
from pathlib import Path
from contextlib import contextmanager

@contextmanager
def file_lock(filepath: Path):
    """Acquire exclusive file lock for state file.

    Args:
        filepath: Path to file to lock

    Yields:
        Open file handle with exclusive lock

    Example:
        >>> with file_lock(state_file) as f:
        ...     state = json.load(f)
        ...     state['count'] += 1
        ...     f.seek(0)
        ...     f.truncate()
        ...     json.dump(state, f)
    """
    with filepath.open('r+') as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            yield f
        finally:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)

**See**: `docs/file-locking.md`, `templates/file-lock-template.py`

---

3. Crash Recovery Pattern

**Definition**: Design state to enable recovery after crashes or interruptions.

**Principles**:

• State includes enough context to resume operations
• Progress tracking enables "resume from last checkpoint"
• State validation detects corruption
• Migration paths handle schema changes

**Example**:

@dataclass
class BatchState:
    """Batch processing state with crash recovery support.

    Attributes:
        batch_id: Unique batch identifier
        features: List of all features to process
        current_index: Index of current feature
        completed: List of completed feature names
        failed: List of failed feature names
        created_at: State creation timestamp
        last_updated: Last update timestamp
    """
    batch_id: str
    features: List[str]
    current_index: int = 0
    completed: List[str] = None
    failed: List[str] = None
    created_at: str = None
    last_updated: str = None

    def __post_init__(self):
        if self.completed is None:
            self.completed = []
        if self.failed is None:
            self.failed = []
        if self.created_at is None:
            self.created_at = datetime.now().isoformat()
        self.last_updated = datetime.now().isoformat()

**See**: `docs/crash-recovery.md`, `examples/crash-recovery-example.py`

---

4. State Versioning and Migration

**Definition**: Version state schemas to enable graceful upgrades.

**Pattern**:

STATE_VERSION = "2.0.0"

def migrate_state(state: Dict[str, Any]) -> Dict[str, Any]:
    """Migrate state from old version to current.

    Args:
        state: State dictionary (any version)

    Returns:
        Migrated state (current version)
    """
    version = state.get("version", "1.0.0")

    if version == "1.0.0":
        # Migrate 1.0.0 → 1.1.0
        state = _migrate_1_0_to_1_1(state)
        version = "1.1.0"

    if version == "1.1.0":
        # Migrate 1.1.0 → 2.0.0
        state = _migrate_1_1_to_2_0(state)
        version = "2.0.0"

    state["version"] = STATE_VERSION
    return state

**See**: `docs/state-versioning.md`, `templates/state-manager-template.py`

---

Real-World Examples

BatchStateManager Pattern

From `plugins/autonomous-dev/lib/batch_state_manager.py`:

**Features**:

• JSON persistence with atomic writes
• Crash recovery via --resume flag
• Progress tracking (completed/failed features)
• Automatic context management via Claude Code (200K token budget)
• State versioning for schema upgrades

**Note** (Issue #218): Deprecated context clearing functions (`should_clear_context()`, `pause_batch_for_clear()`, `get_clear_notification_message()`) have been removed as Claude Code v2.0+ handles context automatically with 200K token budget.

**U