Fix lightcone pass mid circuit measurements

qiskit/transpiler/passes/optimization/light_cone.py

@@ -11,14 +11,15 @@
 # that they have been altered from the originals.
 
 """Cancel the redundant (self-adjoint) gates through commutation relations."""
+
 from __future__ import annotations
 import warnings
 from qiskit.circuit import Gate, Qubit
 from qiskit.circuit.commutation_library import SessionCommutationChecker as scc
-from qiskit.circuit.library import PauliGate, ZGate
+from qiskit.circuit.library import PauliGate, ZGate, PauliProductMeasurement
 from qiskit.dagcircuit import DAGCircuit
 from qiskit.transpiler.basepasses import TransformationPass
-from qiskit.transpiler.passes.utils.remove_final_measurements import calc_final_ops
+from qiskit.quantum_info import Pauli
 
 translation_table = str.maketrans({"+": "X", "-": "X", "l": "Y", "r": "Y", "0": "Z", "1": "Z"})
 
@@ -54,12 +55,18 @@ def __init__(self, bit_terms: str | None = None, indices: list[int] | None = Non
         self.indices = indices
 
     @staticmethod
-    def _find_measurement_qubits(dag: DAGCircuit) -> set[Qubit]:
-        final_nodes = calc_final_ops(dag, {"measure"})
-        qubits_measured = set()
-        for node in final_nodes:
-            qubits_measured |= set(node.qargs)
-        return qubits_measured
+    def _measurement_commutation_ops(node) -> list[tuple]:
+        """Return the list of (measurement_gate, qubits) for a measurement node.
+        For a standard measure, we use one ZGate per qubit.
+        For a PauliProductMeasurement, we use a PauliGate.
+        """
+        if node.op.name == "measure":
+            return [(ZGate(), [qubit]) for qubit in node.qargs]
+        elif isinstance(node.op, PauliProductMeasurement):
+            pauli_label = Pauli((node.op._pauli_z, node.op._pauli_x, 0)).to_label()
+            return [(PauliGate(pauli_label), list(node.qargs))]
+
+        raise ValueError("_measurement_commutation_ops called with unexpected operation")
 
     def _get_initial_lightcone(
         self, dag: DAGCircuit
@@ -69,14 +76,18 @@ def _get_initial_lightcone(
         If a `bit_terms` is provided, the qubits corresponding to the
         non-trivial Paulis define the light-cone.
         """
-        lightcone_qubits = self._find_measurement_qubits(dag)
         if self.bit_terms is None:
-            lightcone_operations = [(ZGate(), [qubit_index]) for qubit_index in lightcone_qubits]
+            # will be discovered dynamically during the reverse traversal in run()
+            return set(), []
         else:
             # Having both measurements and an observable is not allowed
+            all_measured_qubits = set()
+            for node in dag.topological_op_nodes():
+                if node.op.name == "measure" or isinstance(node.op, PauliProductMeasurement):
+                    all_measured_qubits.update(node.qargs)
             if len(dag.qubits) < max(self.indices) + 1:
                 raise ValueError("`indices` contains values outside the qubit range.")
-            if lightcone_qubits:
+            if all_measured_qubits:
                 raise ValueError(
                     "The circuit contains measurements and an observable has been given: "
                     "remove the observable or the measurements."
@@ -105,6 +116,15 @@ def run(self, dag: DAGCircuit) -> DAGCircuit:
 
         # Iterate over the nodes in reverse topological order
         for node in dag.topological_op_nodes(reverse=True):
+            # Each time we encounter a measurement, extend the cone with its
+            # qubits and add the appropriate measurement gate.
+            if node.op.name in ("measure", "pauli_product_measurement"):
+                new_qubits = set(node.qargs) - lightcone_qubits
+                if new_qubits:
+                    lightcone_qubits.update(new_qubits)
+                    lightcone_operations.extend(self._measurement_commutation_ops(node))
+                new_dag.apply_operation_front(node.op, node.qargs, node.cargs)
+                continue
             # Check if the node belongs to the light-cone
             if lightcone_qubits.intersection(node.qargs):
                 # Check commutation with all previous operations

releasenotes/notes/fix-lightcone-mid-circuit-measurment-87f5c72c34d34978.yaml

@@ -0,0 +1,10 @@
+---
+fixes:
+  - |
+    Fixed :class:`.LightCone` transpiler pass incorrectly returning an empty
+    circuit when mid-circuit measurements are present. The pass was using
+    ``calc_final_ops`` to find measured qubits, which only detects measurements
+    in final position in the DAG. Measurements are now discovered dynamically
+    during the reverse-topological review, handling mid-circuit measurements.
+
+    See `#15937 <https://github.com/Qiskit/qiskit/issues/15937>`__.

test/python/transpiler/test_light_cone.py

@@ -21,10 +21,14 @@
     Parameter,
     QuantumCircuit,
 )
-from qiskit.circuit.library import real_amplitudes
+from qiskit.circuit.library import (
+    real_amplitudes,
+    PauliProductRotationGate,
+    PauliProductMeasurement,
+)
 from qiskit.circuit.library.n_local.efficient_su2 import efficient_su2
 from qiskit.converters import circuit_to_dag
-from qiskit.quantum_info import SparsePauliOp, SparseObservable
+from qiskit.quantum_info import SparsePauliOp, SparseObservable, Pauli
 from qiskit.transpiler.passes.optimization.light_cone import LightCone
 from qiskit.transpiler.passmanager import PassManager
 
@@ -354,6 +358,68 @@ def test_raise_error_when_circuit_measurements_and_observable_present(self):
         ):
             light_cone.run(dag)
 
+    def test_mid_circuit_measurement_not_empty(self):
+        """Test that LightCone does not return an empty circuit when
+        mid-circuit measurements are present."""
+        light_cone = LightCone()
+        pm = PassManager([light_cone])
+
+        qc = QuantumCircuit(2, 1)
+        qc.cx(0, 1)
+        qc.z(0)
+        qc.measure(0, 0)
+        qc.h(0)
+
+        new_circuit = pm.run(qc)
+
+        expected = QuantumCircuit(2, 1)
+        expected.measure(0, 0)
+
+        self.assertEqual(expected, new_circuit)
+
+    def test_pauli_product_measurement_circuit(self):
+        """Test LightCone on a circuit containing only PauliProductRotationGates
+        and PauliProductMeasurements. The XZ rotation does not commute with the
+        ZZ measurement observable and must be kept."""
+        light_cone = LightCone()
+        pm = PassManager([light_cone])
+
+        qc = QuantumCircuit(2, 1)
+        qc.append(PauliProductRotationGate(Pauli("XZ"), 0.5), [0, 1])
+        qc.append(PauliProductMeasurement(Pauli("ZZ")), [0, 1], [0])
+
+        new_circuit = pm.run(qc)
+
+        expected = QuantumCircuit(2, 1)
+        expected.append(PauliProductRotationGate(Pauli("XZ"), 0.5), [0, 1])
+        expected.append(PauliProductMeasurement(Pauli("ZZ")), [0, 1], [0])
+
+        self.assertEqual(expected, new_circuit)
+
+    def test_mid_circuit_and_final_measurements(self):
+        """Test LightCone on a circuit with both mid-circuit and final measurements.
+        Both measurements must be kept along with the gates in their light cones.
+        The X gate after the mid-circuit measurement is irrelevant and must be removed."""
+        light_cone = LightCone()
+        pm = PassManager([light_cone])
+
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.measure(0, 0)
+        qc.x(0)
+        qc.measure(1, 1)
+
+        new_circuit = pm.run(qc)
+
+        expected = QuantumCircuit(2, 2)
+        expected.h(0)
+        expected.cx(0, 1)
+        expected.measure(0, 0)
+        expected.measure(1, 1)
+
+        self.assertEqual(expected, new_circuit)
+
 
 if __name__ == "__main__":
     unittest.main()