moved collecting errors to Kirin's InterpreterABC

Author: zhenrongliew

src/bloqade/analysis/validation/nocloning/analysis.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Sequence
 
 from kirin import ir
 from kirin.analysis import Forward
@@ -57,19 +57,15 @@ class _NoCloningAnalysis(Forward[QubitValidation]):
     def __init__(self, dialects):
         super().__init__(dialects)
         self._address_frame: ForwardFrame[Address] | None = None
-        self._validation_errors: list[ValidationError] = []
 
     def method_self(self, method: ir.Method) -> QubitValidation:
         return self.lattice.bottom()
 
     def run(self, method: ir.Method, *args: QubitValidation, **kwargs: QubitValidation):
-        # Set up address frame before analysis if not already cached
         if self._address_frame is None:
             addr_analysis = AddressAnalysis(self.dialects)
             addr_analysis.initialize()
             self._address_frame, _ = addr_analysis.run(method)
-
-        # Now run the forward analysis with address frame populated
         return super().run(method, *args, **kwargs)
 
     def eval_fallback(
@@ -122,9 +118,10 @@ def eval_fallback(
             if qubit_addr in seen:
                 violation = f"Qubit[{qubit_addr}] on {gate_name} Gate"
                 must_violations.append(violation)
-                self._validation_errors.append(
-                    QubitValidationError(node, qubit_addr, gate_name)
+                self.add_validation_error(
+                    node, QubitValidationError(node, qubit_addr, gate_name)
                 )
+
             seen.add(qubit_addr)
 
         if must_violations:
@@ -136,8 +133,8 @@ def eval_fallback(
             else:
                 condition = f", with unknown argument {args_str}"
 
-            self._validation_errors.append(
-                PotentialQubitValidationError(node, gate_name, condition)
+            self.add_validation_error(
+                node, PotentialQubitValidationError(node, gate_name, condition)
             )
 
             usage = May(violations=frozenset([f"{gate_name} Gate{condition}"]))
@@ -195,14 +192,14 @@ def run(self, method: ir.Method) -> tuple[Any, list[ValidationError]]:
         if self._cached_address_frame is not None:
             self._analysis._address_frame = self._cached_address_frame
         frame, _ = self._analysis.run(method)
-
-        return frame, self._analysis._validation_errors
+        return frame, self._analysis.get_validation_errors()
 
     def print_validation_errors(self):
         """Print all collected errors with formatted snippets."""
         if self._analysis is None:
             return
-        for err in self._analysis._validation_errors:
+        validation_errors = self._analysis.get_validation_errors()
+        for err in validation_errors:
             if isinstance(err, QubitValidationError):
                 print(
                     f"\n\033[31mError\033[0m: Cloning qubit [{err.qubit_id}] at {err.gate_name} gate"

src/bloqade/analysis/validation/nocloning/impls.py

@@ -24,71 +24,68 @@ def if_else(
         except Exception:
             cond_validation = Top()
 
-        errors_before_then = len(interp_._validation_errors)
+        errors_before_then_keys = set(interp_._validation_errors.keys())
+
         with interp_.new_frame(stmt, has_parent_access=True) as then_frame:
             interp_.frame_call_region(then_frame, stmt, stmt.then_body, cond_validation)
             frame.set_values(then_frame.entries.keys(), then_frame.entries.values())
-        errors_after_then = len(interp_._validation_errors)
+        then_keys = set(interp_._validation_errors.keys()) - errors_before_then_keys
+        then_errors = interp_.get_validation_errors(keys=then_keys)
 
-        then_had_errors = errors_after_then > errors_before_then
-        then_errors = interp_._validation_errors[errors_before_then:errors_after_then]
         then_state = (
             Must(violations=frozenset(err.args[0] for err in then_errors))
-            if then_had_errors
+            if bool(then_keys)
             else Bottom()
         )
 
         if stmt.else_body:
-            errors_before_else = len(interp_._validation_errors)
+            errors_before_else_keys = set(interp_._validation_errors.keys())
+
             with interp_.new_frame(stmt, has_parent_access=True) as else_frame:
                 interp_.frame_call_region(
                     else_frame, stmt, stmt.else_body, cond_validation
                 )
                 frame.set_values(else_frame.entries.keys(), else_frame.entries.values())
-            errors_after_else = len(interp_._validation_errors)
+            else_keys = set(interp_._validation_errors.keys()) - errors_before_else_keys
+            else_errors = interp_.get_validation_errors(keys=else_keys)
 
-            else_had_errors = errors_after_else > errors_before_else
-            else_errors = interp_._validation_errors[
-                errors_before_else:errors_after_else
-            ]
             else_state = (
                 Must(violations=frozenset(err.args[0] for err in else_errors))
-                if else_had_errors
+                if bool(else_keys)
                 else Bottom()
             )
 
             merged = then_state.join(else_state)
 
             if isinstance(merged, May):
-                interp_._validation_errors = interp_._validation_errors[
-                    :errors_before_then
-                ]
+                branch_keys = then_keys | else_keys
+                for k in branch_keys:
+                    interp_._validation_errors.pop(k, None)
 
-                for err in then_errors + else_errors:
+                for err in then_errors:
                     if isinstance(err, QubitValidationError):
                         potential_err = PotentialQubitValidationError(
-                            err.node,
-                            err.gate_name,
-                            (
-                                ", when condition is true"
-                                if err in then_errors
-                                else ", when condition is false"
-                            ),
+                            err.node, err.gate_name, ", when condition is true"
                         )
-                        interp_._validation_errors.append(potential_err)
+                        interp_.add_validation_error(err.node, potential_err)
+
+                for err in else_errors:
+                    if isinstance(err, QubitValidationError):
+                        potential_err = PotentialQubitValidationError(
+                            err.node, err.gate_name, ", when condition is false"
+                        )
+                        interp_.add_validation_error(err.node, potential_err)
         else:
             merged = then_state.join(Bottom())
 
             if isinstance(merged, May):
-                interp_._validation_errors = interp_._validation_errors[
-                    :errors_before_then
-                ]
-
+                for k in then_keys:
+                    interp_._validation_errors.pop(k, None)
                 for err in then_errors:
                     if isinstance(err, QubitValidationError):
                         potential_err = PotentialQubitValidationError(
                             err.node, err.gate_name, ", when condition is true"
                         )
-                        interp_._validation_errors.append(potential_err)
+                        interp_.add_validation_error(err.node, potential_err)
 
         return (merged,)

test/analysis/validation/nocloning/test_no_cloning.py

@@ -29,8 +29,8 @@ def collect_errors_from_validation(
 
     if validation._analysis is None:
         return (must_count, may_count)
-
-    for err in validation._analysis._validation_errors:
+    print(validation._analysis.get_validation_errors())
+    for err in validation._analysis.get_validation_errors():
         if isinstance(err, QubitValidationError):
             must_count += 1
         elif isinstance(err, PotentialQubitValidationError):

test/analysis/validation/test_compose_validation.py

@@ -9,8 +9,8 @@ def test_validation_suite():
     @squin.kernel
     def bad_kernel(a: int):
         q = squin.qalloc(2)
-        squin.cx(q[0], q[0])  # cloning error
-        squin.cx(q[a], q[1])  # cloning error
+        squin.cx(q[0], q[0])  # definite cloning error
+        squin.cx(q[a], q[1])  # potential cloning error
 
     # Running no-cloning validation multiple times
     suite = ValidationSuite(
@@ -35,9 +35,8 @@ def test_validation_suite2():
     @squin.kernel
     def good_kernel():
         q = squin.qalloc(2)
-        squin.cx(q[0], q[1])  # cloning error
+        squin.cx(q[0], q[1])
 
-    # Running no-cloning validation multiple times
     suite = ValidationSuite(
         [
             NoCloningValidation,