restore_to_operation

Restore CSV editing sessions to a specific operation point using session and operation IDs. Revert changes or continue work from a saved state in the CSV Editor.

Instructions

Restore session data to a specific operation point.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`session_id`	Yes
`operation_id`	Yes

Implementation Reference

src/csv_editor/server.py:526-532 (handler)

MCP tool handler for 'restore_to_operation', decorated with @mcp.tool and delegates to the implementation in history_operations.py

async def restore_to_operation(
    session_id: str,
    operation_id: str,
    ctx: Context = None
) -> Dict[str, Any]:
    """Restore session data to a specific operation point."""
    return await _restore_to_operation(session_id, operation_id, ctx)

src/csv_editor/tools/history_operations.py:186-244 (handler)

Main implementation of restore_to_operation tool logic: retrieves session, calls session.restore_to_operation, handles errors and logging, returns wrapped result

async def restore_to_operation(
    session_id: str,
    operation_id: str,
    ctx: Context = None
) -> Dict[str, Any]:
    """
    Restore session data to a specific operation point.
    
    Args:
        session_id: Session identifier
        operation_id: Operation ID to restore to
        ctx: FastMCP context
        
    Returns:
        Dict with success status and restore result
    """
    try:
        manager = get_session_manager()
        session = manager.get_session(session_id)
        
        if not session:
            return OperationResult(
                success=False,
                message="Session not found",
                error=f"No session with ID: {session_id}"
            ).model_dump()
        
        if ctx:
            await ctx.info(f"Restoring session {session_id} to operation {operation_id}")
        
        result = await session.restore_to_operation(operation_id)
        
        if result["success"]:
            if ctx:
                await ctx.info(f"Successfully restored to operation {operation_id}")
            
            return OperationResult(
                success=True,
                message=result["message"],
                session_id=session_id,
                data=result
            ).model_dump()
        else:
            return OperationResult(
                success=False,
                message="Failed to restore to operation",
                error=result.get("error")
            ).model_dump()
            
    except Exception as e:
        logger.error(f"Error restoring to operation: {str(e)}")
        if ctx:
            await ctx.error(f"Failed to restore to operation: {str(e)}")
        return OperationResult(
            success=False,
            message="Failed to restore to operation",
            error=str(e)
        ).model_dump()

src/csv_editor/models/csv_session.py:307-336 (helper)

CSVSession method that calls history_manager to restore data snapshot and updates session dataframe

async def restore_to_operation(self, operation_id: str) -> Dict[str, Any]:
    """Restore data to a specific operation point."""
    if not self.history_manager:
        return {"success": False, "error": "History is not enabled"}
    
    try:
        data_snapshot = self.history_manager.restore_to_operation(operation_id)
        
        if data_snapshot is not None:
            self.df = data_snapshot
            
            # Trigger auto-save if configured
            if self.auto_save_manager.should_save_after_operation():
                await self.auto_save_manager.trigger_save(self._save_callback, "restore")
            
            return {
                "success": True,
                "message": f"Restored to operation {operation_id}",
                "shape": self.df.shape
            }
        else:
            return {
                "success": False,
                "error": f"Could not restore to operation {operation_id}"
            }
            
    except Exception as e:
        logger.error(f"Error during restore: {str(e)}")
        return {"success": False, "error": str(e)}

src/csv_editor/models/history_manager.py:323-355 (helper)

Core restore logic: finds operation index, locates nearest prior data snapshot, updates current_index and redo stack, returns snapshot DataFrame

def restore_to_operation(self, operation_id: str) -> Optional[pd.DataFrame]:
    """Restore data to a specific operation point."""
    # Find the operation
    target_index = None
    for i, entry in enumerate(self.history):
        if entry.operation_id == operation_id:
            target_index = i
            break
    
    if target_index is None:
        logger.error(f"Operation {operation_id} not found")
        return None
    
    # Find the nearest snapshot at or before target
    snapshot = None
    for i in range(target_index, -1, -1):
        if self.history[i].data_snapshot is not None:
            snapshot = self.history[i].data_snapshot.copy()
            self.current_index = target_index
            
            # Clear redo stack since we're jumping to a specific point
            self.redo_stack.clear()
            
            # Save state
            if self.storage_type != HistoryStorage.MEMORY:
                self._save_history()
            
            logger.info(f"Restored to operation {operation_id}")
            return snapshot
    
    logger.error(f"No snapshot available for operation {operation_id}")
    return None

CSV Editor