Raise an error if dynamical decoupling is used with dynamic circuits (#…

…1593)

* Raising an error if dynamical decoupling is used with dynamic circuits

* linting

* linting

* Change since mypy doesn't like the current way

* linting

* Fixes based on code review

* Linting

* Update qiskit_ibm_runtime/utils/utils.py

Co-authored-by: Jessie Yu <jessieyu@us.ibm.com>

* Returning the `str` check as it crashes tests for `backend.run`

* Change `are_circuits_dynamic` behaviour on QASM inputs

---------

Co-authored-by: ptristan3 <44805021+ptristan3@users.noreply.github.com>
Co-authored-by: Jessie Yu <jessieyu@us.ibm.com>
Co-authored-by: Kevin Tian <kevin.tian@ibm.com>

qiskit_ibm_runtime/base_primitive.py

@@ -35,7 +35,7 @@
 from .ibm_backend import IBMBackend
 from .utils.default_session import get_cm_session
 from .utils.deprecation import issue_deprecation_msg
-from .utils.utils import validate_isa_circuits, is_simulator
+from .utils.utils import validate_isa_circuits, is_simulator, validate_no_dd_with_dynamic_circuits
 from .constants import DEFAULT_DECODERS
 from .qiskit_runtime_service import QiskitRuntimeService
 from .fake_provider.local_service import QiskitRuntimeLocalService
@@ -148,6 +148,7 @@ def _run(self, pubs: Union[list[EstimatorPub], list[SamplerPub]]) -> RuntimeJobV
         primitive_inputs.update(primitive_options)
         runtime_options = self._options_class._get_runtime_options(options_dict)
 
+        validate_no_dd_with_dynamic_circuits([pub.circuit for pub in pubs], self.options)
         if self._backend:
             for pub in pubs:
                 if getattr(self._backend, "target", None) and not is_simulator(self._backend):

qiskit_ibm_runtime/utils/__init__.py

@@ -40,8 +40,14 @@
     local_to_utc,
     seconds_to_duration,
     duration_difference,
+)
+from .utils import (
+    to_python_identifier,
+    is_crn,
+    get_runtime_api_base_url,
+    resolve_crn,
     are_circuits_dynamic,
+    validate_no_dd_with_dynamic_circuits,
 )
-from .utils import to_python_identifier, is_crn, get_runtime_api_base_url, resolve_crn
 from .json import RuntimeEncoder, RuntimeDecoder, to_base64_string
 from . import pubsub

qiskit_ibm_runtime/utils/converters.py

@@ -15,10 +15,9 @@
 import re
 from datetime import datetime, timedelta, timezone
 from math import ceil
-from typing import Union, Tuple, Any, Optional, List
+from typing import Union, Tuple, Any, Optional
 
 from dateutil import tz, parser
-from qiskit.circuit import QuantumCircuit, ControlFlowOp
 from qiskit_ibm_runtime.exceptions import IBMInputValueError
 
 
@@ -223,17 +222,3 @@ def hms_to_seconds(hms: str, msg_prefix: str = "") -> int:
         raise IBMInputValueError(f"{msg_prefix} Invalid input: {parsed_time}")
 
     return total_seconds
-
-
-def are_circuits_dynamic(circuits: List[QuantumCircuit]) -> bool:
-    """Checks if the input circuits are dynamic."""
-    for circuit in circuits:
-        if isinstance(circuit, str):
-            return True
-        for inst in circuit:
-            if (
-                isinstance(inst.operation, ControlFlowOp)
-                or getattr(inst.operation, "condition", None) is not None
-            ):
-                return True
-    return False

qiskit_ibm_runtime/utils/utils.py

@@ -28,7 +28,7 @@
     IAMAuthenticator,
 )
 from ibm_platform_services import ResourceControllerV2  # pylint: disable=import-error
-from qiskit.circuit import QuantumCircuit
+from qiskit.circuit import QuantumCircuit, ControlFlowOp
 from qiskit.transpiler import Target
 from qiskit.providers.backend import BackendV1, BackendV2
 from qiskit_ibm_runtime.exceptions import IBMInputValueError
@@ -99,6 +99,36 @@ def validate_isa_circuits(circuits: Sequence[QuantumCircuit], target: Target) ->
             )
 
 
+def are_circuits_dynamic(circuits: List[QuantumCircuit]) -> bool:
+    """Checks if the input circuits are dynamic."""
+    for circuit in circuits:
+        if isinstance(circuit, str):
+            return False  # currently do not verify QASM inputs
+        for inst in circuit:
+            if (
+                isinstance(inst.operation, ControlFlowOp)
+                or getattr(inst.operation, "condition", None) is not None
+            ):
+                return True
+    return False
+
+
+def validate_no_dd_with_dynamic_circuits(circuits: List[QuantumCircuit], options: Any) -> None:
+    """Validate that if dynamical decoupling options are enabled,
+    no circuit in the pubs is dynamic
+
+    Args:
+        circuits: A list of QuantumCircuits.
+        options: The runtime options
+    """
+    if not hasattr(options, "dynamical_decoupling") or not options.dynamical_decoupling.enable:
+        return
+    if are_circuits_dynamic(circuits):
+        raise IBMInputValueError(
+            "Dynamical decoupling currently cannot be used with dynamic circuits"
+        )
+
+
 def validate_job_tags(job_tags: Optional[List[str]]) -> None:
     """Validates input job tags.
 

test/unit/test_estimator.py

@@ -20,7 +20,7 @@
 from qiskit.quantum_info import SparsePauliOp, Pauli
 from qiskit.primitives.containers.estimator_pub import EstimatorPub
 
-from qiskit_ibm_runtime import Estimator, Session, EstimatorV2, EstimatorOptions
+from qiskit_ibm_runtime import Estimator, Session, EstimatorV2, EstimatorOptions, IBMInputValueError
 
 from .mock.fake_runtime_service import FakeRuntimeService
 from ..ibm_test_case import IBMTestCase
@@ -258,3 +258,22 @@ def test_invalid_basis(self):
             with self.subTest(obs=obs):
                 with self.assertRaises(ValueError):
                     estimator.run((circuit, obs))
+
+    def test_unsupported_dynamical_decoupling_with_dynamic_circuits(self):
+        """Test that running on dynamic circuits with dynamical decoupling enabled is not allowed"""
+        dynamic_circuit = QuantumCircuit(3, 1)
+        dynamic_circuit.h(0)
+        dynamic_circuit.measure(0, 0)
+        dynamic_circuit.if_else(
+            (0, True), QuantumCircuit(3, 1), QuantumCircuit(3, 1), [0, 1, 2], [0]
+        )
+
+        in_pubs = [(dynamic_circuit, ["XXX"])]
+        backend = get_mocked_backend()
+        inst = EstimatorV2(backend=backend)
+        inst.options.dynamical_decoupling.enable = True
+        with self.assertRaisesRegex(
+            IBMInputValueError,
+            "Dynamical decoupling currently cannot be used with dynamic circuits",
+        ):
+            inst.run(in_pubs)

test/unit/test_sampler.py

@@ -20,7 +20,7 @@
 from qiskit import QuantumCircuit, transpile
 from qiskit.primitives.containers.sampler_pub import SamplerPub
 from qiskit.circuit.library import RealAmplitudes
-from qiskit_ibm_runtime import Sampler, Session, SamplerV2, SamplerOptions
+from qiskit_ibm_runtime import Sampler, Session, SamplerV2, SamplerOptions, IBMInputValueError
 
 from ..ibm_test_case import IBMTestCase
 from ..utils import bell, MockSession, dict_paritally_equal, get_mocked_backend, transpile_pubs
@@ -99,6 +99,25 @@ def test_unsupported_values_for_sampler_options(self, opt):
                 inst.options.update(**opt)
             self.assertIn(list(opt.keys())[0], str(exc.exception))
 
+    def test_unsupported_dynamical_decoupling_with_dynamic_circuits(self):
+        """Test that running on dynamic circuits with dynamical decoupling enabled is not allowed"""
+        dynamic_circuit = QuantumCircuit(3, 1)
+        dynamic_circuit.h(0)
+        dynamic_circuit.measure(0, 0)
+        dynamic_circuit.if_else(
+            (0, True), QuantumCircuit(3, 1), QuantumCircuit(3, 1), [0, 1, 2], [0]
+        )
+
+        in_pubs = [(dynamic_circuit,)]
+        backend = get_mocked_backend()
+        inst = SamplerV2(backend=backend)
+        inst.options.dynamical_decoupling.enable = True
+        with self.assertRaisesRegex(
+            IBMInputValueError,
+            "Dynamical decoupling currently cannot be used with dynamic circuits",
+        ):
+            inst.run(in_pubs)
+
     def test_run_default_options(self):
         """Test run using default options."""
         session = MagicMock(spec=MockSession, _backend="common_backend")