feat: EP registration + monitoring — subprocess isolation, structured failures, universal op-tracing dispatch, lazy CLI startup

docs/naming-convention.md

@@ -11,7 +11,7 @@ Domain acronyms in PascalCase class names **retain their uppercase form**, excep
 | Acronym | Meaning | Class Casing | Example |
 |---------|---------|--------------|---------|
 | ONNX | Open Neural Network Exchange | `ONNX` | `ONNXStaticAnalyzer`, `ONNXLoader` |
-| EP | Execution Provider | `EP` | `EPChecker`, `EPConfig`, `EPMonitor` |
+| EP | Execution Provider | `EP` | `EPChecker`, `EPConfig`, `WinMLEPMonitor` |
 | QDQ | Quantize-Dequantize | `QDQ` | `QDQParameterConfig`, `QDQGenerator` |
 | QNN | Qualcomm Neural Network | `QNN` | `QNNMonitor` |
 | Op | Operator (2-letter prefix) | `Op` | `OpUnsupportedError` |

src/winml/modelkit/__init__.py

@@ -43,7 +43,6 @@
 
 logging.getLogger(__name__).addHandler(logging.NullHandler())
 
-
 def _preload_bundled_onnxruntime_dll() -> None:
     # Windows ships C:\Windows\System32\onnxruntime.dll (older API version)
     # as part of the system WindowsML component. When WinML EP plugin DLLs
@@ -74,7 +73,14 @@ def _preload_bundled_onnxruntime_dll() -> None:
 
 _preload_bundled_onnxruntime_dll()
 
-from . import _warnings  # Configure warning filters before importing subpackages
+# _warnings configures filters before any subpackage imports.
+# transformers_compat arms a sys.meta_path hook — the shim fires lazily
+# the first time anything imports optimum.*; lightweight commands
+# (``winml sys``, ``winml --help``) never pay the transformers cost.
+from . import _warnings  # noqa: I001
+from . import transformers_compat
+
+transformers_compat.arm()
 
 
 try:

src/winml/modelkit/_warnings.py

@@ -20,6 +20,7 @@
 
 import logging
 import os
+import sys
 import warnings
 
 from ._env import env_flag_enabled
@@ -91,12 +92,15 @@ def filter(self, record: logging.LogRecord) -> bool:
     for _cat in (FutureWarning, DeprecationWarning, UserWarning):
         warnings.filterwarnings("ignore", category=_cat, module=r"torch\..*")
 
-    # NOTE: TracerWarning (from torch.jit) is intentionally NOT filtered here.
-    # Importing torch.jit at startup would pull all of torch (~1.7s) into
-    # `winml --help` and violate the CLI import budget (tests/cli/test_import_time.py).
-    # During ONNX export, export_pytorch() wraps torch.onnx.export in
-    # `warnings.catch_warnings()` + `filterwarnings("ignore")`, which is strictly
-    # broader than a TracerWarning-only filter.
+    # TracerWarning (from torch.jit, inherits Warning not UserWarning)
+    # fires during ONNX export tracing — safe to suppress in both torch and
+    # transformers. Only register the filter if torch has ALREADY been
+    # imported; otherwise loading torch here would add ~2s to every
+    # lightweight command (``winml sys`` etc.) that never touches ONNX
+    # export. The export path re-triggers this by calling
+    # :func:`install_torch_tracer_filter` after loading torch.
+    if "torch" in sys.modules:
+        install_torch_tracer_filter()
 
     # Diffusers
     warnings.filterwarnings(
@@ -140,5 +144,17 @@ def filter(self, record: logging.LogRecord) -> bool:
     logging.getLogger("transformers.modeling_utils").addFilter(_TransformersWeightsFilter())
 
 
+def install_torch_tracer_filter() -> None:
+    """Register the ``TracerWarning`` filter — call after ``import torch``.
+
+    Idempotent — ``warnings.filterwarnings`` de-duplicates identical entries.
+    """
+    try:
+        from torch.jit import TracerWarning
+    except ImportError:
+        return  # torch not installed
+    warnings.filterwarnings("ignore", category=TracerWarning)
+
+
 # Auto-configure on import
 _configure()

src/winml/modelkit/analyze/analyzer.py

@@ -18,7 +18,7 @@
 from typing import TYPE_CHECKING, Any
 
 from ..optim.config import WinMLOptimizationConfig
-from ..utils.constants import EP_SUPPORTED_DEVICES, EPName, EPNameOrAlias, normalize_ep_name
+from ..utils.constants import normalize_ep_name
 from .models.information import Information
 from .models.output import RuntimeDebugSummaryEntry
 from .models.support_level import SupportLevel
@@ -186,7 +186,7 @@ def __repr__(self) -> str:
         pattern_count = sum(self.output.metadata.detected_pattern_count.values())
         return f"AnalysisResult(patterns={pattern_count})"
 
-    def is_fully_supported(self, ep: EPNameOrAlias | None = None) -> bool:
+    def is_fully_supported(self, ep: str | None = None) -> bool:
         """Check if model is fully supported on the target EP and device.
 
         Args:
@@ -227,7 +227,7 @@ def is_fully_supported(self, ep: EPNameOrAlias | None = None) -> bool:
                 return False
         return found_target
 
-    def has_errors(self, ep: EPNameOrAlias | None = None) -> bool:
+    def has_errors(self, ep: str | None = None) -> bool:
         """Check if there are any unsupported patterns (blocking errors).
 
         Args:
@@ -260,7 +260,7 @@ def has_errors(self, ep: EPNameOrAlias | None = None) -> bool:
                 return True
         return False
 
-    def has_warnings(self, ep: EPNameOrAlias | None = None) -> bool:
+    def has_warnings(self, ep: str | None = None) -> bool:
         """Check if there are any partial patterns (warnings/optimization opportunities).
 
         Args:
@@ -293,7 +293,7 @@ def has_warnings(self, ep: EPNameOrAlias | None = None) -> bool:
                 return True
         return False
 
-    def get_lint_result(self, ep: EPNameOrAlias | None = None) -> LintResult:
+    def get_lint_result(self, ep: str | None = None) -> LintResult:
         """Get lint-style result with error/warning/info counts.
 
         Args:
@@ -374,7 +374,7 @@ def get_lint_result(self, ep: EPNameOrAlias | None = None) -> LintResult:
             optimization_config=optimization_config,
         )
 
-    def get_unsupported_operators(self, ep: EPNameOrAlias | None = None) -> list[str]:
+    def get_unsupported_operators(self, ep: str | None = None) -> list[str]:
         """Get list of unsupported operators for the target EP and device.
 
         Args:
@@ -409,7 +409,7 @@ def get_unsupported_operators(self, ep: EPNameOrAlias | None = None) -> list[str
 
         return unsupported
 
-    def get_optimization_opportunities(self, ep: EPNameOrAlias | None = None) -> list[Action]:
+    def get_optimization_opportunities(self, ep: str | None = None) -> list[Action]:
         """Get actions for patterns that could be optimized (UNSUPPORTED or PARTIAL status).
 
         Args:
@@ -452,7 +452,7 @@ def get_optimization_opportunities(self, ep: EPNameOrAlias | None = None) -> lis
                             seen_patterns.add(pattern_key)
         return actions
 
-    def get_optimization_config(self, ep: EPNameOrAlias | None = None) -> WinMLOptimizationConfig:
+    def get_optimization_config(self, ep: str | None = None) -> WinMLOptimizationConfig:
         """Generate WinML optimization configuration based on action items.
 
         This method extracts optimization settings from action_items in Actions,
@@ -594,7 +594,7 @@ def __init__(self, config: AnalyzerConfig | None = None) -> None:
     def analyze(
         self,
         model_path: str,
-        ep: EPNameOrAlias | None = None,
+        ep: str | None = None,
         device: str | None = None,
         enable_information: bool = True,
         htp_metadata_path: str | None = None,
@@ -728,7 +728,7 @@ def analyze(
     def analyze_from_proto(
         self,
         model_proto: onnx.ModelProto,
-        ep: EPNameOrAlias | None = None,
+        ep: str | None = None,
         device: str | None = None,
         enable_information: bool = True,
         model_path: str | None = None,
@@ -793,25 +793,27 @@ def analyze_from_proto(
 
         # Resolve device — rule files are device-specific (CPU/GPU/NPU).
         if device is not None and device.lower() == "auto":
-            from ..sysinfo import resolve_device
+            from ..session import auto_detect_device
 
-            resolved, _ = resolve_device("auto", ep=ep_normalized)
-            device_to_use = resolved.upper()
+            device_to_use = auto_detect_device().upper()
             logger.info("Device 'auto' resolved to: %s", device_to_use)
         else:
             device_to_use = device if device is not None else "NPU"
             logger.info("Using device: %s", device_to_use)
 
         # Determine which EPs to analyze
-        eps_to_analyze: list[EPName] = []
+        eps_to_analyze: list[str] = []
         if ep_normalized is None:
-            # Analyze all EPs that support the target device
-            eps_to_analyze = [
-                ep_name
-                for ep_name, supported_devices in EP_SUPPORTED_DEVICES.items()
-                if device_to_use.lower() in supported_devices
-            ]
-            logger.info("No EP specified, analyzing all supported EPs: %s", eps_to_analyze)
+            # Derive the EP list from the catalog so future EP additions
+            # are automatically included. sorted() gives deterministic order.
+            from ..session import eps_for_device
+
+            eps_to_analyze = sorted(eps_for_device(device_to_use.lower()))
+            logger.info(
+                "No EP specified, analyzing all %s-capable EPs: %s",
+                device_to_use,
+                eps_to_analyze,
+            )
         else:
             eps_to_analyze = [ep_normalized]
 
@@ -834,8 +836,8 @@ def analyze_from_proto(
         extraction_ms = int((time.perf_counter() - extraction_start) * 1000)
 
         # Step 2: Check runtime support for each EP
-        check_op_results: dict[EPName, list[PatternRuntime]] = {}
-        information_list: dict[EPName, list[Information]] = {}
+        check_op_results: dict[str, list[PatternRuntime]] = {}
+        information_list: dict[str, list[Information]] = {}
         runtime_debug_details_summary: dict[
             str, dict[str, list[str] | dict[str, RuntimeDebugSummaryEntry]]
         ] = {}
@@ -964,7 +966,7 @@ def has_errors(self) -> bool:
 def analyze_onnx(
     model: str | Path,
     *,
-    ep: EPNameOrAlias | None = None,
+    ep: str | None = None,
     device: str | None = None,
     autoconf: bool = True,
     run_unknown_op: bool = False,

src/winml/modelkit/analyze/core/model_validators/model_validator_manager.py

@@ -24,7 +24,6 @@
 
 
 if TYPE_CHECKING:
-    from ....utils.constants import EPName
     from ...models.information import Information
     from ...models.onnx_model import ONNXModel
     from ...models.runtime_checks import PatternRuntime
@@ -78,8 +77,8 @@ def __init__(
         enabled_validators: list[str] | None = None,
         op_runtime_results: list[PatternRuntime] | None = None,
         *,
-        device: str,
-        ep: EPName,
+        device: str | None = None,
+        ep: str | None = None,
     ) -> None:
         """Initialize validator manager.
 
@@ -100,7 +99,7 @@ def __init__(
         self.model = model
         self.model_proto = model.get_model()
         self.op_runtime_results = op_runtime_results or []
-        self.device = device
+        self.device = device or "NPU"
         self.ep = ep
         self.enabled_validators = enabled_validators or list(self.VALIDATORS.keys())
 

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

@@ -81,7 +81,6 @@
 if TYPE_CHECKING:
     from winml.modelkit.pattern.match import PatternMatchResult
 
-    from ...utils.constants import EPName
     from .node_checkers.base import NodeChecker
 
 
@@ -983,7 +982,7 @@ class RuntimeCheckerQuery:
     def __init__(
         self,
         model_proto: onnx.ModelProto,
-        ep_name: EPName,
+        ep_name: str,
         device_type: str,
         model_path: str | Path | None = None,
         dynamic_axis_strict_mode: bool = False,
@@ -1146,28 +1145,44 @@ def _collect_qdq_types(self) -> None:
     def _is_ep_available_locally(self) -> bool:
         """Check if the target EP is available on the local machine.
 
+        Targeted probe through :meth:`WinMLEPRegistry.auto_device`: the
+        registry handles plugin discovery + DLL load lazily, so we no
+        longer need a separate module-level pre-register pass. Negative
+        outcomes (EP not discovered, registration failed, no matching
+        device) are caught and reported as "not available locally"
+        rather than propagated.
+
         Returns:
             True if the EP+device combination is available locally.
         """
         if self._ep_available_locally is not None:
             return self._ep_available_locally
 
-        from ... import winml
-        from ...utils.constants import DEVICE_TO_DEVICE_TYPE
-
-        device_type_enum = DEVICE_TO_DEVICE_TYPE.get(self.device_type)
-        if device_type_enum is None:
-            self._ep_available_locally = False
-            return False
+        from ...session import (
+            DeviceNotFound,
+            EPDeviceTarget,
+            WinMLEPNotDiscovered,
+            WinMLEPRegistrationFailed,
+            WinMLEPRegistry,
+            resolve_device,
+            short_ep_name,
+        )
 
         try:
-            ep_devices = winml.get_registered_ep_devices()
-            self._ep_available_locally = any(
-                ep_dev.ep_name == self.ep_name and ep_dev.device.type == device_type_enum
-                for ep_dev in ep_devices
+            target = EPDeviceTarget(
+                ep=short_ep_name(self.ep_name),
+                device=self.device_type.lower(),
             )
-        except Exception as e:
-            logger.debug("Failed to query EP devices: %s", e)
+            resolved = resolve_device(target)
+            WinMLEPRegistry.instance().auto_device(resolved)
+            self._ep_available_locally = True
+        except (
+            WinMLEPNotDiscovered,
+            WinMLEPRegistrationFailed,
+            DeviceNotFound,
+            ValueError,
+        ) as e:
+            logger.debug("EP %s on %s not available locally: %s", self.ep_name, self.device_type, e)
             self._ep_available_locally = False
 
         return self._ep_available_locally
@@ -1179,11 +1194,11 @@ def _get_ep_checker(self) -> EPChecker:
             EPChecker instance configured for the target EP+device.
         """
         if self._ep_checker is None:
-            from ...utils.constants import DEVICE_TO_DEVICE_TYPE
+            from ...session import DEVICE_TO_DEVICE_TYPE
 
             self._ep_checker = EPChecker(
                 ep_name=self.ep_name,
-                device_type=DEVICE_TO_DEVICE_TYPE[self.device_type],
+                device_type=DEVICE_TO_DEVICE_TYPE[self.device_type.lower()],
             )
         return self._ep_checker
 

src/winml/modelkit/analyze/models/ihv_type.py

@@ -4,10 +4,10 @@
 # --------------------------------------------------------------------------
 """IHV type enum."""
 
-from enum import Enum
+from enum import StrEnum
 
 
-class IHVType(str, Enum):
+class IHVType(StrEnum):
     """IHV (Independent Hardware Vendor) type."""
 
     QC = "QC"

src/winml/modelkit/analyze/models/information.py

@@ -7,7 +7,7 @@
 Represents actionable information for fixing unsupported patterns.
 """
 
-from enum import Enum
+from enum import StrEnum
 from uuid import uuid4
 
 from pydantic import BaseModel, Field
@@ -16,7 +16,7 @@
 from .support_level import SupportLevel
 
 
-class ActionLevel(str, Enum):
+class ActionLevel(StrEnum):
     """Action priority level."""
 
     REQUIRED = "required"

src/winml/modelkit/analyze/models/onnx_model.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import logging
-from enum import Enum
+from enum import StrEnum
 
 import onnx
 from pydantic import BaseModel, Field, field_validator
@@ -23,7 +23,7 @@
 logger = logging.getLogger(__name__)
 
 
-class ModelTag(str, Enum):
+class ModelTag(StrEnum):
     """Tags for marking model-level issues and validation states.
 
     These tags are stored in ONNXModel.model_tags to record various