Function rollback

core/checkpoint.py:190–242

Rollback to a checkpoint. Args: checkpoint_id: Checkpoint ID to rollback to Returns: True if rollback successful

(checkpoint_id: str)

def rollback(checkpoint_id: str) -> bool:
    """
    Rollback to a checkpoint.
    
    Args:
        checkpoint_id: Checkpoint ID to rollback to
        
    Returns:
        True if rollback successful
    """
    backup_dir = CHECKPOINT_DIR / checkpoint_id
    
    if not backup_dir.exists():
        error(f"Rollback: checkpoint '{checkpoint_id}' not found")
        return False
    
    info(f"Rollback: restoring from '{checkpoint_id}'")
    
    # Restore files from backup
    restored = 0
    for backup_file in backup_dir.rglob("*"):
        if backup_file.is_file():
            try:
                rel_path = backup_file.relative_to(backup_dir)
                dest = CODE_DIR / rel_path
                dest.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(backup_file, dest)
                restored += 1
            except Exception as e:
                error(f"Rollback: could not restore {backup_file}: {e}")
    
    # Try to git checkout the commit
    state = get_state_store()
    checkpoint_data = state.get_checkpoint(checkpoint_id)
    
    if checkpoint_data and checkpoint_data.get("git_commit_hash"):
        git_hash = checkpoint_data["git_commit_hash"]
        try:
            subprocess.run(
                ["git", "checkout", git_hash],
                cwd=CODE_DIR,
                capture_output=True
            )
            info(f"Rollback: checked out git commit {git_hash[:8]}")
        except Exception as e:
            warning(f"Rollback: git checkout failed: {e}")
    
    success(f"Rollback: restored {restored} files from checkpoint '{checkpoint_id}'")
    
    # Log in episodic memory
    state.log_action("rollback", f"Restored from checkpoint {checkpoint_id}")
    
    return True


def list_checkpoints(limit: int = 10) -> List[Dict]: