Align pattern support with node-level checks (scoped 15-pattern rules)

src/winml/modelkit/analyze/core/rules_prefilter.py

@@ -0,0 +1,94 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+# --------------------------------------------------------------------------
+"""Rules-based prefilter utilities for runtime-check skip decisions."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from .runtime_checker_query import RuntimeCheckerQuery
+
+
+if TYPE_CHECKING:
+    import onnx
+
+    from ...utils.constants import EPName
+
+
+logger = logging.getLogger(__name__)
+
+
+class RuntimeCheckerRulesPrefilter:
+    """Fixed rules-prefilter service bound to one EP+device pair."""
+
+    def __init__(self, ep_name: EPName, device_type: str) -> None:
+        self._ep_name = ep_name
+        self._device_type = device_type
+
+    def build_skip_check_result_for_rules_all_nodes_compile_run_pass(
+        self,
+        onnx_model: onnx.ModelProto,
+    ) -> dict[str, Any] | None:
+        """Build synthetic check_result when rules indicate all nodes pass."""
+        return build_skip_check_result_for_rules_all_nodes_compile_run_pass(
+            onnx_model,
+            self._ep_name,
+            self._device_type,
+        )
+
+
+def build_skip_check_result_for_rules_all_nodes_compile_run_pass(
+    onnx_model: onnx.ModelProto,
+    ep_name: EPName,
+    device_type: str,
+) -> dict[str, Any] | None:
+    """Return synthetic check_result when all nodes have rules compile/run pass.
+
+    Returns None when rules data is missing, any node is unsupported,
+    or prefilter evaluation fails. Caller should then run real compile/run.
+    """
+    try:
+        query = RuntimeCheckerQuery(
+            model_proto=onnx_model,
+            ep_name=ep_name,
+            device_type=device_type,
+        )
+        node_results = [
+            query.run_for_node(node, run_unknown_op=False) for node in query.model_proto.graph.node
+        ]
+
+        if not node_results:
+            return None
+
+        all_supported = all(
+            result.result.compile and result.result.run and not result.result.no_data
+            for result in node_results
+        )
+        if not all_supported:
+            return None
+
+        reason = f"rules_all_nodes_compile_run_pass_{len(node_results)}_nodes"
+        return {
+            "compile": {
+                "result": {"success": True, "reason": reason},
+                "stdout": "skipped_by_rules_all_nodes_compile_run_pass",
+                "stderr": "",
+            },
+            "run": {
+                "result": {"success": True, "reason": reason},
+                "stdout": "skipped_by_rules_all_nodes_compile_run_pass",
+                "stderr": "",
+            },
+        }
+    except Exception as exc:
+        logger.warning("Rules prefilter failed for one case (%s); fallback to ep_checker.", exc)
+        return None
+
+
+__all__ = [
+    "RuntimeCheckerRulesPrefilter",
+    "build_skip_check_result_for_rules_all_nodes_compile_run_pass",
+]

src/winml/modelkit/analyze/pattern/check_patterns.py

@@ -27,7 +27,6 @@
     get_pattern_input_generator,
     get_registered_pattern_input_generators,
 )
-from ...sysinfo import SysInfo
 
 
 if TYPE_CHECKING:
@@ -38,6 +37,7 @@
 
     from ...utils.constants import EPName
 from ...utils import constants
+from ..core.rules_prefilter import RuntimeCheckerRulesPrefilter
 from ..runtime_checker.ep_checker import EPChecker
 from ..utils import CheckResultWriter, load_case_indices_from_conflict_file
 
@@ -84,8 +84,8 @@ def check_patterns(
     Returns:
         Dictionary mapping pattern names to their test results:
         {
-            "Gelu": {"check_results": [...], "sys_info": {...}, "output_path": "..."},
-            "MatMulAdd": {"check_results": [...], "sys_info": {...}, "output_path": "..."},
+            "Gelu": {"check_results": [...], "output_path": "..."},
+            "MatMulAdd": {"check_results": [...], "output_path": "..."},
             ...
         }
     """
@@ -95,8 +95,6 @@ def check_patterns(
         case_index = load_case_indices_from_conflict_file(conflict_file)
         print(f"Loaded {len(case_index)} case_index values from conflict file: {conflict_file}")
 
-    sys_info = SysInfo().to_dict()
-
     # Create output directory if it doesn't exist
     output_dir = Path(output_dir)
     output_dir.mkdir(parents=True, exist_ok=True)
@@ -112,6 +110,13 @@ def check_patterns(
         ONNXDomain.from_str(domain): version for domain, version in opset_mapping.items()
     }
 
+    ep_checker.set_rules_prefilter(
+        RuntimeCheckerRulesPrefilter(
+            ep_name=ep_checker.ep_name,
+            device_type=ep_checker.device_type.name,
+        )
+    )
+
     # Test each pattern
     for pattern_name in patterns:
         print(f"\n{'=' * 80}")
@@ -151,7 +156,6 @@ def check_patterns(
         # Use writer as context manager (auto-flushes on exit)
         with CheckResultWriter(
             output_path,
-            sys_info,
             save_per_cases=None if dry_run else 20,
             rerun_failed=rerun_failed,
             delta_only=delta_only,
@@ -162,6 +166,11 @@ def check_patterns(
             print(f"Running {pattern_name} tests on {ep_checker.ep_name}...")
             if n_cases is not None:
                 print(f"Limiting to first {n_cases} test cases")
+            if not dry_run:
+                print(
+                    "Rules-first mode: check node support via parquet rules before ep_checker; "
+                    "only fallback to real compile/run when any node is unsupported/no_data."
+                )
 
             check_results_iter = gen.check_on_ep(
                 ep_checker,
@@ -170,7 +179,6 @@ def check_patterns(
                 skip_cases=0,
                 save_failed_model=save_failed_model,
                 skip_signature_fn=writer.should_skip_case,
-                yield_skipped=True,
                 dry_run=dry_run,
             )
 
@@ -209,7 +217,6 @@ def check_patterns(
         # Store results for return
         all_results[pattern_name] = {
             "check_results": check_results,
-            "sys_info": sys_info,
             "output_path": str(output_path),
         }
 
@@ -234,6 +241,19 @@ def __init__(self, device_type: ort.OrtHardwareDeviceType) -> None:
         super().__init__(ep_name="QNNExecutionProvider", device_type=device_type)
 
 
+class RTXChecker(EPChecker):
+    """NVIDIA TensorRT RTX execution provider checker wrapper for pytest compatibility."""
+
+    def __init__(self, device_type: ort.OrtHardwareDeviceType) -> None:
+        """Initialize RTX checker."""
+        if device_type != constants.DEVICE_TO_DEVICE_TYPE["GPU"]:
+            raise ValueError("NvTensorRTRTXExecutionProvider only supports GPU device type")
+        super().__init__(
+            ep_name="NvTensorRTRTXExecutionProvider",
+            device_type=constants.DEVICE_TO_DEVICE_TYPE["GPU"],
+        )
+
+
 def get_ep_checker(ep_name: EPName, device: str) -> EPChecker:
     """Get EPChecker for given execution provider name.
 
@@ -251,13 +271,15 @@ def get_ep_checker(ep_name: EPName, device: str) -> EPChecker:
     ep_name_to_checker: dict[str, Callable[..., EPChecker]] = {
         "QNNExecutionProvider": QNNNPUChecker,
         "OpenVINOExecutionProvider": OpenVINONPUChecker,
+        "NvTensorRTRTXExecutionProvider": RTXChecker,
         # Add other EPChecker subclasses here as needed
     }
     if ep_name not in ep_name_to_checker:
         raise ValueError(
             f"Unsupported execution provider: {ep_name}. "
             f"Available: QNNExecutionProvider, "
-            f"OpenVINOExecutionProvider"
+            f"OpenVINOExecutionProvider, "
+            f"NvTensorRTRTXExecutionProvider"
         )
     return ep_name_to_checker[ep_name](device_type=device_type)
 
@@ -292,10 +314,15 @@ def build_parser() -> argparse.ArgumentParser:
         "--ep",
         type=str,
         required=True,
-        choices=["QNNExecutionProvider", "OpenVINOExecutionProvider"],
+        choices=[
+            "QNNExecutionProvider",
+            "OpenVINOExecutionProvider",
+            "NvTensorRTRTXExecutionProvider",
+        ],
         help=(
             "Execution Provider names to test. "
-            "Available: QNNExecutionProvider, OpenVINOExecutionProvider"
+            "Available: QNNExecutionProvider, OpenVINOExecutionProvider, "
+            "NvTensorRTRTXExecutionProvider"
         ),
     )
     parser.add_argument(

src/winml/modelkit/analyze/runtime_checker/check_ops.py

@@ -37,7 +37,6 @@
 
     from ...utils.constants import EPName
 from ...pattern.op_input_gen.qdq_gen import QDQGenerator
-from ...sysinfo import SysInfo
 from ...utils import constants
 from ..utils import CheckResultWriter, load_case_indices_from_conflict_file
 from ..utils.model_utils import get_op_since_version
@@ -98,7 +97,6 @@ def check_ops(
         case_index = load_case_indices_from_conflict_file(conflict_file)
         print(f"Loaded {len(case_index)} case_index values from conflict file: {conflict_file}")
 
-    sys_info = SysInfo().to_dict()
     domain = ONNXDomain.from_str(opset_domain)
 
     qdq_gen = QDQGenerator(1, ONNXDomain.COM_MICROSOFT) if use_qdq else None
@@ -160,7 +158,6 @@ def check_ops(
             # Use writer as context manager (auto-flushes on exit)
             with CheckResultWriter(
                 output_path,
-                sys_info,
                 save_per_cases=None if dry_run else 100,
                 rerun_failed=rerun_failed,
                 delta_only=delta_only,
@@ -181,9 +178,6 @@ def check_ops(
                     save_model=save_model,
                     model_output_dir=model_output_dir,
                     skip_signature_fn=writer.should_skip_case,
-                    # Also yield skipped cases (with skip
-                    # marker) to maintain order
-                    yield_skipped=True,
                     dry_run=dry_run,
                 )
 
@@ -257,29 +251,29 @@ class VitisAIChecker(EPChecker):
     """VitisAI execution provider checker wrapper for pytest compatibility."""
 
     def __init__(self, device_type: ort.OrtHardwareDeviceType) -> None:
+        """Initialize VitisAI checker."""
         if device_type != ort.OrtHardwareDeviceType.NPU:
             raise ValueError("VitisAIExecutionProvider only supports NPU device type")
-        """Initialize VitisAI checker."""
         super().__init__(ep_name="VitisAIExecutionProvider", device_type=device_type)
 
 
 class MIGraphXChecker(EPChecker):
     """MIGraphX execution provider checker wrapper for pytest compatibility."""
 
     def __init__(self, device_type: ort.OrtHardwareDeviceType) -> None:
+        """Initialize MIGraphX checker."""
         if device_type != ort.OrtHardwareDeviceType.GPU:
             raise ValueError("MIGraphXExecutionProvider only supports GPU device type")
-        """Initialize MIGraphX checker."""
         super().__init__(ep_name="MIGraphXExecutionProvider", device_type=device_type)
 
 
 class RTXChecker(EPChecker):
     """NVIDIA TensorRT RTX execution provider checker wrapper for pytest compatibility."""
 
     def __init__(self, device_type: ort.OrtHardwareDeviceType) -> None:
+        """Initialize RTX checker."""
         if device_type != ort.OrtHardwareDeviceType.GPU:
             raise ValueError("NvTensorRTRTXExecutionProvider only supports GPU device type")
-        """Initialize RTX checker."""
         super().__init__(
             ep_name="NvTensorRTRTXExecutionProvider", device_type=ort.OrtHardwareDeviceType.GPU
         )

src/winml/modelkit/analyze/runtime_checker/ep_checker.py

@@ -6,7 +6,7 @@
 
 import tempfile
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar, Protocol
 
 import onnx
 import onnxruntime as ort
@@ -29,6 +29,16 @@
 # as sess.get_session_options() may return a modified version of session options
 
 
+class _RulesPrefilterProtocol(Protocol):
+    """Protocol for rules-prefilter service used by EPChecker."""
+
+    def build_skip_check_result_for_rules_all_nodes_compile_run_pass(
+        self,
+        onnx_model: onnx.ModelProto,
+    ) -> dict[str, Any] | None:
+        pass
+
+
 class EPChecker:
     """Test execution provider compilation and runtime behavior."""
 
@@ -51,10 +61,12 @@ def __init__(
         ep_name: EPName,
         device_type: ort.OrtHardwareDeviceType,
         provider_options: Sequence[dict[Any, Any]] | None = None,
+        rules_prefilter: _RulesPrefilterProtocol | None = None,
     ) -> None:
         self.device_type = device_type
-        self.ep_name = ep_name
+        self.ep_name: EPName = ep_name
         self._provider_options = provider_options
+        self._rules_prefilter = rules_prefilter
 
     def _get_sess_options(self) -> ort.SessionOptions:
         winml.register_execution_providers(ort=True)
@@ -74,6 +86,21 @@ def needs_case_isolation(self) -> bool:
             return False
         return self.device_type in required_device_types
 
+    def set_rules_prefilter(self, rules_prefilter: _RulesPrefilterProtocol | None) -> None:
+        """Set or clear rules-prefilter service for this checker."""
+        self._rules_prefilter = rules_prefilter
+
+    def build_skip_check_result_for_rules_all_nodes_compile_run_pass(
+        self,
+        onnx_model: onnx.ModelProto,
+    ) -> dict[str, Any] | None:
+        """Build synthetic check_result from fixed rules-prefilter service."""
+        if self._rules_prefilter is None:
+            return None
+        return self._rules_prefilter.build_skip_check_result_for_rules_all_nodes_compile_run_pass(
+            onnx_model
+        )
+
     def check_compile(
         self,
         path_or_bytes: str | bytes | PathLike[Any],

src/winml/modelkit/analyze/runtime_checker/result_sanitizer.py

@@ -0,0 +1,10 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+# --------------------------------------------------------------------------
+"""Backward-compatible re-export for runtime checker result sanitizers."""
+
+from ...utils.result_sanitizer import sanitize_check_result_payload, sanitize_result_text
+
+
+__all__ = ["sanitize_check_result_payload", "sanitize_result_text"]