Fix box duration validation to track delays per qubit timeline

CHANGELOG.md

@@ -25,6 +25,7 @@ Types of changes:
 ### Removed
 
 ### Fixed
+- Fixed `box` duration validation summing `delay` durations across all qubits instead of tracking each qubit's timeline. Delays on disjoint qubits run in parallel, so `box[300ns] { delay[200ns] q[0]; delay[200ns] q[1]; }` was rejected while the identical schedule written as a broadcast delay (`delay[200ns] q;`) was accepted. Delays are now accumulated per qubit and the box is validated against the busiest single timeline; the error message names the offending qubit. Nested boxes now also contribute their declared duration to the enclosing box's timelines (previously the accumulator was reset when an inner box closed, dropping all inner delay accounting). ([#330](https://github.com/qBraid/pyqasm/pull/330))
 - Fixed `reset` on a physical qubit rewriting the operand to the internal pulse register, e.g. `reset $2;` unrolled to `reset __PYQASM_QUBITS__[2];`. That names a register the program never declares, so the unrolled output did not round-trip through `dumps()`/`loads()`, and the qubit was never registered (a program whose only operation was `reset $3;` reported `num_qubits == 0`). Physical qubits are now kept as-is in plain QASM programs, matching how gate and measurement operands already treat them; the rename is still applied for OpenPulse programs, where the pulse visitor expects it. ([#325](https://github.com/qBraid/pyqasm/pull/325))
 - Fixed `measure` and `reset` on a user register whose name merely starts with the reserved internal register name (e.g. `__PYQASM_QUBITS__foo`) being mistaken for the internal register itself. Such statements were short-circuited out of unrolling and emitted verbatim, so `c = measure __PYQASM_QUBITS__foo;` was never expanded per qubit and `reset __PYQASM_QUBITS__foo;` silently reset nothing. The register is now matched on its exact name, or on the name followed by an index or slice. ([#325](https://github.com/qBraid/pyqasm/pull/325))
 - Fixed the `c4x` (4-controlled X) gate, which previously raised `TypeError: c4x_gate() takes 4 positional arguments but 5 were given` because it was declared with four parameters for a five-qubit gate. It is now implemented via qiskit's structured `rc3x`/`c3sx`/`cphaseshift` decomposition. ([#320](https://github.com/qBraid/pyqasm/pull/320))

src/pyqasm/analyzer.py

@@ -251,7 +251,9 @@ def extract_qasm_version(qasm: str) -> float:  # type: ignore[return]
         raise_qasm3_error("Could not determine the OpenQASM version.", err_type=QasmParsingError)
 
     @staticmethod
-    def extract_duplicate_qubit(qubit_list: list[IndexedIdentifier | Identifier]):
+    def extract_duplicate_qubit(
+        qubit_list: list[IndexedIdentifier | Identifier],
+    ) -> tuple[str, int] | None:
         """
         Extracts the duplicate qubit from a list of qubits.
 
@@ -261,21 +263,31 @@ def extract_duplicate_qubit(qubit_list: list[IndexedIdentifier | Identifier]):
         Returns:
             tuple(string, int): The duplicate qubit name and id.
         """
-        qubit_set = set()
+        qubit_set: set[tuple[str, int]] = set()
         for qubit in qubit_list:
-            if isinstance(qubit, Identifier):
-                # Physical qubit: name is "$n", identity is the name itself.
-                qubit_name = qubit.name
-                qubit_id = int(qubit.name[1:])
-            else:
-                assert isinstance(qubit, IndexedIdentifier)
-                qubit_name = qubit.name.name
-                qubit_id = qubit.indices[0][0].value  # type: ignore
-            if (qubit_name, qubit_id) in qubit_set:
-                return (qubit_name, qubit_id)
-            qubit_set.add((qubit_name, qubit_id))
+            qubit_key = Qasm3Analyzer.extract_qubit_key(qubit)
+            if qubit_key in qubit_set:
+                return qubit_key
+            qubit_set.add(qubit_key)
         return None
 
+    @staticmethod
+    def extract_qubit_key(qubit: IndexedIdentifier | Identifier) -> tuple[str, int]:
+        """
+        Extract the (register name, index) identity key for a qubit operand.
+
+        Args:
+            qubit (IndexedIdentifier | Identifier): The qubit operand.
+
+        Returns:
+            tuple(string, int): The qubit register name and index.
+        """
+        if isinstance(qubit, Identifier):
+            # Physical qubit: name is "$n", identity is the name itself.
+            return (qubit.name, int(qubit.name[1:]))
+        assert isinstance(qubit, IndexedIdentifier)
+        return (qubit.name.name, qubit.indices[0][0].value)  # type: ignore
+
     @staticmethod
     def verify_gate_qubits(gate: QuantumGate, span: Optional[Span] = None):
         """

src/pyqasm/visitor.py

@@ -138,7 +138,11 @@ def __init__(  # pylint: disable=too-many-arguments
         self._in_generic_gate_op_scope: int = 0
         self._qubit_register_offsets: OrderedDict = OrderedDict()
         self._qubit_register_max_offset = 0
-        self._total_delay_duration_in_box = 0
+        # Stack of per-box delay accounting frames. Each frame maps a qubit
+        # key (name, index) to the summed delay durations on that qubit's
+        # timeline within the box. Delays on disjoint qubits run in parallel,
+        # so box durations are validated against the per-qubit maximum.
+        self._box_delay_frames: list[dict[tuple[str, int], float]] = []
         self._in_extern_function: bool = False
         self._openpulse_qubit_map: dict[str, set[str]] = {}
         self._total_pulse_qubits: int = 0
@@ -2872,9 +2876,6 @@ def _visit_delay_statement(
                 duration_val, unit=self._resolve_duration_unit(_delay_time_var)
             )
 
-        if self._scope_manager.in_box_scope():
-            self._total_delay_duration_in_box += duration_val
-
         if statement.qubits is not None:
             _is_delay_frame = False
             for qubit in statement.qubits:
@@ -2901,6 +2902,21 @@ def _visit_delay_statement(
                 span=statement.span,
             )
 
+        if self._box_delay_frames and duration_val:
+            # Record this delay on each targeted qubit's timeline. A delay
+            # with no qubit args applies to every qubit in scope.
+            if delay_qubit_bits:
+                delay_keys = [Qasm3Analyzer.extract_qubit_key(bit) for bit in delay_qubit_bits]
+            else:
+                delay_keys = [
+                    (reg_name, reg_id)
+                    for reg_name, reg_size in self._global_qreg_size_map.items()
+                    for reg_id in range(reg_size)
+                ]
+            box_delay_frame = self._box_delay_frames[-1]
+            for key in delay_keys:
+                box_delay_frame[key] = box_delay_frame.get(key, 0) + duration_val
+
         if self._check_only:
             return []
 
@@ -2944,6 +2960,7 @@ def _visit_box_statement(self, statement: qasm3_ast.Box) -> list[qasm3_ast.State
         self._scope_manager.push_scope({})
         self._scope_manager.increment_scope_level()
         self._scope_manager.push_context(Context.BOX)
+        self._box_delay_frames.append({})
 
         if statement.body:
             statements.extend(
@@ -2960,19 +2977,30 @@ def _visit_box_statement(self, statement: qasm3_ast.Box) -> list[qasm3_ast.State
         self._scope_manager.decrement_scope_level()
         self._scope_manager.pop_scope()
 
-        if (
-            _box_time_var
-            and box_duration_val
-            and self._total_delay_duration_in_box > box_duration_val
-        ):
-            time_unit = self._resolve_duration_unit(_box_time_var).name
-            raise_qasm3_error(
-                f"Total delay duration value '{self._total_delay_duration_in_box}{time_unit}' "
-                f"should be less than 'box[{box_duration_val}{time_unit}]' duration.",
-                error_node=statement,
-                span=statement.span,
-            )
-        self._total_delay_duration_in_box = 0
+        delay_frame = self._box_delay_frames.pop()
+        if _box_time_var and box_duration_val and delay_frame:
+            # Delays on different qubits run in parallel, so the box only
+            # needs to fit the busiest single qubit timeline.
+            max_qubit_key = max(delay_frame, key=lambda k: delay_frame[k])
+            max_delay_duration = delay_frame[max_qubit_key]
+            if max_delay_duration > box_duration_val:
+                time_unit = self._resolve_duration_unit(_box_time_var).name
+                qubit_name, qubit_id = max_qubit_key
+                raise_qasm3_error(
+                    f"Total delay duration value '{max_delay_duration}{time_unit}' "
+                    f"on qubit '{qubit_name}[{qubit_id}]' should be less than "
+                    f"'box[{box_duration_val}{time_unit}]' duration.",
+                    error_node=statement,
+                    span=statement.span,
+                )
+        if self._box_delay_frames and delay_frame:
+            # Nested box: contribute this box's time to the enclosing box's
+            # per-qubit timelines. If the box declares a duration, that is
+            # the time it occupies; otherwise use each qubit's delay total.
+            parent_frame = self._box_delay_frames[-1]
+            for key, qubit_total in delay_frame.items():
+                contribution = box_duration_val if box_duration_val else qubit_total
+                parent_frame[key] = parent_frame.get(key, 0) + contribution
 
         if self._check_only:
             return []

tests/qasm3/test_box.py

@@ -223,7 +223,8 @@ def test_box_dt_unit_preserved():
               measure q;
             }
             """,
-            r"Total delay duration value '20.0ns' should be less than 'box[10.0ns]' duration.",
+            r"Total delay duration value '20.0ns' on qubit 'q[0]' "
+            r"should be less than 'box[10.0ns]' duration.",
             r"Error at line 6, column 12",
         ),
         (
@@ -238,7 +239,8 @@ def test_box_dt_unit_preserved():
               measure q;
             }
             """,
-            r"Total delay duration value '20.0dt' should be less than 'box[10.0dt]' duration.",
+            r"Total delay duration value '20.0dt' on qubit 'q[0]' "
+            r"should be less than 'box[10.0dt]' duration.",
             r"Error at line 6, column 12",
         ),
         (
@@ -393,3 +395,120 @@ def test_multiple_box_statements():
     """
     result = loads(qasm_str)
     result.validate()
+
+
+def test_box_parallel_delays_on_disjoint_qubits():
+    """Delays on different qubits run in parallel, so a box only needs to
+    fit the busiest single qubit timeline — not the sum of all delays.
+
+    Regression: box duration validation summed delay durations across all
+    qubits, rejecting valid programs whose per-qubit timelines fit within
+    the box duration. See: https://github.com/qBraid/pyqasm/issues/329
+    """
+    qasm_str = """
+    OPENQASM 3.0;
+    include "stdgates.inc";
+    qubit[2] q;
+
+    box [300ns] {
+        delay[200ns] q[0];
+        delay[200ns] q[1];
+    }
+    """
+    module = loads(qasm_str)
+    module.validate()
+    module.unroll()
+
+
+def test_box_delay_broadcast_matches_per_qubit_statements():
+    """A broadcast delay and the equivalent per-qubit delay statements
+    describe the same schedule, so both must validate identically.
+
+    Regression: the cross-qubit sum counted a broadcast delay once but
+    per-qubit statements once each, so only the broadcast form validated.
+    See: https://github.com/qBraid/pyqasm/issues/329
+    """
+    broadcast = """
+    OPENQASM 3.0;
+    include "stdgates.inc";
+    qubit[2] q;
+
+    box [300ns] {
+        delay[200ns] q;
+    }
+    """
+    per_qubit = """
+    OPENQASM 3.0;
+    include "stdgates.inc";
+    qubit[2] q;
+
+    box [300ns] {
+        delay[200ns] q[0];
+        delay[200ns] q[1];
+    }
+    """
+    loads(broadcast).unroll()
+    loads(per_qubit).unroll()
+
+
+def test_box_sequential_delays_on_same_qubit_exceed_duration():
+    """Sequential delays on the same qubit accumulate on that qubit's
+    timeline and must still be rejected when they exceed the box duration."""
+    qasm_str = """
+    OPENQASM 3.0;
+    include "stdgates.inc";
+    qubit[2] q;
+
+    box [300ns] {
+        delay[200ns] q[0];
+        delay[200ns] q[1];
+        delay[150ns] q[0];
+    }
+    """
+    with pytest.raises(ValidationError) as err:
+        loads(qasm_str).unroll()
+    assert (
+        "Total delay duration value '350.0ns' on qubit 'q[0]' "
+        "should be less than 'box[300.0ns]' duration." in str(err.value)
+    )
+
+
+def test_nested_box_delays_propagate_to_outer_box():
+    """A nested box contributes its declared duration to the enclosing
+    box's per-qubit timelines, and delays after it keep accumulating.
+
+    Regression: the previous global accumulator was reset to zero when the
+    inner box closed, so the outer box lost all inner delay accounting.
+    """
+    valid = """
+    OPENQASM 3.0;
+    include "stdgates.inc";
+    qubit[2] q;
+
+    box [500ns] {
+        box [400ns] {
+            delay[100ns] q[0];
+        }
+        delay[50ns] q[0];
+    }
+    """
+    loads(valid).unroll()
+
+    invalid = """
+    OPENQASM 3.0;
+    include "stdgates.inc";
+    qubit[2] q;
+
+    box [500ns] {
+        box [400ns] {
+            delay[100ns] q[0];
+        }
+        delay[150ns] q[0];
+    }
+    """
+    with pytest.raises(ValidationError) as err:
+        loads(invalid).unroll()
+    assert (
+        "Total delay duration value '550.0ns' on qubit 'q[0]' "
+        "should be less than 'box[500.0ns]' duration." in str(err.value)
+    )