Qualcomm: reject a delegate whose QNN graph I/O does not match its si…

backends/qualcomm/qnn_preprocess.py

@@ -8,6 +8,7 @@
 from collections import defaultdict
 from typing import Dict, final, List
 
+import executorch.backends.qualcomm.python.PyQnnManagerAdaptor as PyQnnManager
 import torch  # noqa: F401
 from executorch.backends.qualcomm._passes.qnn_pass_manager import (
     get_qnn_pass_manager_cls,
@@ -46,6 +47,49 @@
 logger.setLevel(logging.DEBUG)
 
 
+def _check_io_binding(edge_program: ExportedProgram, nodes_to_wrappers) -> None:
+    """Fail here if QNN's graph I/O will not line up with the delegate signature.
+
+    At runtime the delegate binds its arguments positionally: it walks the tensor
+    lists recovered from the context binary and consumes one argument per tensor
+    the name prefixes mark as bindable (QnnExecuTorchBackend::execute). Nothing
+    reconciles that walk with the number of arguments ExecuTorch actually passes,
+    so a graph that publishes extra I/O reads past the end of the argument list on
+    device. Catching it here costs one pass over the wrappers and reports the
+    offending tensor names instead of a register dump.
+    """
+    qnn_inputs, qnn_outputs = set(), set()
+    for wrappers in nodes_to_wrappers.values():
+        for wrapper in wrappers.values():
+            name = PyQnnManager.PyQnnTensorWrapper(wrapper).GetName()
+            # Mutable buffers are threaded through separately and never consume a
+            # delegate argument; the runtime skips them by the same marker.
+            if "mutbuf_" in name:
+                continue
+            if name.startswith("input_"):
+                qnn_inputs.add(name)
+            elif name.startswith("output_"):
+                qnn_outputs.add(name)
+
+    signature = edge_program.graph_signature
+    num_inputs = len(signature.user_inputs)
+    num_outputs = len(signature.user_outputs)
+    if len(qnn_inputs) == num_inputs and len(qnn_outputs) == num_outputs:
+        return
+
+    raise RuntimeError(
+        "QNN graph I/O does not match the delegated program signature. QNN "
+        f"declares {len(qnn_inputs)} graph inputs and {len(qnn_outputs)} graph "
+        f"outputs; the signature declares {num_inputs} user inputs and "
+        f"{num_outputs} user outputs. The runtime binds delegate arguments "
+        "positionally, so this reads past the end of the argument list on device."
+        f"\n  qnn inputs        : {sorted(qnn_inputs)}"
+        f"\n  qnn outputs       : {sorted(qnn_outputs)}"
+        f"\n  signature inputs  : {list(signature.user_inputs)}"
+        f"\n  signature outputs : {list(signature.user_outputs)}"
+    )
+
+
 @final
 class QnnBackend(BackendDetails):
     @staticmethod
@@ -111,6 +155,7 @@ def _build_op_wrappers(
             else:
                 raise RuntimeError(f"{node.op} is not supported in Qnn")
 
+        _check_io_binding(edge_program, nodes_to_wrappers)
         return py_op_wrapper_list
 
     @staticmethod

backends/qualcomm/tests/test_passes.py

@@ -848,5 +848,79 @@ def test_backend_bundle_cache_survives_an_expired_entry(self):
         self.assertGreaterEqual(reused, 1, "third manager must reuse the live bundle")
 
 
+class TestIoBindingCheck(unittest.TestCase):
+    """_check_io_binding must fire when QNN publishes I/O the signature lacks.
+
+    The runtime binds delegate arguments positionally against the tensor list in
+    the context binary, so a graph that declares more bindable tensors than the
+    program passes reads past the end of args. That has to be caught at lowering.
+    """
+
+    @staticmethod
+    def _edge_program(num_inputs, num_outputs):
+        signature = MagicMock()
+        signature.user_inputs = [f"arg_{i}" for i in range(num_inputs)]
+        signature.user_outputs = [f"out_{i}" for i in range(num_outputs)]
+        edge_program = MagicMock()
+        edge_program.graph_signature = signature
+        return edge_program
+
+    def _run(self, names, num_inputs, num_outputs):
+        from executorch.backends.qualcomm import qnn_preprocess
+
+        wrappers = {f"n{i}": {0: object()} for i in range(len(names))}
+        by_id = {
+            id(next(iter(v.values()))): name
+            for v, name in zip(wrappers.values(), names)
+        }
+
+        class _StubWrapper:
+            def __init__(self, wrapper):
+                self._name = by_id[id(wrapper)]
+
+            def GetName(self):
+                return self._name
+
+        with unittest.mock.patch.object(
+            qnn_preprocess.PyQnnManager, "PyQnnTensorWrapper", _StubWrapper
+        ):
+            qnn_preprocess._check_io_binding(
+                self._edge_program(num_inputs, num_outputs), wrappers
+            )
+
+    def test_matching_io_passes(self):
+        self._run(
+            ["input_0_x", "output_y", "conv2d_3", "_frozen_param0"],
+            num_inputs=1,
+            num_outputs=1,
+        )
+
+    def test_mutable_buffers_do_not_consume_arguments(self):
+        # A mutable buffer is threaded through separately and is not a user input.
+        self._run(
+            ["input_0_x", "input_1_mutbuf_0_cache", "output_y"],
+            num_inputs=1,
+            num_outputs=1,
+        )
+
+    def test_extra_published_inputs_are_rejected(self):
+        with self.assertRaises(RuntimeError) as ctx:
+            self._run(
+                ["input_0_x", "input_1_stray", "input_2_stray", "output_y"],
+                num_inputs=1,
+                num_outputs=1,
+            )
+        message = str(ctx.exception)
+        self.assertIn("QNN declares 3 graph inputs", message)
+        self.assertIn("input_1_stray", message)
+
+    def test_extra_published_outputs_are_rejected(self):
+        with self.assertRaises(RuntimeError) as ctx:
+            self._run(
+                ["input_0_x", "output_y", "output_stray"], num_inputs=1, num_outputs=1
+            )
+        self.assertIn("2 graph outputs", str(ctx.exception))
+
+
 if __name__ == "__main__":
     unittest.main()