Use POST /sessions in backend.run sessions

qiskit_ibm_runtime/ibm_backend.py

@@ -756,16 +756,11 @@ def _runtime_run(
             warnings.warn(
                 "A Primitive session is open but Backend.run() jobs will not be run within this session"
             )
+        session_id = None
         if self._session:
             if not self._session.active:
                 raise RuntimeError(f"The session {self._session.session_id} is closed.")
             session_id = self._session.session_id
-            start_session = session_id is None
-            max_session_time = self._session._max_time
-        else:
-            session_id = None
-            start_session = False
-            max_session_time = None
 
         log_level = getattr(self.options, "log_level", None)  # temporary
         try:
@@ -777,15 +772,11 @@ def _runtime_run(
                 log_level=log_level,
                 job_tags=job_tags,
                 session_id=session_id,
-                start_session=start_session,
-                session_time=max_session_time,
+                start_session=False,
                 image=image,
             )
         except RequestsApiError as ex:
             raise IBMBackendApiError("Error submitting job: {}".format(str(ex))) from ex
-        session_id = response.get("session_id", None)
-        if self._session:
-            self._session._session_id = session_id
         try:
             job = RuntimeJob(
                 backend=self,
@@ -831,7 +822,13 @@ def _get_run_config(self, program_id: str, **kwargs: Any) -> Dict:
 
     def open_session(self, max_time: Optional[Union[int, str]] = None) -> ProviderSession:
         """Open session"""
-        self._session = ProviderSession(max_time=max_time)
+        if not self._configuration.simulator:
+            new_session = self._service._api_client.create_session(
+                self.name, self._instance, max_time, self._service.channel
+            )
+            self._session = ProviderSession(max_time=max_time, session_id=new_session.get("id"))
+        else:
+            self._session = ProviderSession()
         return self._session
 
     @property

qiskit_ibm_runtime/provider_session.py

@@ -57,6 +57,7 @@ class Session:
     def __init__(
         self,
         max_time: Optional[Union[int, str]] = None,
+        session_id: Optional[str] = None,
     ):
         """Session constructor.
 
@@ -72,7 +73,7 @@ def __init__(
             ValueError: If an input value is invalid.
         """
         self._instance = None
-        self._session_id: Optional[str] = None
+        self._session_id = session_id
         self._active = True
 
         self._max_time = (

releasenotes/notes/sessions-endpoint-e617b028fe4a68c4.yaml

@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    Sessions started with :meth:`qiskit_ibm_runtime.IBMBackend.open_session` will use the 
+    new ``/sessions`` endpoint. 
+
+    The sessions functionality will not change but note that ``backend.run()`` sessions 
+    prior to this release will no longer be supported after March 31, 2024. 
+    Please update your ``qiskit-ibm-runtime`` version as soon as possible before this date. 

test/integration/test_session.py

@@ -113,9 +113,7 @@ def test_session_id(self):
         self.assertEqual(backend.session.session_id, None)
         self.assertTrue(backend.session.active)
         job1 = backend.run(bell())
-        self.assertEqual(job1._session_id, job1.job_id())
-        job2 = backend.run(bell())
-        self.assertFalse(job2._session_id == job2.job_id())
+        self.assertEqual(job1._session_id, None)
 
     def test_backend_run_with_session(self):
         """Test that 'shots' parameter is transferred correctly"""
@@ -163,32 +161,15 @@ def test_session_close(self):
         backend.close_session()
         self.assertIsNone(backend.session)
 
-    def test_run_after_cancel(self):
-        """Test running after session is cancelled."""
-        backend = self.service.backend("ibmq_qasm_simulator")
-        job1 = backend.run(circuits=bell())
-        self.assertIsNone(backend.session)
-        self.assertIsNone(job1._session_id)
-
-        backend.open_session()
-        job2 = backend.run(bell())
-        self.assertIsNotNone(job2._session_id)
-        backend.cancel_session()
-
-        job3 = backend.run(circuits=bell())
-        self.assertIsNone(backend.session)
-        self.assertIsNone(job3._session_id)
-
     def test_session_as_context_manager(self):
         """Test session as a context manager"""
         backend = self.service.backend("ibmq_qasm_simulator")
 
         with backend.open_session() as session:
             job1 = backend.run(bell())
             session_id = session.session_id
-            self.assertEqual(session_id, job1.job_id())
-            job2 = backend.run(bell())
-            self.assertFalse(session_id == job2.job_id())
+            self.assertTrue(job1.result())
+            self.assertIsNone(session_id)
 
     def test_run_after_cancel_as_context_manager(self):
         """Test run after cancel in context manager"""