Raise warning if instance limit is reached

qiskit_ibm_runtime/qiskit_runtime_service.py

@@ -17,6 +17,7 @@
 from datetime import datetime
 from typing import Any
 from collections.abc import Callable, Sequence
+from urllib.parse import quote
 
 from qiskit.providers.backend import BackendV2 as Backend
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
@@ -974,6 +975,7 @@ def _run(
             IBMRuntimeError: An error occurred running the program.
         """
 
+        self._check_instance_usage()
         qrt_options: RuntimeOptions = options  # type: ignore[assignment]
         if options is None:
             qrt_options = RuntimeOptions()
@@ -1172,6 +1174,26 @@ def usage(self) -> dict[str, Any]:
             usage_dict["usage_remaining_seconds"] = usage_remaining
         return usage_dict
 
+    def _check_instance_usage(self) -> None:
+        """Emit a warning if instance usage has been reached."""
+        usage_dict = self.usage()
+
+        if usage_dict.get("usage_limit_reached"):
+            if usage_dict.get("usage_limit_seconds") and usage_dict["usage_remaining_seconds"] <= 0:
+                warnings.warn(
+                    "This instance has met its usage limit. Workloads will not run until time is made "
+                    "available. Check "
+                    f"https://quantum.cloud.ibm.com/instances/{quote(self.active_instance(), safe='')} "
+                    "for more details."
+                )
+            else:
+                warnings.warn(
+                    "There is currently no more time available for this instance's plan on the account. "
+                    "Workloads will not run until time is made available. Check "
+                    f"https://quantum.cloud.ibm.com/instances/{quote(self.active_instance(), safe='')} "
+                    "for more details."
+                )
+
     def _decode_job(self, raw_data: dict) -> RuntimeJobV2:
         """Decode job data received from the server.
 

release-notes/unreleased/2458.feat.rst

@@ -0,0 +1,2 @@
+When a job is submitted, there will now be warnings if the instance being used
+has reached its limit.

test/unit/mock/fake_runtime_client.py

@@ -451,6 +451,21 @@ def session_details(self, session_id: str) -> dict[str, Any]:
         """Return the details of the session."""
         return {"id": session_id, "mode": "dedicated", "backend_name": "common_backend"}
 
+    def cloud_usage(self) -> dict[str, Any]:
+        """Return cloud instance usage information."""
+        return {
+            "instance_id": "instance_id",
+            "plan_id": "plan_id",
+            "usage_consumed_seconds": 6000,
+            "usage_period": {
+                "start_time": "2025-10-01T17:40:06.269Z",
+                "end_time": "2025-10-29T17:40:06.269Z",
+            },
+            "usage_allocation_seconds": 90000,
+            "usage_remaining_seconds": 84000,
+            "usage_limit_reached": False,
+        }
+
     def _find_backend(self, backend_name):
         for back in self._backends:
             if back.name == backend_name:

test/unit/test_jobs.py

@@ -14,6 +14,7 @@
 
 import random
 import time
+from unittest.mock import patch
 
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 
@@ -28,6 +29,7 @@
     FailedRuntimeJob,
     FailedRanTooLongRuntimeJob,
     CancelableRuntimeJob,
+    BaseFakeRuntimeClient,
 )
 from ..ibm_test_case import IBMTestCase
 from ..decorators import run_cloud_fake
@@ -144,3 +146,36 @@ def test_wait_for_final_state(self, service):
         with mock_wait_for_final_state(service, job):
             job.wait_for_final_state()
         self.assertEqual("DONE", job.status())
+
+    @run_cloud_fake
+    def test_instance_limit_warning(self, service):
+        """Test emitting a warning if instance usage has been reached."""
+        # All relevant fields present, account limit reached.
+        instance_usage_msg_1 = {
+            "usage_consumed_seconds": 1,
+            "usage_limit_seconds": 2,
+            "usage_limit_reached": True,
+        }
+        # All relevant fields present, instance limit reached.
+        instance_usage_msg_2 = {
+            "usage_consumed_seconds": 3,
+            "usage_limit_seconds": 2,
+            "usage_limit_reached": True,
+        }
+        # Missing `usage_limit_seconds`, account limit reached.
+        instance_usage_msg_3 = {
+            "usage_consumed_seconds": 1,
+            "usage_limit_reached": True,
+        }
+
+        with patch.object(BaseFakeRuntimeClient, "cloud_usage", return_value=instance_usage_msg_1):
+            with self.assertWarnsRegex(UserWarning, r"There is currently no more time available"):
+                run_program(service=service)
+
+        with patch.object(BaseFakeRuntimeClient, "cloud_usage", return_value=instance_usage_msg_2):
+            with self.assertWarnsRegex(UserWarning, r"This instance has met its usage limit"):
+                run_program(service=service)
+
+        with patch.object(BaseFakeRuntimeClient, "cloud_usage", return_value=instance_usage_msg_3):
+            with self.assertWarnsRegex(UserWarning, r"There is currently no more time available"):
+                run_program(service=service)