orchestrata_ultimate.session.recovery

Reference › orchestrata_ultimate › session › recovery

RecoveryManager implements the IRecoveryManager interface from interfaces on top of the CheckpointManager. create_checkpoint captures the state of a session, recover restores it from the latest checkpoint and returns the state as a dictionary, rollback restores a specific checkpoint (or returns nothing if it does not exist), list_checkpoints lists them, can_recover says whether any exist, and cleanup_checkpoints prunes old ones, keeping five by default. It is distinct from recover_from_git in ContextFileManager, which rebuilds lost context history from git.

At a glance

Class diagram

Dependency diagram

API

Classes

RecoveryManager

Bases: IRecoveryManager · Decorators: traces(spec='006', story='US-005', reqs=['REQ-006-US-005-001', 'REQ-006-FR6-001']) · Line 25

Manages session recovery from checkpoints.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/recovery.py

"""Session recovery manager.

Handles session recovery from checkpoints after failures.
"""

import logging

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.checkpoint import CheckpointManager
from orchestrata_ultimate.session.interfaces import IRecoveryManager
from orchestrata_ultimate.session.redis_store import RedisSessionStore
from orchestrata_ultimate.session.types import Checkpoint, SessionState

logger = logging.getLogger(__name__)


@traces(
    spec="006",
    story="US-005",
    reqs=[
        "REQ-006-US-005-001",  # Session context management
        "REQ-006-FR6-001",  # Error handling / recovery
    ],
)
class RecoveryManager(IRecoveryManager):
    """Manages session recovery from checkpoints."""

    def __init__(self, redis_store: RedisSessionStore) -> None:
        self._redis = redis_store
        self._checkpoint_mgr = CheckpointManager(redis_store)

    def create_checkpoint(self, session_id: str) -> Checkpoint:
        """Create a recovery checkpoint for the current session state."""
        return self._checkpoint_mgr.create(session_id)

    def recover(self, session_id: str) -> dict:
        """Recover session from the latest checkpoint.

        Returns:
            Recovered session data dict.

        Raises:
            ValueError: If no checkpoint found.
        """
        latest = self._checkpoint_mgr.get_latest(session_id)
        if latest is None:
            raise ValueError(f"No checkpoint found for session {session_id}")

        # Mark session as recovering
        session_data = self._redis.get_session(session_id) or {}
        session_data["status"] = SessionState.RECOVERING.value
        session_data["id"] = session_id
        self._redis.save_session(session_data)

        # Restore from checkpoint
        restored = self._checkpoint_mgr.restore(latest.id, session_id)

        # Mark session as active after recovery
        if restored:
            session_data = self._redis.get_session(session_id) or {}
            session_data["status"] = SessionState.ACTIVE.value
            session_data["id"] = session_id
            self._redis.save_session(session_data)
            self._redis.set_active(session_id)

        logger.info("Recovered session %s from checkpoint %s", session_id, latest.id)
        return session_data

    def list_checkpoints(self, session_id: str) -> list[Checkpoint]:
        """List all checkpoints for a session."""
        return self._checkpoint_mgr.list(session_id)

    def rollback(self, session_id: str, checkpoint_id: str) -> dict | None:
        """Rollback to a specific checkpoint."""
        restored = self._checkpoint_mgr.restore(checkpoint_id, session_id)
        if restored:
            session_data = self._redis.get_session(session_id) or {}
            session_data["status"] = SessionState.ACTIVE.value
            session_data["id"] = session_id
            self._redis.save_session(session_data)
            self._redis.set_active(session_id)
            logger.info(
                "Rolled back session %s to checkpoint %s", session_id, checkpoint_id
            )
        return restored

    def can_recover(self, session_id: str) -> bool:
        """Check if a session has any checkpoints for recovery."""
        latest = self._checkpoint_mgr.get_latest(session_id)
        return latest is not None

    def cleanup_checkpoints(self, session_id: str, keep_last: int = 5) -> int:
        """Prune old checkpoints."""
        return self._checkpoint_mgr.prune(session_id, keep_last)