Function prune_checkpoints

core/checkpoint.py:277–305

Remove old checkpoints, keeping only the most recent ones. Args: keep_count: Number of recent checkpoints to keep

(keep_count: int = 5)

def prune_checkpoints(keep_count: int = 5):
    """
    Remove old checkpoints, keeping only the most recent ones.
    
    Args:
        keep_count: Number of recent checkpoints to keep
    """
    state = get_state_store()
    checkpoints = state.get_checkpoints(100)  # Get all
    
    if len(checkpoints) <= keep_count:
        return
    
    to_remove = checkpoints[keep_count:]
    
    for cp in to_remove:
        checkpoint_id = cp["id"]
        backup_dir = CHECKPOINT_DIR / checkpoint_id
        
        # Remove backup directory
        if backup_dir.exists():
            shutil.rmtree(backup_dir)
        
        # Remove from database
        state.delete_checkpoint(checkpoint_id)
        
        info(f"Checkpoint: pruned '{checkpoint_id}'")
    
    success(f"Checkpoint: pruned {len(to_remove)} old checkpoints")


# State store extensions for checkpoints