CSV Editor

Instructions

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