diff --git a/CLAUDE.md b/CLAUDE.md index 96de68a002..d714a1f9ec 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -743,6 +743,67 @@ if is_open: - Never suppress InfraUnavailableError from circuit breaker - Never use circuit breaker for non-transient errors +### Node Introspection Security Considerations + +The `MixinNodeIntrospection` mixin uses Python reflection (via the `inspect` module) to automatically discover node capabilities. This provides powerful service discovery but has security implications that developers should understand. + +**What Gets Exposed via Introspection**: +- Public method names (potential operations a node can perform) +- Method signatures (parameter names and type annotations) +- Protocol and mixin implementations (discovered capabilities) +- FSM state information (if `MixinFSM` is present) +- Endpoint URLs (health, API, metrics paths) +- Node metadata (name, version, type from contract) + +**Built-in Protections**: +The mixin includes several filtering mechanisms to limit exposure: +- **Private method exclusion**: Methods prefixed with `_` are excluded from capability discovery +- **Utility method filtering**: Common utility prefixes (`get_*`, `set_*`, `initialize*`, `start_*`, `stop_*`) are filtered out +- **Operation keyword matching**: Only methods matching operation keywords (`execute`, `process`, `handle`, `run`, `perform`, `compute`, `validate`, `transform`, `aggregate`, `orchestrate`) are reported as capabilities +- **Configurable exclusions**: The `exclude_prefixes` parameter allows additional filtering + +**Best Practices for Node Developers**: +- Prefix internal/sensitive methods with `_` to exclude them from introspection +- Avoid exposing sensitive business logic in public method names +- Use generic operation names that don't reveal implementation details (e.g., `process_request` instead of `decrypt_and_forward_to_payment_gateway`) +- Review exposed capabilities before deploying to production environments +- Consider network segmentation for introspection event topics in multi-tenant environments +- Use the `exclude_prefixes` parameter to filter additional method patterns if needed + +**Example - Reviewing Exposed Capabilities**: +```python +from omnibase_infra.mixins import MixinNodeIntrospection + +class MyNode(MixinNodeIntrospection): + def execute_operation(self, data: dict) -> dict: + """Public operation - WILL be exposed.""" + return self._internal_process(data) + + def _internal_process(self, data: dict) -> dict: + """Private method - will NOT be exposed.""" + return {"processed": True} + + def get_status(self) -> str: + """Utility method - will NOT be exposed (get_* prefix filtered).""" + return "healthy" + +# Review what gets exposed +node = MyNode() +capabilities = node.get_capabilities() +# capabilities will only include "execute_operation" +``` + +**Network Security Considerations**: +- Introspection data is published to Kafka topics (`*.introspection.*`) +- In multi-tenant environments, ensure proper topic ACLs +- Consider whether introspection topics should be accessible outside the cluster +- Monitor introspection topic consumers for unauthorized access + +**Related**: +- Implementation: `src/omnibase_infra/mixins/mixin_node_introspection.py` +- Ticket: OMN-893 +- See `MixinNodeIntrospection.get_capabilities()` for filtering logic details + ### Service Integration Architecture - **Adapter Pattern** - External services wrapped in ONEX adapters (Consul, Kafka, Vault) - **Connection Pooling** - Database connections managed through dedicated pool managers diff --git a/pyproject.toml b/pyproject.toml index 8dddfbb582..3ed4b63140 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -114,11 +114,10 @@ line-ending = "auto" # Matches omnibase_core settings for consistency across ONEX ecosystem select = ["E", "F", "W", "C90", "I", "N", "UP", "YTT", "S", "BLE", "FBT", "B", "A", "COM", "C4", "DTZ", "T10", "EM", "EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PT", "Q", "RSE", "RET", "SLF", "SIM", "TID", "TCH", "ARG", "PTH", "ERA", "PD", "PGH", "PL", "TRY", "NPY", "RUF"] ignore = [ - # Type annotation style rules (both needed - UP007 handles Union, UP045 handles Optional) - # Note: ONEX prefers X | None syntax (PEP 604). These ignores allow legacy Optional[X]/Union[X,Y] + # Type annotation style rules + # Note: ONEX prefers X | None syntax (PEP 604). This ignore allows legacy Union[X,Y] # if introduced, but codebase is already compliant. Can be removed to enforce strict PEP 604. "UP007", # Union[X, Y] -> X | Y (allow legacy Union syntax if present) - "UP045", # Optional[X] -> X | None (allow legacy Optional syntax if present) "S101", # Allow assert statements in tests "E501", # Line too long (technical debt - will fix incrementally) "BLE001", # Blind exception catching (technical debt) diff --git a/src/omnibase_infra/handlers/handler_db.py b/src/omnibase_infra/handlers/handler_db.py index 60b11fb914..dcad5669a0 100644 --- a/src/omnibase_infra/handlers/handler_db.py +++ b/src/omnibase_infra/handlers/handler_db.py @@ -105,7 +105,7 @@ async def initialize(self, config: dict[str, object]) -> None: ) timeout_raw = config.get("timeout", _DEFAULT_TIMEOUT_SECONDS) - if isinstance(timeout_raw, (int, float)): + if isinstance(timeout_raw, int | float): self._timeout = float(timeout_raw) try: diff --git a/src/omnibase_infra/mixins/__init__.py b/src/omnibase_infra/mixins/__init__.py index 0c2295e080..3cd6d3c5c5 100644 --- a/src/omnibase_infra/mixins/__init__.py +++ b/src/omnibase_infra/mixins/__init__.py @@ -1,11 +1,8 @@ # SPDX-License-Identifier: MIT # Copyright (c) 2025 OmniNode Team -"""Infrastructure mixins for common cross-cutting concerns. +"""ONEX Infrastructure Mixins. -This module provides reusable mixins for infrastructure components to implement -common patterns such as circuit breakers, retry logic, health checks, and more. - -Mixins follow ONEX patterns: +Reusable mixin classes providing: - Thread-safe async operations - Infrastructure error integration - Correlation ID propagation @@ -16,5 +13,16 @@ CircuitState, MixinAsyncCircuitBreaker, ) +from omnibase_infra.mixins.mixin_node_introspection import ( + CapabilitiesDict, + IntrospectionCacheDict, + MixinNodeIntrospection, +) -__all__ = ["MixinAsyncCircuitBreaker", "CircuitState"] +__all__ = [ + "CapabilitiesDict", + "CircuitState", + "IntrospectionCacheDict", + "MixinAsyncCircuitBreaker", + "MixinNodeIntrospection", +] diff --git a/src/omnibase_infra/mixins/mixin_node_introspection.py b/src/omnibase_infra/mixins/mixin_node_introspection.py new file mode 100644 index 0000000000..53df711a76 --- /dev/null +++ b/src/omnibase_infra/mixins/mixin_node_introspection.py @@ -0,0 +1,1883 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2025 OmniNode Team +"""Node introspection mixin providing automatic capability discovery. + +This module provides a reusable mixin for ONEX nodes to implement automatic +capability discovery, endpoint reporting, and periodic heartbeat broadcasting. +It uses reflection to discover node capabilities and integrates with the event +bus for distributed service discovery. + +Features: + - Automatic capability discovery via reflection + - Endpoint URL discovery (health, api, metrics) + - FSM state reporting if applicable + - Cached introspection data with configurable TTL + - Background heartbeat task for periodic health broadcasts + - Registry listener for REQUEST_INTROSPECTION events + - Graceful degradation when event bus is unavailable + +Note: + - active_operations_count in heartbeats is currently hardcoded to 0. + Full implementation deferred - see TODO in _publish_heartbeat(). + +Security Considerations: + This mixin uses Python reflection (via the ``inspect`` module) to automatically + discover node capabilities. While this enables powerful service discovery, it + has security implications that developers must understand. + + **What Gets Exposed via Introspection**: + + - Public method names (potential operations a node can perform) + - Method signatures (parameter names and type annotations) + - Protocol and mixin implementations (discovered capabilities) + - FSM state information (if ``MixinFSM`` or similar state attributes are present) + - Endpoint URLs (health, API, metrics paths) + - Node metadata (name, version, type) + + **Built-in Protections**: + + The mixin includes filtering mechanisms to limit exposure: + + - **Private method exclusion**: Methods prefixed with ``_`` are excluded from + capability discovery. + - **Utility method filtering**: Common utility prefixes (``get_*``, ``set_*``, + ``initialize*``, ``start_*``, ``stop_*``) are filtered out by default. + - **Operation keyword matching**: Only methods containing operation keywords + (``execute``, ``handle``, ``process``, ``run``, ``invoke``, ``call``) are + reported as capabilities in the operations list. + - **Configurable exclusions**: The ``exclude_prefixes`` parameter in + ``initialize_introspection()`` allows additional filtering. + + **Best Practices for Node Developers**: + + - Prefix internal/sensitive methods with ``_`` to exclude them from introspection. + - Avoid exposing sensitive business logic in public method names (e.g., use + ``process_request`` instead of ``decrypt_and_forward_to_payment_gateway``). + - Review exposed capabilities before deploying to production environments. + - Consider network segmentation for introspection event topics in multi-tenant + environments. + - Use the ``exclude_prefixes`` parameter to filter additional method patterns + if needed. + + **Network Security Considerations**: + + - Introspection data is published to Kafka topics (``node.introspection``). + - In multi-tenant environments, ensure proper topic ACLs are configured. + - Consider whether introspection topics should be accessible outside the cluster. + - Monitor introspection topic consumers for unauthorized access. + + For more details, see the "Node Introspection Security Considerations" section + in ``CLAUDE.md``. + +Usage: + ```python + from omnibase_infra.mixins import MixinNodeIntrospection + + class MyNode(MixinNodeIntrospection): + def __init__(self, config, event_bus=None): + self.initialize_introspection( + node_id=config.node_id, + node_type="EFFECT", + event_bus=event_bus, + ) + + async def startup(self): + # Publish initial introspection on startup + await self.publish_introspection(reason="startup") + + # Start background tasks + await self.start_introspection_tasks( + enable_heartbeat=True, + heartbeat_interval_seconds=30.0, + enable_registry_listener=True, + ) + + async def shutdown(self): + # Publish shutdown introspection + await self.publish_introspection(reason="shutdown") + + # Stop background tasks + await self.stop_introspection_tasks() + ``` + +Integration Requirements: + Classes using this mixin must: + 1. Call `initialize_introspection()` during initialization + 2. Optionally call `start_introspection_tasks()` for background operations + 3. Call `stop_introspection_tasks()` during shutdown + 4. Ensure event_bus has `publish_envelope()` method if provided + +See Also: + - MixinAsyncCircuitBreaker for circuit breaker pattern + - ModelNodeIntrospectionEvent for event model + - ModelNodeHeartbeatEvent for heartbeat model + - CLAUDE.md "Node Introspection Security Considerations" section +""" + +from __future__ import annotations + +import asyncio +import inspect +import json +import logging +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, ClassVar, TypedDict, cast +from uuid import NAMESPACE_DNS, UUID, uuid4, uuid5 + +from omnibase_infra.models.discovery import ModelNodeIntrospectionEvent +from omnibase_infra.models.registration import ModelNodeHeartbeatEvent + +if TYPE_CHECKING: + from omnibase_core.protocols.event_bus import ProtocolEventBus + + from omnibase_infra.event_bus.models import ModelEventMessage + +logger = logging.getLogger(__name__) + +# Event topic constants +INTROSPECTION_TOPIC = "node.introspection" +HEARTBEAT_TOPIC = "node.heartbeat" +REQUEST_INTROSPECTION_TOPIC = "node.request_introspection" + +# Type alias for capabilities dictionary structure +# operations: list of method names, protocols: list of protocol names +# has_fsm: boolean, method_signatures: dict of method name to signature string +CapabilitiesDict = dict[str, list[str] | bool | dict[str, str]] + +# Performance threshold constants (in milliseconds) +PERF_THRESHOLD_GET_CAPABILITIES_MS = 50.0 +PERF_THRESHOLD_DISCOVER_CAPABILITIES_MS = 30.0 +PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS = 50.0 +PERF_THRESHOLD_CACHE_HIT_MS = 1.0 + + +@dataclass +class IntrospectionPerformanceMetrics: + """Performance metrics for introspection operations. + + This dataclass captures timing information for introspection operations, + enabling performance monitoring and alerting when operations exceed + the <50ms target threshold. + + Attributes: + get_capabilities_ms: Time taken by get_capabilities() in milliseconds. + discover_capabilities_ms: Time taken by _discover_capabilities() in ms. + get_endpoints_ms: Time taken by get_endpoints() in milliseconds. + get_current_state_ms: Time taken by get_current_state() in milliseconds. + total_introspection_ms: Total time for get_introspection_data() in ms. + cache_hit: Whether the result was served from cache. + method_count: Number of methods discovered during reflection. + threshold_exceeded: Whether any operation exceeded performance thresholds. + slow_operations: List of operation names that exceeded their thresholds. + + Example: + ```python + metrics = node.get_performance_metrics() + if metrics.threshold_exceeded: + logger.warning( + "Introspection performance degraded", + extra={ + "slow_operations": metrics.slow_operations, + "total_ms": metrics.total_introspection_ms, + } + ) + ``` + """ + + get_capabilities_ms: float = 0.0 + discover_capabilities_ms: float = 0.0 + get_endpoints_ms: float = 0.0 + get_current_state_ms: float = 0.0 + total_introspection_ms: float = 0.0 + cache_hit: bool = False + method_count: int = 0 + threshold_exceeded: bool = False + slow_operations: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, float | bool | int | list[str]]: + """Convert metrics to dictionary for logging/serialization. + + Returns: + Dictionary with all metric fields. + """ + return { + "get_capabilities_ms": self.get_capabilities_ms, + "discover_capabilities_ms": self.discover_capabilities_ms, + "get_endpoints_ms": self.get_endpoints_ms, + "get_current_state_ms": self.get_current_state_ms, + "total_introspection_ms": self.total_introspection_ms, + "cache_hit": self.cache_hit, + "method_count": self.method_count, + "threshold_exceeded": self.threshold_exceeded, + "slow_operations": self.slow_operations, + } + + +class IntrospectionCacheDict(TypedDict): + """TypedDict representing the JSON-serialized ModelNodeIntrospectionEvent. + + This type matches the output of ModelNodeIntrospectionEvent.model_dump(mode="json"), + enabling proper type checking for cache operations without requiring type: ignore comments. + + Note: + The capabilities field uses a permissive type to accommodate all possible + values from model_dump(), including nested structures and dynamic values. + The actual structure is: operations (list[str]), protocols (list[str]), + has_fsm (bool), method_signatures (dict[str, str]). + """ + + node_id: str + node_type: str + # Permissive type for capabilities to handle all model_dump() output variations + # Actual structure: {"operations": list, "protocols": list, "has_fsm": bool, "method_signatures": dict} + capabilities: dict[str, list[str] | bool | dict[str, str] | int | float | None] + endpoints: dict[str, str] + current_state: str | None + version: str + reason: str + correlation_id: str | None # UUID serializes to string in JSON mode + timestamp: str # datetime serializes to ISO string in JSON mode + + +class MixinNodeIntrospection: + """Mixin providing node introspection capabilities. + + Provides automatic capability discovery using reflection, endpoint + reporting, and periodic heartbeat broadcasting for ONEX nodes. + + State Variables: + _introspection_cache: Cached introspection data + _introspection_cache_ttl: Cache time-to-live in seconds + _introspection_cached_at: Timestamp when cache was populated + + Background Task Variables: + _heartbeat_task: Background heartbeat task + _registry_listener_task: Background registry listener task + _introspection_stop_event: Event to signal task shutdown + + Configuration Variables: + _introspection_node_id: Node identifier + _introspection_node_type: Node type classification + _introspection_event_bus: Optional event bus for publishing + _introspection_version: Node version string + _introspection_start_time: Node startup timestamp + + Security Considerations: + This mixin uses Python reflection (via the ``inspect`` module) to + automatically discover node capabilities. While this enables powerful + service discovery, it has security implications: + + **Exposed Information**: + + - Public method names (potential operations a node can perform) + - Method signatures (parameter names and type annotations) + - Protocol and mixin implementations (discovered capabilities) + - FSM state information (if state attributes are present) + - Endpoint URLs (health, API, metrics paths) + - Node metadata (name, version, type) + + **Built-in Protections**: + + - Private methods (prefixed with ``_``) are excluded by default + - Utility method prefixes (``get_*``, ``set_*``, etc.) are filtered + - Only methods containing operation keywords are reported as operations + - Configure ``exclude_prefixes`` in ``initialize_introspection()`` for + additional filtering + + **Recommendations for Production**: + + - Prefix internal/sensitive methods with ``_`` to exclude them + - Use generic operation names that don't reveal implementation details + - Review ``get_capabilities()`` output before production deployment + - In multi-tenant environments, configure Kafka topic ACLs for + introspection events (``node.introspection``, ``node.heartbeat``, + ``node.request_introspection``) + - Monitor introspection topic consumers for unauthorized access + - Consider network segmentation for introspection event topics + + See Also: + - Module docstring for detailed security documentation + - CLAUDE.md "Node Introspection Security Considerations" section + - ``get_capabilities()`` for filtering logic details + + Example: + ```python + class PostgresAdapter(MixinNodeIntrospection): + def __init__(self, config): + self.initialize_introspection( + node_id="postgres-adapter-001", + node_type="EFFECT", + event_bus=config.event_bus, + ) + + async def execute(self, query: str) -> list[dict]: + # Node operation - WILL be exposed via introspection + ... + + def _internal_helper(self, data: dict) -> dict: + # Private method - will NOT be exposed + ... + ``` + """ + + # Class-level cache for method signatures (populated once per class) + # Maps class -> {method_name: signature_string} + # This avoids expensive reflection on each introspection call since + # method signatures don't change after class definition. + # NOTE: ClassVar is intentionally shared across all instances - this is correct + # behavior for a per-class cache of immutable method signatures. + _class_method_cache: ClassVar[dict[type, dict[str, str]]] = {} + + # Type annotations for instance attributes (no default values to avoid shared state) + # All of these are initialized in initialize_introspection() + # + # Caching attributes + _introspection_cache: IntrospectionCacheDict | None + _introspection_cache_ttl: float + _introspection_cached_at: float | None + + # Background task attributes + _heartbeat_task: asyncio.Task[None] | None + _registry_listener_task: asyncio.Task[None] | None + _introspection_stop_event: asyncio.Event | None + _registry_unsubscribe: Callable[[], None] | Callable[[], Awaitable[None]] | None + + # Configuration attributes + _introspection_node_id: str | None + _introspection_node_type: str | None + _introspection_event_bus: ProtocolEventBus | None + _introspection_version: str + _introspection_start_time: float | None + + # Capability discovery configuration + _introspection_operation_keywords: set[str] + _introspection_exclude_prefixes: set[str] + + # Registry listener callback error tracking (instance-level) + # Used for rate-limiting error logging to prevent log spam during + # sustained failures. These are initialized in initialize_introspection(). + _registry_callback_consecutive_failures: int + _registry_callback_last_failure_time: float + _registry_callback_failure_log_threshold: int + + # Performance metrics tracking (instance-level) + # Stores the most recent performance metrics from introspection operations + _introspection_last_metrics: IntrospectionPerformanceMetrics | None + + # Default operation keywords for capability discovery + DEFAULT_OPERATION_KEYWORDS: ClassVar[set[str]] = { + "execute", + "handle", + "process", + "run", + "invoke", + "call", + } + + # Default prefixes to exclude from capability discovery + DEFAULT_EXCLUDE_PREFIXES: ClassVar[set[str]] = { + "_", + "get_", + "set_", + "initialize", + "start_", + "stop_", + } + + # Node-type-specific operation keyword suggestions + NODE_TYPE_OPERATION_KEYWORDS: ClassVar[dict[str, set[str]]] = { + "EFFECT": { + "execute", + "handle", + "process", + "run", + "invoke", + "call", + "fetch", + "send", + "query", + "connect", + }, + "COMPUTE": { + "execute", + "handle", + "process", + "run", + "compute", + "transform", + "calculate", + "convert", + "parse", + }, + "REDUCER": { + "execute", + "handle", + "process", + "run", + "aggregate", + "reduce", + "merge", + "combine", + "accumulate", + }, + "ORCHESTRATOR": { + "execute", + "handle", + "process", + "run", + "orchestrate", + "coordinate", + "schedule", + "dispatch", + }, + } + + def initialize_introspection( + self, + node_id: str, + node_type: str, + event_bus: ProtocolEventBus | None = None, + version: str = "1.0.0", + cache_ttl: float = 300.0, + operation_keywords: set[str] | None = None, + exclude_prefixes: set[str] | None = None, + ) -> None: + """Initialize introspection configuration. + + Must be called during class initialization before any introspection + operations are performed. + + Args: + node_id: Unique identifier for this node instance + node_type: Node type classification (EFFECT, COMPUTE, REDUCER, ORCHESTRATOR) + event_bus: Optional event bus for publishing introspection events. + Must have `publish_envelope()` method if provided. + version: Node version string (default: "1.0.0") + cache_ttl: Cache time-to-live in seconds (default: 300.0) + operation_keywords: Optional set of keywords to identify operation methods. + Methods containing these keywords are reported as operations. + If None, uses DEFAULT_OPERATION_KEYWORDS. + exclude_prefixes: Optional set of prefixes to exclude from capability + discovery. Methods starting with these prefixes are filtered out. + If None, uses DEFAULT_EXCLUDE_PREFIXES. + + Raises: + ValueError: If node_id or node_type is empty + + Example: + ```python + class MyNode(MixinNodeIntrospection): + def __init__(self, config): + self.initialize_introspection( + node_id=config.node_id, + node_type="EFFECT", + event_bus=config.event_bus, + version="1.2.0", + ) + + # With custom operation keywords + class MyEffectNode(MixinNodeIntrospection): + def __init__(self, config): + self.initialize_introspection( + node_id=config.node_id, + node_type="EFFECT", + event_bus=config.event_bus, + operation_keywords={"fetch", "upload", "download"}, + ) + ``` + """ + if not node_id: + raise ValueError("node_id cannot be empty") + if not node_type: + raise ValueError("node_type cannot be empty") + + # Configuration + self._introspection_node_id = node_id + self._introspection_node_type = node_type + self._introspection_event_bus = event_bus + self._introspection_version = version + self._introspection_cache_ttl = cache_ttl + + # Capability discovery configuration - use copies to avoid mutation + self._introspection_operation_keywords = ( + operation_keywords + if operation_keywords is not None + else self.DEFAULT_OPERATION_KEYWORDS.copy() + ) + self._introspection_exclude_prefixes = ( + exclude_prefixes + if exclude_prefixes is not None + else self.DEFAULT_EXCLUDE_PREFIXES.copy() + ) + + # State + self._introspection_cache = None + self._introspection_cached_at = None + self._introspection_start_time = time.time() + + # Background tasks + self._heartbeat_task = None + self._registry_listener_task = None + self._introspection_stop_event = asyncio.Event() + self._registry_unsubscribe = None + + # Registry listener callback error tracking + # Used for rate-limiting error logging to prevent log spam + self._registry_callback_consecutive_failures = 0 + self._registry_callback_last_failure_time = 0.0 + # Only log every Nth consecutive failure to prevent log spam + self._registry_callback_failure_log_threshold = 5 + + # Performance metrics tracking + self._introspection_last_metrics = None + + if event_bus is None: + logger.warning( + f"Introspection initialized without event bus for {node_id}", + extra={ + "node_id": node_id, + "node_type": node_type, + }, + ) + + logger.debug( + f"Introspection initialized for {node_id}", + extra={ + "node_id": node_id, + "node_type": node_type, + "version": version, + "cache_ttl": cache_ttl, + "has_event_bus": event_bus is not None, + "operation_keywords_count": len(self._introspection_operation_keywords), + "exclude_prefixes_count": len(self._introspection_exclude_prefixes), + }, + ) + + def _ensure_initialized(self) -> None: + """Ensure introspection has been initialized. + + This method validates that `initialize_introspection()` was called + before using introspection methods. It should be called at the start + of public entry point methods. + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + async def get_introspection_data(self) -> ModelNodeIntrospectionEvent: + self._ensure_initialized() + # ... rest of method + ``` + """ + if self._introspection_node_id is None: + raise RuntimeError( + "MixinNodeIntrospection not initialized. " + "Call initialize_introspection() before using introspection methods." + ) + + def _get_class_method_signatures(self) -> dict[str, str]: + """Get method signatures from class-level cache. + + This method returns cached method signatures for the current class, + populating the cache on first access. The cache is shared across all + instances of the same class, avoiding expensive reflection operations + on each introspection call. + + Security Note: + This method uses Python's ``inspect`` module to extract method + signatures, which exposes detailed type information: + + - Parameter names may reveal business logic (e.g., ``user_id``, + ``payment_token``, ``decrypt_key``) + - Type annotations expose internal data structures + - Return types reveal output formats + + **Filtering Applied**: + + - Only public methods (not starting with ``_``) are included + - Methods without inspectable signatures get ``(...)`` placeholder + + **Mitigation**: + + - Use generic parameter names for public methods + - Prefix sensitive helper methods with ``_`` + + Returns: + Dictionary mapping public method names to signature strings. + + Note: + The cache is populated lazily on first access and persists for + the lifetime of the class. Use `_invalidate_class_method_cache()` + if methods are added dynamically at runtime. + + Example: + ```python + # First call populates cache + signatures = self._get_class_method_signatures() + # {"execute": "(query: str) -> list[dict]", ...} + + # Subsequent calls return cached data + signatures = self._get_class_method_signatures() + ``` + """ + cls = type(self) + if cls not in MixinNodeIntrospection._class_method_cache: + # Populate cache for this class + signatures: dict[str, str] = {} + for name in dir(self): + if name.startswith("_"): + continue + attr = getattr(self, name, None) + if callable(attr) and inspect.ismethod(attr): + try: + sig = inspect.signature(attr) + signatures[name] = str(sig) + except (ValueError, TypeError): + # Some methods don't have inspectable signatures + signatures[name] = "(...)" + MixinNodeIntrospection._class_method_cache[cls] = signatures + return MixinNodeIntrospection._class_method_cache[cls] + + @classmethod + def _invalidate_class_method_cache(cls, target_class: type | None = None) -> None: + """Invalidate the class-level method signature cache. + + Call this method when methods are dynamically added or removed from + a class at runtime. For most use cases, this is not necessary as + class methods are defined at class creation time. + + Args: + target_class: Specific class to invalidate cache for. + If None, clears cache for all classes. + + Example: + ```python + # Invalidate cache for a specific class + MixinNodeIntrospection._invalidate_class_method_cache(MyNodeClass) + + # Invalidate cache for all classes + MixinNodeIntrospection._invalidate_class_method_cache() + ``` + + Note: + This is typically only needed in testing scenarios or when + using dynamic method registration patterns. + """ + if target_class is not None: + cls._class_method_cache.pop(target_class, None) + else: + cls._class_method_cache.clear() + + def _should_skip_method(self, method_name: str) -> bool: + """Check if method should be excluded from capability discovery. + + Uses the configured exclude_prefixes set for efficient prefix matching. + + Args: + method_name: Name of the method to check + + Returns: + True if method should be skipped, False otherwise + """ + return any( + method_name.startswith(prefix) + for prefix in self._introspection_exclude_prefixes + ) + + def _is_operation_method(self, method_name: str) -> bool: + """Check if method name indicates an operation. + + Uses the configured operation_keywords set to identify methods + that represent node operations. + + Args: + method_name: Name of the method to check + + Returns: + True if method appears to be an operation, False otherwise + """ + name_lower = method_name.lower() + return any( + keyword in name_lower for keyword in self._introspection_operation_keywords + ) + + def _discover_protocols(self) -> list[str]: + """Discover protocol and mixin implementations from class hierarchy. + + Security Note: + This method exposes the inheritance hierarchy by returning class + names that start with ``Protocol`` or ``Mixin``. This reveals what + capabilities the node implements (e.g., ``ProtocolDatabaseAdapter``, + ``MixinAsyncCircuitBreaker``). This information is generally safe + to expose as it describes the node's public interface contracts, + but be aware that it may reveal architectural decisions. + + Returns: + List of protocol and mixin class names implemented by this class + """ + protocols: list[str] = [] + for base in type(self).__mro__: + base_name = base.__name__ + if base_name.startswith(("Protocol", "Mixin")): + protocols.append(base_name) + return protocols + + def _has_fsm_state(self) -> bool: + """Check if this class has FSM state management. + + Looks for common FSM state attribute patterns. + + Returns: + True if FSM state attributes are found, False otherwise + """ + fsm_indicators = {"_state", "current_state", "_current_state", "state"} + return any(hasattr(self, indicator) for indicator in fsm_indicators) + + async def get_capabilities(self) -> CapabilitiesDict: + """Extract node capabilities via reflection. + + Uses the inspect module to discover: + - Public methods (potential operations) + - Protocol implementations + - FSM state attributes + + Method signatures are cached at the class level for performance + optimization, as they don't change after class definition. + + Security Note: + This method exposes information about the node's public interface. + The returned data includes method names, parameter signatures, and + type annotations which may reveal implementation details. + + **What Gets Exposed**: + + - Method names matching operation keywords (execute, handle, etc.) + - Full method signatures including parameter names and types + - Protocol/mixin class names from the inheritance hierarchy + - Whether FSM state management is present + + **Filtering Applied**: + + - Private methods (``_`` prefix) are excluded + - Utility methods (``get_*``, ``set_*``, ``initialize*``, etc.) are + filtered based on ``exclude_prefixes`` configuration + - Only methods containing configured ``operation_keywords`` are + listed in the ``operations`` field + + **Best Practices**: + + - Review this output before production deployment + - Use generic operation names (e.g., ``process_request`` instead of + ``decrypt_and_forward_to_payment_gateway``) + - Prefix sensitive internal methods with ``_`` + - Configure additional ``exclude_prefixes`` if needed + + Returns: + Dictionary containing: + - operations: List of public method names that may be operations + - protocols: List of protocol/interface names implemented + - has_fsm: Boolean indicating if node has FSM state management + - method_signatures: Dict of method names to signature strings + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + capabilities = await node.get_capabilities() + # { + # "operations": ["execute", "query", "batch_execute"], + # "protocols": ["ProtocolDatabaseAdapter"], + # "has_fsm": True, + # "method_signatures": { + # "execute": "(query: str) -> list[dict]", + # ... + # } + # } + + # Review exposed capabilities before production + for op in capabilities["operations"]: + print(f"Exposed operation: {op}") + ``` + """ + self._ensure_initialized() + start_time = time.perf_counter() + + # Get cached method signatures (class-level, computed once per class) + # Track discovery time separately for performance analysis + discover_start = time.perf_counter() + cached_signatures = self._get_class_method_signatures() + discover_elapsed_ms = (time.perf_counter() - discover_start) * 1000 + + # Filter signatures and identify operations + operations: list[str] = [] + method_signatures: dict[str, str] = {} + + for name, sig in cached_signatures.items(): + # Skip utility methods based on configured prefixes + if self._should_skip_method(name): + continue + + # Add method signature to filtered results + method_signatures[name] = sig + + # Add methods that look like operations + if self._is_operation_method(name): + operations.append(name) + + # Build capabilities dict + capabilities: CapabilitiesDict = { + "operations": operations, + "protocols": self._discover_protocols(), + "has_fsm": self._has_fsm_state(), + "method_signatures": method_signatures, + } + + # Performance instrumentation + elapsed_ms = (time.perf_counter() - start_time) * 1000 + if elapsed_ms > PERF_THRESHOLD_GET_CAPABILITIES_MS: + logger.warning( + "Capability discovery exceeded 50ms target", + extra={ + "node_id": self._introspection_node_id, + "elapsed_ms": round(elapsed_ms, 2), + "discover_elapsed_ms": round(discover_elapsed_ms, 2), + "method_count": len(cached_signatures), + "operation_count": len(operations), + "threshold_ms": PERF_THRESHOLD_GET_CAPABILITIES_MS, + }, + ) + + return capabilities + + async def get_endpoints(self) -> dict[str, str]: + """Discover endpoint URLs for this node. + + Looks for common endpoint attributes and methods to build + a dictionary of available endpoints. + + Returns: + Dictionary mapping endpoint names to URLs. + Common keys: health, api, metrics, readiness, liveness + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + endpoints = await node.get_endpoints() + # { + # "health": "http://localhost:8080/health", + # "metrics": "http://localhost:8080/metrics", + # } + ``` + """ + self._ensure_initialized() + endpoints: dict[str, str] = {} + + # Check for endpoint attributes + endpoint_attrs = [ + ("health_url", "health"), + ("health_endpoint", "health"), + ("api_url", "api"), + ("api_endpoint", "api"), + ("metrics_url", "metrics"), + ("metrics_endpoint", "metrics"), + ("readiness_url", "readiness"), + ("readiness_endpoint", "readiness"), + ("liveness_url", "liveness"), + ("liveness_endpoint", "liveness"), + ] + + for attr_name, endpoint_name in endpoint_attrs: + if hasattr(self, attr_name): + value = getattr(self, attr_name) + if value and isinstance(value, str): + endpoints[endpoint_name] = value + + # Check for endpoint methods + endpoint_methods = [ + ("get_health_url", "health"), + ("get_api_url", "api"), + ("get_metrics_url", "metrics"), + ] + + for method_name, endpoint_name in endpoint_methods: + if hasattr(self, method_name) and endpoint_name not in endpoints: + method = getattr(self, method_name) + if callable(method): + try: + # Handle both sync and async methods + result = method() + if asyncio.iscoroutine(result): + result = await result + if result and isinstance(result, str): + endpoints[endpoint_name] = result + except Exception as e: + logger.debug( + f"Failed to get endpoint from {method_name}: {e}", + extra={"method": method_name, "error": str(e)}, + ) + + return endpoints + + async def get_current_state(self) -> str | None: + """Get the current FSM state if applicable. + + Checks common FSM state attribute patterns and returns + the current state value if found. + + Returns: + Current state string if FSM state is found, None otherwise. + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + state = await node.get_current_state() + # "connected" or None + ``` + """ + self._ensure_initialized() + # Check for state attributes in order of preference + state_attrs = ["_state", "current_state", "_current_state", "state"] + + for attr_name in state_attrs: + if hasattr(self, attr_name): + state = getattr(self, attr_name) + if state is not None: + # Handle enum states + if hasattr(state, "value"): + return str(state.value) + return str(state) + + # Check for get_state method + if hasattr(self, "get_state"): + method = self.get_state # type: ignore[attr-defined] + if callable(method): + try: + result = method() + if asyncio.iscoroutine(result): + result = await result + if result is not None: + if hasattr(result, "value"): + return str(result.value) + return str(result) + except Exception as e: + logger.debug( + f"Failed to get state from get_state method: {e}", + extra={"error": str(e)}, + ) + + return None + + async def get_introspection_data(self) -> ModelNodeIntrospectionEvent: + """Get introspection data with caching support. + + Returns cached data if available and not expired, otherwise + builds fresh introspection data and caches it. + + Performance metrics are captured for each call and stored in + ``_introspection_last_metrics``. Use ``get_performance_metrics()`` + to retrieve the most recent metrics. + + Returns: + ModelNodeIntrospectionEvent containing full introspection data. + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + data = await node.get_introspection_data() + print(f"Node {data.node_id} has capabilities: {data.capabilities}") + ``` + """ + self._ensure_initialized() + total_start = time.perf_counter() + current_time = time.time() + + # Initialize metrics for this call + metrics = IntrospectionPerformanceMetrics() + + # Check cache validity + if ( + self._introspection_cache is not None + and self._introspection_cached_at is not None + and current_time - self._introspection_cached_at + < self._introspection_cache_ttl + ): + # Return cached data with updated timestamp + cached_event = ModelNodeIntrospectionEvent(**self._introspection_cache) + + # Record cache hit metrics + elapsed_ms = (time.perf_counter() - total_start) * 1000 + metrics.total_introspection_ms = elapsed_ms + metrics.cache_hit = True + + # Check cache hit threshold + if elapsed_ms > PERF_THRESHOLD_CACHE_HIT_MS: + metrics.threshold_exceeded = True + metrics.slow_operations.append("cache_hit") + + self._introspection_last_metrics = metrics + return cached_event + + # Build fresh introspection data with timing for each component + cap_start = time.perf_counter() + capabilities = await self.get_capabilities() + metrics.get_capabilities_ms = (time.perf_counter() - cap_start) * 1000 + + # Extract method count from capabilities + method_sigs = capabilities.get("method_signatures", {}) + metrics.method_count = len(method_sigs) if isinstance(method_sigs, dict) else 0 + + endpoints_start = time.perf_counter() + endpoints = await self.get_endpoints() + metrics.get_endpoints_ms = (time.perf_counter() - endpoints_start) * 1000 + + state_start = time.perf_counter() + current_state = await self.get_current_state() + metrics.get_current_state_ms = (time.perf_counter() - state_start) * 1000 + + # Get node_id and node_type with fallback logging + # The "unknown" fallback indicates a potential initialization issue + node_id = self._introspection_node_id + if node_id is None: + logger.warning( + "Node ID not initialized, using 'unknown' - " + "ensure initialize_introspection() was called correctly", + extra={"operation": "get_introspection_data"}, + ) + node_id = "unknown" + + node_type = self._introspection_node_type + if node_type is None: + logger.warning( + "Node type not initialized, using 'unknown' - " + "ensure initialize_introspection() was called correctly", + extra={"node_id": node_id, "operation": "get_introspection_data"}, + ) + node_type = "unknown" + + event = ModelNodeIntrospectionEvent( + node_id=node_id, + node_type=node_type, + capabilities=capabilities, + endpoints=endpoints, + current_state=current_state, + version=self._introspection_version, + reason="cache_refresh", + correlation_id=uuid4(), + ) + + # Update cache - cast the model_dump output to our typed dict since we know + # the structure matches (model_dump returns dict[str, Any] by default) + self._introspection_cache = cast( + IntrospectionCacheDict, event.model_dump(mode="json") + ) + self._introspection_cached_at = current_time + + # Extract operations list with proper type narrowing + operations_value = capabilities.get("operations", []) + operations_count = ( + len(operations_value) if isinstance(operations_value, list) else 0 + ) + + # Finalize metrics + metrics.total_introspection_ms = (time.perf_counter() - total_start) * 1000 + metrics.cache_hit = False + + # Check thresholds and identify slow operations + if metrics.get_capabilities_ms > PERF_THRESHOLD_GET_CAPABILITIES_MS: + metrics.threshold_exceeded = True + metrics.slow_operations.append("get_capabilities") + + if metrics.total_introspection_ms > PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS: + metrics.threshold_exceeded = True + if "total_introspection" not in metrics.slow_operations: + metrics.slow_operations.append("total_introspection") + + # Store metrics for later retrieval + self._introspection_last_metrics = metrics + + # Log if any threshold was exceeded + if metrics.threshold_exceeded: + logger.warning( + "Introspection exceeded performance threshold", + extra={ + "node_id": self._introspection_node_id, + "total_ms": round(metrics.total_introspection_ms, 2), + "get_capabilities_ms": round(metrics.get_capabilities_ms, 2), + "get_endpoints_ms": round(metrics.get_endpoints_ms, 2), + "get_current_state_ms": round(metrics.get_current_state_ms, 2), + "method_count": metrics.method_count, + "slow_operations": metrics.slow_operations, + "threshold_ms": PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS, + }, + ) + + logger.debug( + f"Introspection data refreshed for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "capabilities_count": operations_count, + "endpoints_count": len(endpoints), + "total_ms": round(metrics.total_introspection_ms, 2), + }, + ) + + return event + + async def publish_introspection( + self, + reason: str = "startup", + correlation_id: UUID | None = None, + ) -> bool: + """Publish introspection event to the event bus. + + Gracefully degrades if event bus is unavailable - logs warning + and returns False instead of raising an exception. + + Args: + reason: Reason for the introspection event + (startup, shutdown, request, heartbeat) + correlation_id: Optional correlation ID for tracing + + Returns: + True if published successfully, False otherwise + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + # On startup + success = await node.publish_introspection(reason="startup") + + # On shutdown + success = await node.publish_introspection(reason="shutdown") + ``` + """ + self._ensure_initialized() + if self._introspection_event_bus is None: + logger.warning( + f"Cannot publish introspection - no event bus configured for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "reason": reason, + }, + ) + return False + + try: + # Get introspection data + event = await self.get_introspection_data() + + # Create publish event with updated reason and correlation_id + # Use model_copy for clean field updates (Pydantic v2) + final_correlation_id = correlation_id or uuid4() + publish_event = event.model_copy( + update={ + "reason": reason, + "correlation_id": final_correlation_id, + } + ) + + # Publish to event bus + if hasattr(self._introspection_event_bus, "publish_envelope"): + await self._introspection_event_bus.publish_envelope( # type: ignore[union-attr] + envelope=publish_event, + topic=INTROSPECTION_TOPIC, + ) + else: + # Fallback to publish method with raw bytes + event_data = publish_event.model_dump(mode="json") + value = json.dumps(event_data).encode("utf-8") + await self._introspection_event_bus.publish( + topic=INTROSPECTION_TOPIC, + key=self._introspection_node_id.encode("utf-8") + if self._introspection_node_id + else None, + value=value, + ) + + logger.info( + f"Published introspection event for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "reason": reason, + "correlation_id": str(final_correlation_id), + }, + ) + return True + + except Exception as e: + # Use error() with exc_info=True instead of exception() to include + # structured error_type and error_message fields for log aggregation + logger.error( # noqa: G201 + f"Failed to publish introspection for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "reason": reason, + "error_type": type(e).__name__, + "error_message": str(e), + }, + exc_info=True, + ) + return False + + async def _publish_heartbeat(self) -> bool: + """Publish heartbeat event to the event bus. + + Internal method for heartbeat broadcasting. Calculates uptime + and publishes heartbeat event. + + Returns: + True if published successfully, False otherwise + """ + if self._introspection_event_bus is None: + return False + + try: + # Calculate uptime + uptime_seconds = 0.0 + if self._introspection_start_time is not None: + uptime_seconds = time.time() - self._introspection_start_time + + # Get node_id and node_type with fallback logging + # The "unknown" fallback indicates a potential initialization issue + node_id_str = self._introspection_node_id + if node_id_str is None: + logger.warning( + "Node ID not initialized, using 'unknown' in heartbeat - " + "ensure initialize_introspection() was called correctly", + extra={"operation": "_publish_heartbeat"}, + ) + node_id_str = "unknown" + + node_type = self._introspection_node_type + if node_type is None: + logger.warning( + "Node type not initialized, using 'unknown' in heartbeat - " + "ensure initialize_introspection() was called correctly", + extra={"node_id": node_id_str, "operation": "_publish_heartbeat"}, + ) + node_type = "unknown" + + # Convert node_id to UUID for heartbeat model compatibility + # Try to parse as UUID first, otherwise generate deterministic UUID from string + try: + node_id_uuid = UUID(node_id_str) + except ValueError: + # Generate deterministic UUID from string using uuid5 with DNS namespace + node_id_uuid = uuid5(NAMESPACE_DNS, node_id_str) + + # Create heartbeat event + heartbeat = ModelNodeHeartbeatEvent( + node_id=node_id_uuid, + node_type=node_type, + uptime_seconds=uptime_seconds, + # TODO(OMN-XXX): Implement active operation tracking + # Currently hardcoded to 0. Full implementation requires: + # - Operation counter increment/decrement around async operations + # - Thread-safe counter for concurrent operations + # - Integration with node's actual execution context + # This is intentionally left as 0 until the tracking infrastructure + # is implemented. See MixinNodeIntrospection docstring note. + active_operations_count=0, + correlation_id=uuid4(), + ) + + # Publish to event bus + if hasattr(self._introspection_event_bus, "publish_envelope"): + await self._introspection_event_bus.publish_envelope( # type: ignore[union-attr] + envelope=heartbeat, + topic=HEARTBEAT_TOPIC, + ) + else: + value = json.dumps(heartbeat.model_dump(mode="json")).encode("utf-8") + await self._introspection_event_bus.publish( + topic=HEARTBEAT_TOPIC, + key=self._introspection_node_id.encode("utf-8") + if self._introspection_node_id + else None, + value=value, + ) + + logger.debug( + f"Published heartbeat for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "uptime_seconds": uptime_seconds, + }, + ) + return True + + except Exception as e: + # Use error() with exc_info=True instead of exception() to include + # structured error_type and error_message fields for log aggregation + logger.error( # noqa: G201 + f"Failed to publish heartbeat for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(e).__name__, + "error_message": str(e), + }, + exc_info=True, + ) + return False + + async def _heartbeat_loop(self, interval: float) -> None: + """Background loop for periodic heartbeat publishing. + + Runs until stop event is set, publishing heartbeats at the + specified interval. + + Args: + interval: Time between heartbeats in seconds + """ + # Ensure stop event is initialized + if self._introspection_stop_event is None: + self._introspection_stop_event = asyncio.Event() + + logger.info( + f"Starting heartbeat loop for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "interval_seconds": interval, + }, + ) + + while not self._introspection_stop_event.is_set(): + try: + await self._publish_heartbeat() + except asyncio.CancelledError: + logger.debug( + f"Heartbeat loop cancelled for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + break + except Exception as e: + # Use error() with exc_info=True instead of exception() to include + # structured error_type and error_message fields for log aggregation + logger.error( # noqa: G201 + f"Error in heartbeat loop for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(e).__name__, + "error_message": str(e), + }, + exc_info=True, + ) + + # Wait for next interval or stop event + try: + await asyncio.wait_for( + self._introspection_stop_event.wait(), + timeout=interval, + ) + # Stop event was set + break + except TimeoutError: + # Normal timeout, continue loop + pass + + logger.info( + f"Heartbeat loop stopped for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + async def _registry_listener_loop( + self, + max_retries: int = 3, + base_backoff_seconds: float = 1.0, + ) -> None: + """Background loop listening for REQUEST_INTROSPECTION events. + + Subscribes to the request_introspection topic and responds + with introspection data when requests are received. Includes + retry logic with exponential backoff for subscription failures. + + Security Note: + This method subscribes to the ``node.request_introspection`` Kafka + topic and responds with full introspection data to any request. + This creates a network-accessible endpoint for capability discovery. + + **Network Exposure**: + + - Any consumer on the Kafka cluster can request introspection data + - Responses are published to ``node.introspection`` topic + - No authentication is performed on incoming requests + + **Multi-tenant Considerations**: + + - Configure Kafka topic ACLs to restrict access to introspection + topics in multi-tenant environments + - Consider whether introspection topics should be accessible + outside the cluster boundary + - Monitor topic consumers for unauthorized access patterns + - Use separate Kafka clusters for different security domains + + **Request Validation**: + + - The ``target_node_id`` field allows filtering requests to + specific nodes - only matching requests are processed + - Malformed requests are handled gracefully without crashing + - Correlation IDs are validated but invalid IDs don't block + processing + + Args: + max_retries: Maximum subscription retry attempts (default: 3) + base_backoff_seconds: Base backoff time for exponential retry + """ + if self._introspection_event_bus is None: + logger.warning( + f"Cannot start registry listener - no event bus for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + return + + # Ensure stop event is initialized + if self._introspection_stop_event is None: + self._introspection_stop_event = asyncio.Event() + + logger.info( + f"Starting registry listener for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + def _parse_correlation_id(raw_value: str | None) -> UUID | None: + """Parse correlation ID from request data with graceful fallback. + + Args: + raw_value: Raw correlation_id value from request JSON + + Returns: + Parsed UUID or None if parsing fails or value is empty + """ + if not raw_value: + return None + + try: + # UUID() raises ValueError for malformed strings, + # TypeError for non-string inputs (e.g., int, list). + # Convert to string first for safer handling of unexpected types. + return UUID(str(raw_value)) + except (ValueError, TypeError) as e: + # Log warning with structured fields for monitoring. + # Truncate received value preview to avoid log bloat + # from potentially malicious oversized input. + logger.warning( + "Invalid correlation_id format in introspection " + "request, generating new correlation_id", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(e).__name__, + "error_message": str(e), + "received_value_type": type(raw_value).__name__, + "received_value_preview": str(raw_value)[:50], + }, + ) + return None + + def _should_log_failure(consecutive_failures: int, threshold: int) -> bool: + """Determine if failure should be logged based on rate limiting. + + Logs first failure and every Nth consecutive failure to prevent log spam. + + Args: + consecutive_failures: Current consecutive failure count + threshold: Log every Nth failure + + Returns: + True if this failure should be logged at error level + """ + return consecutive_failures == 1 or consecutive_failures % threshold == 0 + + async def on_request(message: ModelEventMessage) -> None: + """Handle incoming introspection request. + + Includes error recovery with rate-limited logging to prevent + log spam during sustained failures. Continues processing on + non-fatal errors to maintain graceful degradation. + """ + try: + # Early exit if message has no parseable value + if not hasattr(message, "value") or not message.value: + await self.publish_introspection( + reason="request", + correlation_id=None, + ) + self._registry_callback_consecutive_failures = 0 + return + + # Parse request data + request_data = json.loads(message.value.decode("utf-8")) + + # Check if request targets a specific node (early exit if not us) + target_node_id = request_data.get("target_node_id") + if target_node_id and target_node_id != self._introspection_node_id: + return + + # Parse correlation ID with graceful fallback + correlation_id = _parse_correlation_id( + request_data.get("correlation_id") + ) + + # Respond with introspection data + await self.publish_introspection( + reason="request", + correlation_id=correlation_id, + ) + + # Reset failure counter on success + self._registry_callback_consecutive_failures = 0 + + except Exception as e: + # Track consecutive failures for rate-limited logging + self._registry_callback_consecutive_failures += 1 + self._registry_callback_last_failure_time = time.time() + + # Rate-limit error logging to prevent log spam during sustained failures + if _should_log_failure( + self._registry_callback_consecutive_failures, + self._registry_callback_failure_log_threshold, + ): + logger.error( # noqa: G201 + f"Error handling introspection request for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(e).__name__, + "error_message": str(e), + "consecutive_failures": ( + self._registry_callback_consecutive_failures + ), + "log_rate_limited": ( + self._registry_callback_consecutive_failures > 1 + ), + }, + exc_info=True, + ) + else: + # Log at debug level for rate-limited failures + logger.debug( + f"Suppressed error log for introspection request " + f"(failure {self._registry_callback_consecutive_failures})", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(e).__name__, + "consecutive_failures": ( + self._registry_callback_consecutive_failures + ), + }, + ) + + # Continue processing - graceful degradation + # The callback should not raise exceptions that would disrupt the listener + + # Helper function to clean up subscription + async def cleanup_subscription() -> None: + """Clean up the current subscription.""" + if self._registry_unsubscribe is not None: + try: + result = self._registry_unsubscribe() + if asyncio.iscoroutine(result): + await result + except Exception as cleanup_error: + logger.debug( + "Error unsubscribing registry listener for " + f"{self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(cleanup_error).__name__, + "error_message": str(cleanup_error), + }, + ) + self._registry_unsubscribe = None + + # Retry loop with exponential backoff for subscription failures + retry_count = 0 + while not self._introspection_stop_event.is_set(): + try: + # Subscribe to request topic + if hasattr(self._introspection_event_bus, "subscribe"): + unsubscribe = await self._introspection_event_bus.subscribe( + topic=REQUEST_INTROSPECTION_TOPIC, + group_id=f"introspection-{self._introspection_node_id}", + on_message=on_request, + ) + self._registry_unsubscribe = unsubscribe + + # Reset retry count on successful subscription + retry_count = 0 + + logger.info( + f"Registry listener subscribed for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "topic": REQUEST_INTROSPECTION_TOPIC, + }, + ) + + # Wait for stop signal + await self._introspection_stop_event.wait() + # Stop signal received, exit loop + break + + logger.warning( + "Event bus does not support subscribe for " + f"{self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + break + + except asyncio.CancelledError: + logger.debug( + f"Registry listener cancelled for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + break + except Exception as e: + retry_count += 1 + logger.error( # noqa: G201 + f"Error in registry listener for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "error_type": type(e).__name__, + "error_message": str(e), + "retry_count": retry_count, + "max_retries": max_retries, + }, + exc_info=True, + ) + + # Clean up any partial subscription before retry + await cleanup_subscription() + + # Check if we should retry + if retry_count >= max_retries: + # Use error() with exc_info=True instead of exception() + # to include structured error_type and error_message fields + # for log aggregation + logger.error( # noqa: G201 + "Registry listener exhausted retries", + extra={ + "node_id": self._introspection_node_id, + "retry_count": retry_count, + "max_retries": max_retries, + "error_type": type(e).__name__, + "error_message": str(e), + }, + exc_info=True, + ) + break + + # Exponential backoff before retry + backoff = base_backoff_seconds * (2 ** (retry_count - 1)) + logger.info( + f"Registry listener retrying in {backoff}s for " + f"{self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "backoff_seconds": backoff, + "retry_count": retry_count, + }, + ) + + # Wait for backoff period or stop signal + try: + await asyncio.wait_for( + self._introspection_stop_event.wait(), + timeout=backoff, + ) + # Stop signal received during backoff + break + except TimeoutError: + # Normal timeout, continue to retry + pass + + # Final cleanup + await cleanup_subscription() + + logger.info( + f"Registry listener stopped for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + async def start_introspection_tasks( + self, + enable_heartbeat: bool = True, + heartbeat_interval_seconds: float = 30.0, + enable_registry_listener: bool = True, + ) -> None: + """Start background introspection tasks. + + Starts the heartbeat loop and/or registry listener as background + tasks. Safe to call multiple times - won't start duplicate tasks. + + Args: + enable_heartbeat: Whether to start the heartbeat loop + heartbeat_interval_seconds: Interval between heartbeats in seconds + enable_registry_listener: Whether to start the registry listener + + Raises: + RuntimeError: If initialize_introspection() was not called. + + Example: + ```python + await node.start_introspection_tasks( + enable_heartbeat=True, + heartbeat_interval_seconds=30.0, + enable_registry_listener=True, + ) + ``` + """ + self._ensure_initialized() + # Reset stop event if previously set + if self._introspection_stop_event is None: + self._introspection_stop_event = asyncio.Event() + elif self._introspection_stop_event.is_set(): + self._introspection_stop_event.clear() + + # Start heartbeat task if enabled and not running + if enable_heartbeat and self._heartbeat_task is None: + self._heartbeat_task = asyncio.create_task( + self._heartbeat_loop(heartbeat_interval_seconds), + name=f"heartbeat-{self._introspection_node_id}", + ) + logger.debug( + f"Started heartbeat task for {self._introspection_node_id}", + extra={ + "node_id": self._introspection_node_id, + "interval": heartbeat_interval_seconds, + }, + ) + + # Start registry listener if enabled and not running + if enable_registry_listener and self._registry_listener_task is None: + self._registry_listener_task = asyncio.create_task( + self._registry_listener_loop(), + name=f"registry-listener-{self._introspection_node_id}", + ) + logger.debug( + f"Started registry listener task for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + async def stop_introspection_tasks(self) -> None: + """Stop all background introspection tasks. + + Signals tasks to stop and waits for clean shutdown. + Safe to call multiple times. + + Example: + ```python + await node.stop_introspection_tasks() + ``` + """ + logger.info( + f"Stopping introspection tasks for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + # Signal tasks to stop + if self._introspection_stop_event is not None: + self._introspection_stop_event.set() + + # Cancel and wait for heartbeat task + if self._heartbeat_task is not None: + self._heartbeat_task.cancel() + try: + await self._heartbeat_task + except asyncio.CancelledError: + pass + self._heartbeat_task = None + + # Cancel and wait for registry listener task + if self._registry_listener_task is not None: + self._registry_listener_task.cancel() + try: + await self._registry_listener_task + except asyncio.CancelledError: + pass + self._registry_listener_task = None + + logger.info( + f"Introspection tasks stopped for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + def invalidate_introspection_cache(self) -> None: + """Invalidate the introspection cache. + + Call this when node capabilities change to ensure fresh + data is reported on next introspection request. + + Example: + ```python + node.register_new_handler(handler) + node.invalidate_introspection_cache() + ``` + """ + self._introspection_cache = None + self._introspection_cached_at = None + logger.debug( + f"Introspection cache invalidated for {self._introspection_node_id}", + extra={"node_id": self._introspection_node_id}, + ) + + def get_performance_metrics(self) -> IntrospectionPerformanceMetrics | None: + """Get the most recent performance metrics from introspection operations. + + Returns the performance metrics captured during the last call to + ``get_introspection_data()``. Use this to monitor introspection + performance and detect when operations exceed the <50ms threshold. + + Returns: + IntrospectionPerformanceMetrics if introspection has been called, + None if no introspection has been performed yet. + + Example: + ```python + # After calling introspection + await node.get_introspection_data() + + # Check performance metrics + metrics = node.get_performance_metrics() + if metrics and metrics.threshold_exceeded: + logger.warning( + "Slow introspection detected", + extra={ + "slow_operations": metrics.slow_operations, + "total_ms": metrics.total_introspection_ms, + } + ) + + # Access individual timings + if metrics: + print(f"Total time: {metrics.total_introspection_ms:.2f}ms") + print(f"Cache hit: {metrics.cache_hit}") + print(f"Methods discovered: {metrics.method_count}") + ``` + """ + return self._introspection_last_metrics + + +__all__ = [ + "MixinNodeIntrospection", + "INTROSPECTION_TOPIC", + "HEARTBEAT_TOPIC", + "REQUEST_INTROSPECTION_TOPIC", + "CapabilitiesDict", + "IntrospectionCacheDict", + "IntrospectionPerformanceMetrics", + "PERF_THRESHOLD_GET_CAPABILITIES_MS", + "PERF_THRESHOLD_DISCOVER_CAPABILITIES_MS", + "PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS", + "PERF_THRESHOLD_CACHE_HIT_MS", +] diff --git a/src/omnibase_infra/models/__init__.py b/src/omnibase_infra/models/__init__.py index 80f101d394..547f5944c1 100644 --- a/src/omnibase_infra/models/__init__.py +++ b/src/omnibase_infra/models/__init__.py @@ -2,17 +2,22 @@ # Copyright (c) 2025 OmniNode Team """ONEX Infrastructure Models. -This module provides shared models for the omnibase_infra package. +This module exports all infrastructure-specific Pydantic models. """ from omnibase_infra.models.registration import ( + ModelNodeCapabilities, ModelNodeHeartbeatEvent, ModelNodeIntrospectionEvent, + ModelNodeMetadata, ModelNodeRegistration, ) __all__ = [ + # Registration models + "ModelNodeCapabilities", "ModelNodeHeartbeatEvent", "ModelNodeIntrospectionEvent", + "ModelNodeMetadata", "ModelNodeRegistration", ] diff --git a/src/omnibase_infra/models/discovery/__init__.py b/src/omnibase_infra/models/discovery/__init__.py new file mode 100644 index 0000000000..3710375944 --- /dev/null +++ b/src/omnibase_infra/models/discovery/__init__.py @@ -0,0 +1,11 @@ +# Copyright 2025 OmniNode Team. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Discovery models for node introspection and capability reporting.""" + +from omnibase_infra.models.discovery.model_node_introspection_event import ( + ModelNodeIntrospectionEvent, +) + +__all__ = [ + "ModelNodeIntrospectionEvent", +] diff --git a/src/omnibase_infra/models/discovery/model_node_introspection_event.py b/src/omnibase_infra/models/discovery/model_node_introspection_event.py new file mode 100644 index 0000000000..d4618f2b38 --- /dev/null +++ b/src/omnibase_infra/models/discovery/model_node_introspection_event.py @@ -0,0 +1,141 @@ +# Copyright 2025 OmniNode Team. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Node introspection event model for capability discovery and reporting.""" + +from datetime import UTC, datetime +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + + +class ModelNodeIntrospectionEvent(BaseModel): + """Event model for node introspection and capability discovery. + + This model represents introspection events emitted by nodes to report their + capabilities, endpoints, and current state. Used by MixinNodeIntrospection + to broadcast node information to the event bus for service discovery and + registry coordination. + + Attributes: + node_id: Unique identifier for the node instance. + node_type: Type classification of the node (e.g., EFFECT, COMPUTE). + capabilities: Dictionary of node capabilities discovered via reflection. + endpoints: Dictionary of endpoint URLs (health, api, metrics). + current_state: Current FSM state if the node has state management. + version: Node version string. + reason: Reason for the introspection event (startup, shutdown, request). + correlation_id: Optional correlation ID for tracing. + timestamp: UTC timestamp when the introspection was generated. + + Example: + ```python + event = ModelNodeIntrospectionEvent( + node_id="node-postgres-adapter-001", + node_type="EFFECT", + capabilities={ + "operations": ["execute", "query", "batch_execute"], + "protocols": ["ProtocolDatabaseAdapter"], + "has_fsm": True, + }, + endpoints={ + "health": "http://localhost:8080/health", + "metrics": "http://localhost:8080/metrics", + }, + current_state="connected", + version="1.0.0", + reason="startup", + ) + ``` + """ + + node_id: str = Field(..., description="Unique node identifier") + node_type: str = Field(..., description="Node type classification") + + # Capabilities discovered via reflection + capabilities: dict[str, list[str] | bool | dict[str, str]] = Field( + default_factory=dict, + description="Node capabilities discovered via reflection", + ) + + # Endpoint URLs + # + # Design Decision: We use dict[str, str] instead of dict[str, HttpUrl] for endpoints. + # + # Rationale: + # 1. Flexibility: Internal endpoints may use non-standard schemes (e.g., "consul://", + # "grpc://", or relative paths like "/health") that HttpUrl rejects. + # 2. Simplicity: Endpoint validation is deferred to the consuming services, which + # know their specific requirements and can apply appropriate validation. + # 3. Performance: HttpUrl validation adds parsing overhead for every endpoint, + # which is unnecessary for internal service discovery where endpoints are + # typically constructed programmatically. + # 4. Consistency: Other infrastructure components (e.g., Consul, Kubernetes) + # also use string-based endpoint formats. + # + # If strict URL validation is needed, consumers can wrap endpoint access with + # validation logic or use pydantic.HttpUrl when processing individual endpoints. + endpoints: dict[str, str] = Field( + default_factory=dict, + description="Endpoint URLs (health, api, metrics). Values are strings to " + "support various URL schemes including internal service protocols. " + "URL validation is deferred to consuming services.", + ) + + # State information + current_state: str | None = Field( + default=None, + description="Current FSM state if applicable", + ) + + # Version information + version: str = Field( + default="1.0.0", + description="Node version string", + ) + + # Event metadata + reason: str = Field( + default="startup", + description="Reason for introspection event (startup, shutdown, request, heartbeat)", + ) + + # Tracing + correlation_id: UUID | None = Field( + default=None, + description="Correlation ID for distributed tracing", + ) + + timestamp: datetime = Field( + default_factory=lambda: datetime.now(UTC), + description="UTC timestamp of introspection generation", + ) + + model_config = ConfigDict( + frozen=False, + extra="forbid", + json_schema_extra={ + "examples": [ + { + "node_id": "node-postgres-adapter-001", + "node_type": "EFFECT", + "capabilities": { + "operations": ["execute", "query", "batch_execute"], + "protocols": ["ProtocolDatabaseAdapter"], + "has_fsm": True, + }, + "endpoints": { + "health": "http://localhost:8080/health", + "metrics": "http://localhost:8080/metrics", + }, + "current_state": "connected", + "version": "1.0.0", + "reason": "startup", + "correlation_id": "550e8400-e29b-41d4-a716-446655440000", + "timestamp": "2025-01-15T10:30:00Z", + } + ] + }, + ) + + +__all__ = ["ModelNodeIntrospectionEvent"] diff --git a/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py b/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py index 857c0b9843..9d7380dd85 100644 --- a/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py +++ b/src/omnibase_infra/plugins/examples/plugin_json_normalizer.py @@ -165,7 +165,7 @@ def _sort_keys_recursively( # Early exit for primitives (most common case in large structures) # This optimization avoids isinstance checks for dict/list on every primitive - if not isinstance(obj, (dict, list)): + if not isinstance(obj, dict | list): return obj if isinstance(obj, dict): @@ -239,7 +239,7 @@ def validate_input(self, input_data: PluginInputData) -> None: def _is_json_compatible(self, value: object) -> bool: """Type guard to check if value is JSON-compatible.""" - return isinstance(value, (dict, list, str, int, float, bool, type(None))) + return isinstance(value, dict | list | str | int | float | bool | type(None)) def _validate_json_structure(self, obj: JsonValue, _depth: int = 0) -> None: """Recursively validate JSON structure for non-JSON-compatible types. diff --git a/src/omnibase_infra/runtime/runtime_host_process.py b/src/omnibase_infra/runtime/runtime_host_process.py index b8c63d267e..d859328b8c 100644 --- a/src/omnibase_infra/runtime/runtime_host_process.py +++ b/src/omnibase_infra/runtime/runtime_host_process.py @@ -251,7 +251,7 @@ def __init__( # Values outside bounds are clamped with a warning _drain_timeout_raw = config.get("drain_timeout_seconds") drain_timeout_value: float = DEFAULT_DRAIN_TIMEOUT_SECONDS - if isinstance(_drain_timeout_raw, (int, float)): + if isinstance(_drain_timeout_raw, int | float): drain_timeout_value = float(_drain_timeout_raw) elif isinstance(_drain_timeout_raw, str): try: diff --git a/tests/unit/mixins/test_mixin_async_circuit_breaker_race_conditions.py b/tests/unit/mixins/test_mixin_async_circuit_breaker_race_conditions.py index a968c5d95c..f646e694b3 100644 --- a/tests/unit/mixins/test_mixin_async_circuit_breaker_race_conditions.py +++ b/tests/unit/mixins/test_mixin_async_circuit_breaker_race_conditions.py @@ -591,7 +591,7 @@ async def cycle_circuit() -> None: # Filter out expected InfraUnavailableError unexpected_errors = [ - e for e in errors if not isinstance(e, (ValueError, InfraUnavailableError)) + e for e in errors if not isinstance(e, ValueError | InfraUnavailableError) ] assert len(unexpected_errors) == 0, f"Unexpected errors: {unexpected_errors}" diff --git a/tests/unit/mixins/test_mixin_node_introspection.py b/tests/unit/mixins/test_mixin_node_introspection.py new file mode 100644 index 0000000000..3f3d48a9e1 --- /dev/null +++ b/tests/unit/mixins/test_mixin_node_introspection.py @@ -0,0 +1,2257 @@ +# SPDX-License-Identifier: MIT +# Copyright (c) 2025 OmniNode Team +""" +Unit tests for MixinNodeIntrospection. + +This test suite validates: +- Initialization and attribute setup +- Capability extraction via reflection +- Endpoint discovery for health checks and operations +- FSM state extraction +- Caching behavior with TTL expiration +- Event bus publishing (with and without event bus) +- Background task management (heartbeat) +- Graceful degradation on errors +- Performance requirements (<50ms with CI buffer) + +Test Organization: + - TestMixinNodeIntrospectionInit: Initialization tests + - TestMixinNodeIntrospectionCapabilities: Capability extraction + - TestMixinNodeIntrospectionEndpoints: Endpoint discovery + - TestMixinNodeIntrospectionState: FSM state extraction + - TestMixinNodeIntrospectionCaching: Caching behavior + - TestMixinNodeIntrospectionPublishing: Event bus publishing + - TestMixinNodeIntrospectionTasks: Background tasks + - TestMixinNodeIntrospectionGracefulDegradation: Error handling + - TestMixinNodeIntrospectionPerformance: Performance requirements + - TestMixinNodeIntrospectionBenchmark: Detailed benchmarks with instrumentation + - TestMixinNodeIntrospectionEdgeCases: Edge cases and boundary conditions + +Coverage Goals: + - >90% code coverage for mixin + - All public methods tested + - Error paths validated + - Performance requirements verified +""" + +import asyncio +import json +import os +import time +from uuid import uuid4 + +import pytest + +from omnibase_infra.mixins.mixin_node_introspection import ( + PERF_THRESHOLD_CACHE_HIT_MS, + PERF_THRESHOLD_GET_CAPABILITIES_MS, + PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS, + IntrospectionPerformanceMetrics, + MixinNodeIntrospection, +) +from omnibase_infra.models.discovery import ModelNodeIntrospectionEvent + +# CI environments may be slower - apply multiplier for performance thresholds +_CI_MODE: bool = os.environ.get("CI", "false").lower() == "true" +PERF_MULTIPLIER: float = 3.0 if _CI_MODE else 2.0 + +# Type alias for event bus published event structure +PublishedEventDict = dict[ + str, str | bytes | None | dict[str, str | int | bool | list[str]] +] + + +class MockEventBus: + """Mock event bus for testing introspection publishing.""" + + def __init__(self, should_fail: bool = False) -> None: + """Initialize mock event bus. + + Args: + should_fail: If True, publish operations will raise exceptions. + """ + self.should_fail = should_fail + self.published_envelopes: list[tuple[ModelNodeIntrospectionEvent, str]] = [] + self.published_events: list[PublishedEventDict] = [] + + async def publish_envelope( + self, + envelope: ModelNodeIntrospectionEvent, + topic: str, + ) -> None: + """Mock publish_envelope method. + + Args: + envelope: Event envelope to publish. + topic: Event topic. + + Raises: + RuntimeError: If should_fail is True. + """ + if self.should_fail: + raise RuntimeError("Event bus publish failed") + self.published_envelopes.append((envelope, topic)) + + async def publish( + self, + topic: str, + key: bytes | None, + value: bytes, + ) -> None: + """Mock publish method (fallback). + + Args: + topic: Event topic. + key: Event key. + value: Event payload as bytes. + + Raises: + RuntimeError: If should_fail is True. + """ + if self.should_fail: + raise RuntimeError("Event bus publish failed") + + self.published_events.append( + { + "topic": topic, + "key": key, + "value": json.loads(value.decode("utf-8")), + } + ) + + +class MockNode(MixinNodeIntrospection): + """Mock node class for testing the mixin.""" + + def __init__(self) -> None: + """Initialize mock node with test state.""" + self._state = "idle" + self.health_url = "http://localhost:8080/health" + self.metrics_url = "http://localhost:8080/metrics" + + async def execute(self, operation: str, payload: dict[str, str]) -> dict[str, str]: + """Mock execute method. + + Args: + operation: Operation to execute. + payload: Operation payload. + + Returns: + Operation result. + """ + _ = payload # Silence unused parameter warning + return {"result": "ok", "operation": operation} + + async def health_check(self) -> dict[str, bool | str]: + """Mock health check. + + Returns: + Health status. + """ + return {"healthy": True, "state": self._state} + + async def handle_event(self, event: dict[str, str]) -> None: + """Mock handle_event method (should be discovered as operation). + + Args: + event: Event to handle. + """ + _ = event # Silence unused parameter warning + + async def process_batch(self, items: list[dict[str, str]]) -> list[dict[str, str]]: + """Mock process method (should be discovered as operation). + + Args: + items: Items to process. + + Returns: + Processed items. + """ + return items + + +class MockNodeNoHealth(MixinNodeIntrospection): + """Mock node without health endpoint URLs.""" + + def __init__(self) -> None: + """Initialize mock node.""" + self._state = "active" + + async def process(self, data: dict[str, str]) -> dict[str, bool]: + """Mock process method. + + Args: + data: Data to process. + + Returns: + Processed result. + """ + _ = data # Silence unused parameter warning + return {"processed": True} + + +class MockNodeNoState(MixinNodeIntrospection): + """Mock node without _state attribute.""" + + async def execute(self, operation: str) -> dict[str, str]: + """Mock execute method. + + Args: + operation: Operation name. + + Returns: + Operation result. + """ + return {"executed": operation} + + +class MockNodeWithEnumState(MixinNodeIntrospection): + """Mock node with enum-style state.""" + + def __init__(self) -> None: + """Initialize mock node.""" + + class State: + """Mock state class with value attribute to simulate enum-style state.""" + + value: str = "running" + + self._state: State = State() + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionInit: + """Tests for introspection initialization.""" + + async def test_initialize_introspection_sets_attributes(self) -> None: + """Test that initialize_introspection properly sets all attributes.""" + node = MockNode() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=None, + ) + + assert node._introspection_node_id == "test-node-001" + assert node._introspection_node_type == "EFFECT" + assert node._introspection_event_bus is None + assert node._introspection_version == "1.0.0" + assert node._introspection_start_time is not None + + async def test_initialize_introspection_with_event_bus(self) -> None: + """Test initialization with event bus.""" + node = MockNode() + event_bus = MockEventBus() + + node.initialize_introspection( + node_id="test-node-002", + node_type="COMPUTE", + event_bus=event_bus, + ) + + assert node._introspection_event_bus is event_bus + + async def test_initialize_introspection_custom_cache_ttl(self) -> None: + """Test initialization with custom cache TTL.""" + node = MockNode() + node.initialize_introspection( + node_id="test-node-003", + node_type="REDUCER", + event_bus=None, + cache_ttl=120.0, + ) + + assert node._introspection_cache_ttl == 120.0 + + async def test_initialize_introspection_custom_version(self) -> None: + """Test initialization with custom version.""" + node = MockNode() + node.initialize_introspection( + node_id="test-node-004", + node_type="ORCHESTRATOR", + event_bus=None, + version="2.1.0", + ) + + assert node._introspection_version == "2.1.0" + + async def test_initialize_introspection_defaults(self) -> None: + """Test initialization uses correct defaults.""" + node = MockNode() + node.initialize_introspection( + node_id="test-node-005", + node_type="EFFECT", + event_bus=None, + ) + + # Default cache TTL is 300 seconds + assert node._introspection_cache_ttl == 300.0 + # Default version is 1.0.0 + assert node._introspection_version == "1.0.0" + # Cache starts empty + assert node._introspection_cache is None + assert node._introspection_cached_at is None + + async def test_initialize_introspection_empty_node_id_raises(self) -> None: + """Test that empty node_id raises ValueError.""" + node = MockNode() + + with pytest.raises(ValueError, match="node_id cannot be empty"): + node.initialize_introspection( + node_id="", + node_type="EFFECT", + ) + + async def test_initialize_introspection_empty_node_type_raises(self) -> None: + """Test that empty node_type raises ValueError.""" + node = MockNode() + + with pytest.raises(ValueError, match="node_type cannot be empty"): + node.initialize_introspection( + node_id="test-node", + node_type="", + ) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionCapabilities: + """Tests for capability extraction.""" + + @pytest.fixture + def mock_node(self) -> MockNode: + """Create initialized mock node fixture. + + Returns: + Initialized MockNode instance. + """ + node = MockNode() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=None, + ) + return node + + async def test_get_capabilities_extracts_operations( + self, mock_node: MockNode + ) -> None: + """Test that get_capabilities extracts operation methods.""" + capabilities = await mock_node.get_capabilities() + + # Should discover methods with operation keywords + operations = capabilities["operations"] + assert isinstance(operations, list) + assert "execute" in operations + assert "handle_event" in operations + assert "process_batch" in operations + + async def test_get_capabilities_excludes_private_methods( + self, mock_node: MockNode + ) -> None: + """Test that get_capabilities excludes private methods.""" + capabilities = await mock_node.get_capabilities() + + # Private methods should not be in operations + operations = capabilities["operations"] + assert isinstance(operations, list) + for op in operations: + assert not op.startswith("_") + + async def test_get_capabilities_detects_fsm(self, mock_node: MockNode) -> None: + """Test that get_capabilities detects FSM state management.""" + capabilities = await mock_node.get_capabilities() + + # MockNode has _state attribute + assert capabilities["has_fsm"] is True + + async def test_get_capabilities_detects_protocols( + self, mock_node: MockNode + ) -> None: + """Test that get_capabilities discovers protocols.""" + capabilities = await mock_node.get_capabilities() + + # Should include MixinNodeIntrospection in protocols + protocols = capabilities["protocols"] + assert isinstance(protocols, list) + assert "MixinNodeIntrospection" in protocols + + async def test_get_capabilities_includes_method_signatures( + self, mock_node: MockNode + ) -> None: + """Test that get_capabilities captures method signatures.""" + capabilities = await mock_node.get_capabilities() + + # Should have method signatures + assert "method_signatures" in capabilities + assert isinstance(capabilities["method_signatures"], dict) + + async def test_get_capabilities_returns_dict(self, mock_node: MockNode) -> None: + """Test that get_capabilities returns a dictionary.""" + capabilities = await mock_node.get_capabilities() + + assert isinstance(capabilities, dict) + assert "operations" in capabilities + assert "protocols" in capabilities + assert "has_fsm" in capabilities + assert "method_signatures" in capabilities + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionEndpoints: + """Tests for endpoint discovery.""" + + @pytest.fixture + def mock_node(self) -> MockNode: + """Create initialized mock node fixture. + + Returns: + Initialized MockNode instance. + """ + node = MockNode() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=None, + ) + return node + + async def test_get_endpoints_discovers_health(self, mock_node: MockNode) -> None: + """Test that get_endpoints discovers health endpoint.""" + endpoints = await mock_node.get_endpoints() + + assert "health" in endpoints + assert endpoints["health"] == "http://localhost:8080/health" + + async def test_get_endpoints_discovers_metrics(self, mock_node: MockNode) -> None: + """Test that get_endpoints discovers metrics endpoint.""" + endpoints = await mock_node.get_endpoints() + + assert "metrics" in endpoints + assert endpoints["metrics"] == "http://localhost:8080/metrics" + + async def test_get_endpoints_no_endpoints(self) -> None: + """Test endpoint discovery when no endpoints defined.""" + node = MockNodeNoHealth() + node.initialize_introspection( + node_id="test-node-no-health", + node_type="EFFECT", + event_bus=None, + ) + + endpoints = await node.get_endpoints() + + # Should return empty dict + assert isinstance(endpoints, dict) + assert len(endpoints) == 0 + + async def test_get_endpoints_returns_dict(self, mock_node: MockNode) -> None: + """Test that get_endpoints returns a dictionary.""" + endpoints = await mock_node.get_endpoints() + + assert isinstance(endpoints, dict) + for key, value in endpoints.items(): + assert isinstance(key, str) + assert isinstance(value, str) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionState: + """Tests for FSM state extraction.""" + + @pytest.fixture + def mock_node(self) -> MockNode: + """Create initialized mock node fixture. + + Returns: + Initialized MockNode instance. + """ + node = MockNode() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=None, + ) + return node + + async def test_get_current_state_returns_state(self, mock_node: MockNode) -> None: + """Test that get_current_state returns the node's state.""" + state = await mock_node.get_current_state() + + assert state == "idle" + + async def test_get_current_state_reflects_changes( + self, mock_node: MockNode + ) -> None: + """Test that get_current_state reflects state changes.""" + mock_node._state = "processing" + state = await mock_node.get_current_state() + + assert state == "processing" + + async def test_get_current_state_no_state_attribute(self) -> None: + """Test get_current_state when _state is missing.""" + node = MockNodeNoState() + node.initialize_introspection( + node_id="test-node-no-state", + node_type="EFFECT", + event_bus=None, + ) + + state = await node.get_current_state() + + assert state is None + + async def test_get_current_state_with_enum_state(self) -> None: + """Test get_current_state with enum-style state (has .value).""" + node = MockNodeWithEnumState() + node.initialize_introspection( + node_id="test-node-enum", + node_type="EFFECT", + event_bus=None, + ) + + state = await node.get_current_state() + assert state == "running" + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionCaching: + """Tests for caching behavior.""" + + @pytest.fixture + def mock_node(self) -> MockNode: + """Create initialized mock node with short TTL. + + Returns: + Initialized MockNode instance with 0.1s cache TTL. + """ + node = MockNode() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=None, + cache_ttl=0.1, # Short TTL for testing + ) + return node + + async def test_get_introspection_data_caches_result( + self, mock_node: MockNode + ) -> None: + """Test that get_introspection_data caches the result.""" + # First call - should compute + data1 = await mock_node.get_introspection_data() + timestamp1 = data1.timestamp + + # Immediate second call - should return cached + data2 = await mock_node.get_introspection_data() + timestamp2 = data2.timestamp + + # Same timestamp means cached result + assert timestamp1 == timestamp2 + + async def test_cache_expires_after_ttl(self, mock_node: MockNode) -> None: + """Test that cache expires after TTL.""" + # First call - populates cache + data1 = await mock_node.get_introspection_data() + timestamp1 = data1.timestamp + + # Wait for TTL to expire (0.1s + buffer) + await asyncio.sleep(0.15) + + # Next call should recompute + data2 = await mock_node.get_introspection_data() + timestamp2 = data2.timestamp + + # Different timestamp means cache was refreshed + assert timestamp2 > timestamp1 + + async def test_get_introspection_data_structure(self, mock_node: MockNode) -> None: + """Test that get_introspection_data returns expected model.""" + data = await mock_node.get_introspection_data() + + assert isinstance(data, ModelNodeIntrospectionEvent) + assert data.node_id == "test-node-001" + assert data.node_type == "EFFECT" + assert isinstance(data.capabilities, dict) + assert isinstance(data.endpoints, dict) + assert data.version == "1.0.0" + + async def test_cache_not_used_before_initialization(self) -> None: + """Test that cache starts empty.""" + node = MockNode() + node.initialize_introspection( + node_id="test-node-cache", + node_type="EFFECT", + event_bus=None, + ) + + assert node._introspection_cache is None + assert node._introspection_cached_at is None + + async def test_invalidate_introspection_cache(self, mock_node: MockNode) -> None: + """Test that invalidate_introspection_cache clears the cache.""" + # Populate cache + await mock_node.get_introspection_data() + assert mock_node._introspection_cache is not None + + # Invalidate + mock_node.invalidate_introspection_cache() + + assert mock_node._introspection_cache is None + assert mock_node._introspection_cached_at is None + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionPublishing: + """Tests for event bus publishing.""" + + @pytest.fixture + def mock_node_with_bus(self) -> MockNode: + """Create initialized mock node with event bus. + + Returns: + MockNode with MockEventBus attached. + """ + node = MockNode() + event_bus = MockEventBus() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=event_bus, + ) + return node + + @pytest.fixture + def mock_node_without_bus(self) -> MockNode: + """Create initialized mock node without event bus. + + Returns: + MockNode without event bus. + """ + node = MockNode() + node.initialize_introspection( + node_id="test-node-001", + node_type="EFFECT", + event_bus=None, + ) + return node + + async def test_publish_introspection_returns_false_without_event_bus( + self, mock_node_without_bus: MockNode + ) -> None: + """Test that publish_introspection returns False without event bus.""" + result = await mock_node_without_bus.publish_introspection() + + assert result is False + + async def test_publish_introspection_succeeds_with_event_bus( + self, mock_node_with_bus: MockNode + ) -> None: + """Test that publish_introspection succeeds with event bus.""" + result = await mock_node_with_bus.publish_introspection() + + assert result is True + + # Verify envelope was published + event_bus = mock_node_with_bus._introspection_event_bus + assert isinstance(event_bus, MockEventBus) + assert len(event_bus.published_envelopes) == 1 + + envelope, topic = event_bus.published_envelopes[0] + assert topic == "node.introspection" + assert isinstance(envelope, ModelNodeIntrospectionEvent) + + async def test_publish_introspection_with_correlation_id( + self, mock_node_with_bus: MockNode + ) -> None: + """Test that publish_introspection passes correlation_id.""" + correlation_id = uuid4() + await mock_node_with_bus.publish_introspection(correlation_id=correlation_id) + + event_bus = mock_node_with_bus._introspection_event_bus + assert isinstance(event_bus, MockEventBus) + + envelope, _ = event_bus.published_envelopes[0] + # Correlation ID should be set (it may be regenerated in the publish method) + assert envelope.correlation_id is not None + + async def test_publish_introspection_with_reason( + self, mock_node_with_bus: MockNode + ) -> None: + """Test that publish_introspection sets the reason.""" + await mock_node_with_bus.publish_introspection(reason="shutdown") + + event_bus = mock_node_with_bus._introspection_event_bus + assert isinstance(event_bus, MockEventBus) + + envelope, _ = event_bus.published_envelopes[0] + assert envelope.reason == "shutdown" + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionTasks: + """Tests for background task management.""" + + async def test_start_introspection_tasks_starts_heartbeat(self) -> None: + """Test that start_introspection_tasks creates heartbeat task.""" + node = MockNode() + event_bus = MockEventBus() + node.initialize_introspection( + node_id="test-node-tasks", + node_type="EFFECT", + event_bus=event_bus, + ) + + # Start tasks with fast heartbeat + await node.start_introspection_tasks( + enable_heartbeat=True, + heartbeat_interval_seconds=0.05, + enable_registry_listener=False, + ) + + try: + assert node._heartbeat_task is not None + assert not node._heartbeat_task.done() + + # Wait for at least one heartbeat + await asyncio.sleep(0.1) + + # Should have published at least one event + assert len(event_bus.published_envelopes) >= 1 + finally: + # Clean up + await node.stop_introspection_tasks() + + async def test_stop_introspection_tasks_cancels_tasks(self) -> None: + """Test that stop_introspection_tasks cancels all tasks.""" + node = MockNode() + event_bus = MockEventBus() + node.initialize_introspection( + node_id="test-node-stop", + node_type="EFFECT", + event_bus=event_bus, + ) + + # Start and then stop tasks + await node.start_introspection_tasks( + enable_heartbeat=True, + heartbeat_interval_seconds=0.1, + enable_registry_listener=False, + ) + assert node._heartbeat_task is not None + + await node.stop_introspection_tasks() + + # Task should be None after stop + assert node._heartbeat_task is None + + async def test_stop_introspection_tasks_idempotent(self) -> None: + """Test that stop_introspection_tasks can be called multiple times.""" + node = MockNode() + node.initialize_introspection( + node_id="test-node-idempotent", + node_type="EFFECT", + event_bus=None, + ) + + # Stop without starting should be safe + await node.stop_introspection_tasks() + await node.stop_introspection_tasks() + + assert node._heartbeat_task is None + + async def test_heartbeat_publishes_periodically(self) -> None: + """Test that heartbeat publishes at regular intervals.""" + node = MockNode() + event_bus = MockEventBus() + node.initialize_introspection( + node_id="test-node-heartbeat", + node_type="EFFECT", + event_bus=event_bus, + ) + + await node.start_introspection_tasks( + enable_heartbeat=True, + heartbeat_interval_seconds=0.05, + enable_registry_listener=False, + ) + + try: + # Wait for multiple heartbeats + await asyncio.sleep(0.2) + + # Should have multiple events (at least 3) + assert len(event_bus.published_envelopes) >= 3 + finally: + await node.stop_introspection_tasks() + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionGracefulDegradation: + """Tests for graceful degradation on errors.""" + + async def test_publish_graceful_degradation_on_error(self) -> None: + """Test that publish_introspection handles errors gracefully.""" + node = MockNode() + failing_event_bus = MockEventBus(should_fail=True) + node.initialize_introspection( + node_id="test-node-graceful", + node_type="EFFECT", + event_bus=failing_event_bus, + ) + + # Should not raise, just return False + result = await node.publish_introspection() + + assert result is False + + async def test_publish_does_not_crash_on_exception(self) -> None: + """Test that publish_introspection catches all exceptions.""" + node = MockNode() + + # Create event bus that raises unexpected exception + class BrokenEventBus: + async def publish_envelope( + self, envelope: ModelNodeIntrospectionEvent, topic: str + ) -> None: + raise ValueError("Unexpected error") + + node.initialize_introspection( + node_id="test-node-broken", + node_type="EFFECT", + event_bus=BrokenEventBus(), + ) + + # Should not raise + result = await node.publish_introspection() + assert result is False + + async def test_heartbeat_continues_after_publish_failure(self) -> None: + """Test that heartbeat continues even if publish fails.""" + node = MockNode() + event_bus = MockEventBus(should_fail=True) + node.initialize_introspection( + node_id="test-node-continue", + node_type="EFFECT", + event_bus=event_bus, + ) + + await node.start_introspection_tasks( + enable_heartbeat=True, + heartbeat_interval_seconds=0.05, + enable_registry_listener=False, + ) + + try: + # Let heartbeat run with failing publishes + await asyncio.sleep(0.15) + + # Task should still be running (not crashed) + assert node._heartbeat_task is not None + assert not node._heartbeat_task.done() + finally: + await node.stop_introspection_tasks() + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionPerformance: + """Tests for performance requirements. + + Note: Performance thresholds are multiplied by PERF_MULTIPLIER to account + for CI environments which may be slower than local development machines. + """ + + @pytest.fixture + def mock_node(self) -> MockNode: + """Create initialized mock node fixture. + + Returns: + Initialized MockNode instance. + """ + node = MockNode() + node.initialize_introspection( + node_id="test-node-perf", + node_type="EFFECT", + event_bus=None, + ) + return node + + async def test_introspection_extraction_under_50ms( + self, mock_node: MockNode + ) -> None: + """Test that introspection data extraction completes within threshold.""" + # Clear cache to force full computation + mock_node._introspection_cache = None + mock_node._introspection_cached_at = None + + threshold_ms = 50 * PERF_MULTIPLIER + start = time.time() + await mock_node.get_introspection_data() + elapsed_ms = (time.time() - start) * 1000 + + assert elapsed_ms < threshold_ms, ( + f"Introspection took {elapsed_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_cached_introspection_under_1ms(self, mock_node: MockNode) -> None: + """Test that cached introspection returns within threshold.""" + # Populate cache + await mock_node.get_introspection_data() + + threshold_ms = 1 * PERF_MULTIPLIER + start = time.time() + await mock_node.get_introspection_data() + elapsed_ms = (time.time() - start) * 1000 + + assert elapsed_ms < threshold_ms, ( + f"Cached introspection took {elapsed_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_capability_extraction_under_10ms(self, mock_node: MockNode) -> None: + """Test that capability extraction completes within threshold.""" + threshold_ms = 10 * PERF_MULTIPLIER + start = time.time() + await mock_node.get_capabilities() + elapsed_ms = (time.time() - start) * 1000 + + assert elapsed_ms < threshold_ms, ( + f"Capability extraction took {elapsed_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_endpoint_discovery_under_10ms(self, mock_node: MockNode) -> None: + """Test that endpoint discovery completes within threshold.""" + threshold_ms = 10 * PERF_MULTIPLIER + start = time.time() + await mock_node.get_endpoints() + elapsed_ms = (time.time() - start) * 1000 + + assert elapsed_ms < threshold_ms, ( + f"Endpoint discovery took {elapsed_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_state_extraction_under_1ms(self, mock_node: MockNode) -> None: + """Test that state extraction completes within threshold.""" + threshold_ms = 1 * PERF_MULTIPLIER + start = time.time() + await mock_node.get_current_state() + elapsed_ms = (time.time() - start) * 1000 + + assert elapsed_ms < threshold_ms, ( + f"State extraction took {elapsed_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_multiple_introspection_calls_consistent_performance( + self, mock_node: MockNode + ) -> None: + """Test that multiple introspection calls have consistent performance.""" + times: list[float] = [] + + for _ in range(10): + # Clear cache each time + mock_node._introspection_cache = None + mock_node._introspection_cached_at = None + + start = time.time() + await mock_node.get_introspection_data() + elapsed_ms = (time.time() - start) * 1000 + times.append(elapsed_ms) + + avg_time = sum(times) / len(times) + max_time = max(times) + + avg_threshold_ms = 30 * PERF_MULTIPLIER + max_threshold_ms = 50 * PERF_MULTIPLIER + + assert avg_time < avg_threshold_ms, ( + f"Average time {avg_time:.2f}ms, expected <{avg_threshold_ms:.0f}ms" + ) + assert max_time < max_threshold_ms, ( + f"Max time {max_time:.2f}ms, expected <{max_threshold_ms:.0f}ms" + ) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionBenchmark: + """Detailed performance benchmarks with instrumentation. + + These tests verify the <50ms requirement and provide + detailed timing breakdowns for optimization. + + Note: Performance thresholds are multiplied by PERF_MULTIPLIER to account + for CI environments which may be slower than local development machines. + """ + + async def test_introspection_benchmark_with_instrumentation(self) -> None: + """Benchmark introspection with detailed timing breakdown.""" + node = MockNode() + node.initialize_introspection( + node_id="benchmark-node", + node_type="EFFECT", + event_bus=None, + ) + + # Clear cache for full computation + node._introspection_cache = None + node._introspection_cached_at = None + + timings: dict[str, list[float]] = { + "get_capabilities": [], + "get_endpoints": [], + "get_current_state": [], + "total_introspection": [], + } + + iterations = 20 + for _ in range(iterations): + node._introspection_cache = None + node._introspection_cached_at = None + + # Time individual components + start = time.perf_counter() + await node.get_capabilities() + timings["get_capabilities"].append((time.perf_counter() - start) * 1000) + + start = time.perf_counter() + await node.get_endpoints() + timings["get_endpoints"].append((time.perf_counter() - start) * 1000) + + start = time.perf_counter() + await node.get_current_state() + timings["get_current_state"].append((time.perf_counter() - start) * 1000) + + node._introspection_cache = None + node._introspection_cached_at = None + + start = time.perf_counter() + await node.get_introspection_data() + timings["total_introspection"].append((time.perf_counter() - start) * 1000) + + # Calculate statistics + for name, times in timings.items(): + avg = sum(times) / len(times) + min_t = min(times) + max_t = max(times) + p95 = sorted(times)[int(len(times) * 0.95)] + + # Log timing breakdown for debugging + print( + f"{name}: avg={avg:.2f}ms, min={min_t:.2f}ms, " + f"max={max_t:.2f}ms, p95={p95:.2f}ms" + ) + + # Assert <50ms requirement (with CI buffer) + threshold_ms = 50 * PERF_MULTIPLIER + avg_total = sum(timings["total_introspection"]) / len( + timings["total_introspection"] + ) + assert avg_total < threshold_ms, ( + f"Average introspection {avg_total:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + + async def test_introspection_concurrent_load_benchmark(self) -> None: + """Benchmark introspection under concurrent load.""" + node = MockNode() + node.initialize_introspection( + node_id="concurrent-benchmark-node", + node_type="EFFECT", + event_bus=None, + cache_ttl=0.001, # Force cache misses + ) + + async def single_introspection() -> float: + start = time.perf_counter() + await node.get_introspection_data() + return (time.perf_counter() - start) * 1000 + + # 50 concurrent introspection requests + tasks = [single_introspection() for _ in range(50)] + times = await asyncio.gather(*tasks) + + avg_time = sum(times) / len(times) + max_time = max(times) + p95_time = sorted(times)[int(len(times) * 0.95)] + + # Log benchmark results + print( + f"Concurrent load (50 requests): avg={avg_time:.2f}ms, " + f"max={max_time:.2f}ms, p95={p95_time:.2f}ms" + ) + + threshold_ms = 100 * PERF_MULTIPLIER # Higher threshold for concurrent load + assert avg_time < threshold_ms, ( + f"Average concurrent time {avg_time:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + assert max_time < threshold_ms * 2, ( + f"Max concurrent time {max_time:.2f}ms exceeds {threshold_ms * 2:.0f}ms" + ) + + async def test_cache_hit_performance(self) -> None: + """Verify cache hits are sub-millisecond.""" + node = MockNode() + node.initialize_introspection( + node_id="cache-hit-node", + node_type="EFFECT", + event_bus=None, + ) + + # Warm cache + await node.get_introspection_data() + + # Measure cache hits + times: list[float] = [] + for _ in range(100): + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + avg_time = sum(times) / len(times) + p99 = sorted(times)[int(len(times) * 0.99)] + min_time = min(times) + max_time = max(times) + + # Log cache hit performance + print( + f"Cache hits (100 requests): avg={avg_time:.3f}ms, " + f"min={min_time:.3f}ms, max={max_time:.3f}ms, p99={p99:.3f}ms" + ) + + # Cache hits should be very fast + threshold_ms = 0.5 * PERF_MULTIPLIER + assert avg_time < threshold_ms, ( + f"Cache hit avg {avg_time:.3f}ms exceeds {threshold_ms:.1f}ms" + ) + + async def test_introspection_p95_latency(self) -> None: + """Test that p95 latency meets requirements.""" + node = MockNode() + node.initialize_introspection( + node_id="p95-node", + node_type="EFFECT", + event_bus=None, + ) + + times: list[float] = [] + iterations = 50 + + for _ in range(iterations): + # Clear cache for each iteration + node._introspection_cache = None + node._introspection_cached_at = None + + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + p95 = sorted(times)[int(len(times) * 0.95)] + p99 = sorted(times)[int(len(times) * 0.99)] + avg_time = sum(times) / len(times) + + # Log p95 and p99 latencies + print( + f"Latency distribution ({iterations} iterations): " + f"avg={avg_time:.2f}ms, p95={p95:.2f}ms, p99={p99:.2f}ms" + ) + + # p95 should be under 50ms threshold (with CI buffer) + threshold_ms = 50 * PERF_MULTIPLIER + assert p95 < threshold_ms, ( + f"p95 latency {p95:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + + async def test_component_timing_breakdown(self) -> None: + """Test timing breakdown of individual introspection components.""" + node = MockNode() + node.initialize_introspection( + node_id="breakdown-node", + node_type="EFFECT", + event_bus=None, + ) + + # Time each component individually + components = { + "capabilities": node.get_capabilities, + "endpoints": node.get_endpoints, + "state": node.get_current_state, + } + + component_times: dict[str, float] = {} + + for name, func in components.items(): + times: list[float] = [] + for _ in range(10): + start = time.perf_counter() + await func() + times.append((time.perf_counter() - start) * 1000) + component_times[name] = sum(times) / len(times) + + # Log component breakdown + print("Component timing breakdown:") + for name, avg_ms in component_times.items(): + print(f" {name}: {avg_ms:.2f}ms") + + # Capabilities is typically the slowest (reflection-based) + cap_threshold_ms = 20 * PERF_MULTIPLIER + assert component_times["capabilities"] < cap_threshold_ms, ( + f"Capabilities extraction {component_times['capabilities']:.2f}ms " + f"exceeds {cap_threshold_ms:.0f}ms" + ) + + # State extraction should be very fast + state_threshold_ms = 1 * PERF_MULTIPLIER + assert component_times["state"] < state_threshold_ms, ( + f"State extraction {component_times['state']:.2f}ms " + f"exceeds {state_threshold_ms:.0f}ms" + ) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionEdgeCases: + """Tests for edge cases and boundary conditions.""" + + async def test_empty_node_introspection(self) -> None: + """Test introspection on a minimal node.""" + + class MinimalNode(MixinNodeIntrospection): + pass + + node = MinimalNode() + node.initialize_introspection( + node_id="minimal-node", + node_type="EFFECT", + event_bus=None, + ) + + data = await node.get_introspection_data() + + assert data.node_id == "minimal-node" + assert data.current_state is None # No state attribute + + async def test_large_capability_list(self) -> None: + """Test introspection with many public methods.""" + + class LargeNode(MixinNodeIntrospection): + async def execute_task_001(self) -> None: + pass + + async def handle_event_002(self) -> None: + pass + + async def process_data_003(self) -> None: + pass + + async def run_operation_004(self) -> None: + pass + + async def invoke_action_005(self) -> None: + pass + + async def call_service_006(self) -> None: + pass + + async def execute_job_007(self) -> None: + pass + + async def handle_request_008(self) -> None: + pass + + async def process_queue_009(self) -> None: + pass + + async def run_batch_010(self) -> None: + pass + + node = LargeNode() + node.initialize_introspection( + node_id="large-node", + node_type="COMPUTE", + event_bus=None, + ) + + threshold_ms = 50 * PERF_MULTIPLIER + start = time.time() + capabilities = await node.get_capabilities() + elapsed_ms = (time.time() - start) * 1000 + + # Should include all 10 operation methods + operations = capabilities["operations"] + assert isinstance(operations, list) + assert len(operations) >= 10 + assert elapsed_ms < threshold_ms, ( + f"Large capability extraction took {elapsed_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_concurrent_introspection_calls(self) -> None: + """Test concurrent introspection data requests.""" + node = MockNode() + node.initialize_introspection( + node_id="concurrent-node", + node_type="EFFECT", + event_bus=None, + cache_ttl=0.001, # Very short TTL + ) + + # Make 100 concurrent calls + tasks = [node.get_introspection_data() for _ in range(100)] + results = await asyncio.gather(*tasks) + + # All should succeed + assert len(results) == 100 + for result in results: + assert result.node_id == "concurrent-node" + + async def test_introspection_with_special_characters_in_state(self) -> None: + """Test introspection with special characters in state.""" + node = MockNode() + node._state = "statespecial&chars\"quote'" + node.initialize_introspection( + node_id="special-node", + node_type="EFFECT", + event_bus=None, + ) + + state = await node.get_current_state() + assert state == "statespecial&chars\"quote'" + + async def test_introspection_preserves_node_functionality(self) -> None: + """Test that introspection mixin doesn't affect node functionality.""" + node = MockNode() + node.initialize_introspection( + node_id="functional-node", + node_type="EFFECT", + event_bus=None, + ) + + # Node methods should still work normally + result = await node.execute("test_op", {"data": "value"}) + assert result["result"] == "ok" + assert result["operation"] == "test_op" + + health = await node.health_check() + assert health["healthy"] is True + + +@pytest.mark.asyncio(loop_scope="function") +class TestMixinNodeIntrospectionClassLevelCache: + """Test class-level method signature caching for performance optimization.""" + + def setup_method(self) -> None: + """Clear class-level cache before each test.""" + MixinNodeIntrospection._invalidate_class_method_cache() + + def teardown_method(self) -> None: + """Clear class-level cache after each test.""" + MixinNodeIntrospection._invalidate_class_method_cache() + + async def test_class_method_cache_populated_on_first_access(self) -> None: + """Test that class-level cache is populated on first access.""" + node = MockNode() + node.initialize_introspection( + node_id="cache-test-node", + node_type="EFFECT", + event_bus=None, + ) + + # Cache should be empty before first access + assert MockNode not in MixinNodeIntrospection._class_method_cache + + # Access capabilities to trigger cache population + await node.get_capabilities() + + # Cache should now contain MockNode + assert MockNode in MixinNodeIntrospection._class_method_cache + cached_signatures = MixinNodeIntrospection._class_method_cache[MockNode] + assert isinstance(cached_signatures, dict) + assert len(cached_signatures) > 0 + + async def test_class_method_cache_shared_across_instances(self) -> None: + """Test that class-level cache is shared across instances.""" + node1 = MockNode() + node1.initialize_introspection( + node_id="cache-node-1", + node_type="EFFECT", + event_bus=None, + ) + + node2 = MockNode() + node2.initialize_introspection( + node_id="cache-node-2", + node_type="EFFECT", + event_bus=None, + ) + + # First node populates cache + await node1.get_capabilities() + assert MockNode in MixinNodeIntrospection._class_method_cache + + # Second node uses same cache (no re-population) + cached_before = id(MixinNodeIntrospection._class_method_cache[MockNode]) + await node2.get_capabilities() + cached_after = id(MixinNodeIntrospection._class_method_cache[MockNode]) + + # Cache object identity should be the same (not recreated) + assert cached_before == cached_after + + async def test_invalidate_class_method_cache_specific_class(self) -> None: + """Test invalidating cache for a specific class.""" + node = MockNode() + node.initialize_introspection( + node_id="invalidate-test", + node_type="EFFECT", + event_bus=None, + ) + + # Populate cache + await node.get_capabilities() + assert MockNode in MixinNodeIntrospection._class_method_cache + + # Invalidate specific class + MixinNodeIntrospection._invalidate_class_method_cache(MockNode) + + # Cache should be cleared for MockNode + assert MockNode not in MixinNodeIntrospection._class_method_cache + + async def test_invalidate_class_method_cache_all_classes(self) -> None: + """Test invalidating cache for all classes.""" + node = MockNode() + node.initialize_introspection( + node_id="invalidate-all-test", + node_type="EFFECT", + event_bus=None, + ) + + # Populate cache + await node.get_capabilities() + assert MockNode in MixinNodeIntrospection._class_method_cache + + # Invalidate all + MixinNodeIntrospection._invalidate_class_method_cache() + + # Cache should be empty + assert len(MixinNodeIntrospection._class_method_cache) == 0 + + async def test_cached_signatures_match_direct_extraction(self) -> None: + """Test that cached signatures match direct signature extraction.""" + node = MockNode() + node.initialize_introspection( + node_id="signature-match-test", + node_type="EFFECT", + event_bus=None, + ) + + # Get capabilities (uses cache) + capabilities = await node.get_capabilities() + cached_signatures = capabilities["method_signatures"] + assert isinstance(cached_signatures, dict) + + # Get cached signatures directly + direct_cached = MixinNodeIntrospection._class_method_cache.get(MockNode, {}) + + # The capabilities method filters some prefixes, but the direct cache + # should have all public methods. Verify cached signatures are used. + assert len(cached_signatures) > 0 + assert len(direct_cached) >= len(cached_signatures) + + async def test_class_level_cache_performance_benefit(self) -> None: + """Test that class-level caching provides performance benefit.""" + node = MockNode() + node.initialize_introspection( + node_id="perf-cache-test", + node_type="EFFECT", + event_bus=None, + ) + + # First call (cold cache) - populates cache + start1 = time.time() + await node.get_capabilities() + first_call_ms = (time.time() - start1) * 1000 + + # Subsequent calls (warm cache) - uses cached signatures + times_warm = [] + for _ in range(10): + start = time.time() + await node.get_capabilities() + times_warm.append((time.time() - start) * 1000) + + avg_warm_ms = sum(times_warm) / len(times_warm) + + # Warm cache calls should be reasonably fast + threshold_ms = 5 * PERF_MULTIPLIER + assert avg_warm_ms < threshold_ms, ( + f"Warm cache calls averaged {avg_warm_ms:.2f}ms, expected <{threshold_ms:.0f}ms" + ) + + async def test_different_classes_have_separate_cache_entries(self) -> None: + """Test that different classes have separate cache entries.""" + + class CustomNode1(MixinNodeIntrospection): + async def execute_custom1(self, data: str) -> dict[str, str]: + return {"custom1": data} + + class CustomNode2(MixinNodeIntrospection): + async def execute_custom2(self, value: int) -> dict[str, int]: + return {"custom2": value} + + node1 = CustomNode1() + node1.initialize_introspection( + node_id="custom-1", + node_type="COMPUTE", + event_bus=None, + ) + + node2 = CustomNode2() + node2.initialize_introspection( + node_id="custom-2", + node_type="COMPUTE", + event_bus=None, + ) + + # Both populate their respective caches + await node1.get_capabilities() + await node2.get_capabilities() + + # Both classes should have entries + assert CustomNode1 in MixinNodeIntrospection._class_method_cache + assert CustomNode2 in MixinNodeIntrospection._class_method_cache + + # Entries should be different + cache1 = MixinNodeIntrospection._class_method_cache[CustomNode1] + cache2 = MixinNodeIntrospection._class_method_cache[CustomNode2] + + # CustomNode1 should have execute_custom1 + assert "execute_custom1" in cache1 + # CustomNode2 should have execute_custom2 + assert "execute_custom2" in cache2 + + # Each should NOT have the other's method + assert "execute_custom2" not in cache1 + assert "execute_custom1" not in cache2 + + async def test_cache_handles_methods_without_signatures(self) -> None: + """Test that cache handles methods without inspectable signatures.""" + + class NodeWithBuiltins(MixinNodeIntrospection): + # Built-in methods that may not have inspectable signatures + pass + + node = NodeWithBuiltins() + node.initialize_introspection( + node_id="builtins-test", + node_type="EFFECT", + event_bus=None, + ) + + # Should not raise exception + capabilities = await node.get_capabilities() + assert isinstance(capabilities, dict) + assert "method_signatures" in capabilities + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionConfigurableKeywords: + """Tests for configurable operation_keywords and exclude_prefixes.""" + + async def test_default_operation_keywords_used_when_not_specified(self) -> None: + """Test that DEFAULT_OPERATION_KEYWORDS is used when not specified.""" + node = MockNode() + node.initialize_introspection( + node_id="default-keywords-node", + node_type="EFFECT", + event_bus=None, + ) + assert ( + node._introspection_operation_keywords + == MixinNodeIntrospection.DEFAULT_OPERATION_KEYWORDS + ) + + async def test_default_exclude_prefixes_used_when_not_specified(self) -> None: + """Test that DEFAULT_EXCLUDE_PREFIXES is used when not specified.""" + node = MockNode() + node.initialize_introspection( + node_id="default-prefixes-node", + node_type="EFFECT", + event_bus=None, + ) + assert ( + node._introspection_exclude_prefixes + == MixinNodeIntrospection.DEFAULT_EXCLUDE_PREFIXES + ) + + async def test_custom_operation_keywords_are_stored(self) -> None: + """Test that custom operation_keywords are stored correctly.""" + custom_keywords = {"fetch", "upload", "download", "sync"} + node = MockNode() + node.initialize_introspection( + node_id="custom-keywords-node", + node_type="EFFECT", + event_bus=None, + operation_keywords=custom_keywords, + ) + assert node._introspection_operation_keywords == custom_keywords + + async def test_custom_exclude_prefixes_are_stored(self) -> None: + """Test that custom exclude_prefixes are stored correctly.""" + custom_prefixes = {"_", "helper_", "internal_"} + node = MockNode() + node.initialize_introspection( + node_id="custom-prefixes-node", + node_type="EFFECT", + event_bus=None, + exclude_prefixes=custom_prefixes, + ) + assert node._introspection_exclude_prefixes == custom_prefixes + + async def test_custom_operation_keywords_affect_capability_discovery(self) -> None: + """Test that custom operation_keywords affect which methods are discovered.""" + + class CustomMethodsNode(MixinNodeIntrospection): + async def fetch_data(self, source: str) -> dict[str, str]: + return {"source": source} + + async def upload_file(self, file_path: str) -> bool: + return True + + async def execute_task(self, task_id: str) -> None: + pass + + node = CustomMethodsNode() + node.initialize_introspection( + node_id="custom-ops-node", + node_type="EFFECT", + event_bus=None, + operation_keywords={"fetch", "upload"}, + ) + capabilities = await node.get_capabilities() + operations = capabilities["operations"] + assert isinstance(operations, list) + assert "fetch_data" in operations + assert "upload_file" in operations + assert "execute_task" not in operations + + async def test_node_type_specific_keywords_constant_exists(self) -> None: + """Test that NODE_TYPE_OPERATION_KEYWORDS constant exists.""" + assert hasattr(MixinNodeIntrospection, "NODE_TYPE_OPERATION_KEYWORDS") + keywords_map = MixinNodeIntrospection.NODE_TYPE_OPERATION_KEYWORDS + assert "EFFECT" in keywords_map + assert "COMPUTE" in keywords_map + assert "REDUCER" in keywords_map + assert "ORCHESTRATOR" in keywords_map + for keywords in keywords_map.values(): + assert isinstance(keywords, set) + + async def test_empty_operation_keywords_discovers_no_operations(self) -> None: + """Test that empty operation_keywords results in no operations discovered.""" + node = MockNode() + node.initialize_introspection( + node_id="empty-keywords-node", + node_type="EFFECT", + event_bus=None, + operation_keywords=set(), + ) + capabilities = await node.get_capabilities() + operations = capabilities["operations"] + assert isinstance(operations, list) + assert len(operations) == 0 + + async def test_configuration_is_instance_specific(self) -> None: + """Test that configuration is instance-specific, not shared.""" + + class MultiInstanceNode(MixinNodeIntrospection): + async def execute_task(self) -> None: + pass + + async def fetch_data(self) -> None: + pass + + node1 = MultiInstanceNode() + node1.initialize_introspection( + node_id="node-1", + node_type="EFFECT", + event_bus=None, + operation_keywords={"execute"}, + ) + node2 = MultiInstanceNode() + node2.initialize_introspection( + node_id="node-2", + node_type="EFFECT", + event_bus=None, + operation_keywords={"fetch"}, + ) + caps1 = await node1.get_capabilities() + caps2 = await node2.get_capabilities() + ops1 = caps1["operations"] + ops2 = caps2["operations"] + assert isinstance(ops1, list) + assert isinstance(ops2, list) + assert "execute_task" in ops1 + assert "fetch_data" not in ops1 + assert "fetch_data" in ops2 + assert "execute_task" not in ops2 + + async def test_default_keywords_not_mutated(self) -> None: + """Test that DEFAULT_OPERATION_KEYWORDS is not mutated by instances.""" + original_defaults = MixinNodeIntrospection.DEFAULT_OPERATION_KEYWORDS.copy() + node = MockNode() + node.initialize_introspection( + node_id="no-mutation-node", + node_type="EFFECT", + event_bus=None, + ) + node._introspection_operation_keywords.add("custom_keyword") + assert original_defaults == MixinNodeIntrospection.DEFAULT_OPERATION_KEYWORDS + assert "custom_keyword" not in MixinNodeIntrospection.DEFAULT_OPERATION_KEYWORDS + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionPerformanceMetrics: + """Tests for performance metrics tracking and retrieval.""" + + async def test_get_performance_metrics_returns_none_before_introspection( + self, + ) -> None: + """Test that get_performance_metrics returns None before introspection.""" + node = MockNode() + node.initialize_introspection( + node_id="metrics-test-node", + node_type="EFFECT", + event_bus=None, + ) + + metrics = node.get_performance_metrics() + assert metrics is None + + async def test_get_performance_metrics_returns_metrics_after_introspection( + self, + ) -> None: + """Test that get_performance_metrics returns metrics after introspection.""" + node = MockNode() + node.initialize_introspection( + node_id="metrics-test-node", + node_type="EFFECT", + event_bus=None, + ) + + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + assert isinstance(metrics, IntrospectionPerformanceMetrics) + + async def test_performance_metrics_contains_expected_fields(self) -> None: + """Test that performance metrics contain all expected fields.""" + node = MockNode() + node.initialize_introspection( + node_id="metrics-fields-node", + node_type="EFFECT", + event_bus=None, + ) + + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + # Check all fields exist + assert hasattr(metrics, "get_capabilities_ms") + assert hasattr(metrics, "discover_capabilities_ms") + assert hasattr(metrics, "get_endpoints_ms") + assert hasattr(metrics, "get_current_state_ms") + assert hasattr(metrics, "total_introspection_ms") + assert hasattr(metrics, "cache_hit") + assert hasattr(metrics, "method_count") + assert hasattr(metrics, "threshold_exceeded") + assert hasattr(metrics, "slow_operations") + + # Check types + assert isinstance(metrics.get_capabilities_ms, float) + assert isinstance(metrics.total_introspection_ms, float) + assert isinstance(metrics.cache_hit, bool) + assert isinstance(metrics.method_count, int) + assert isinstance(metrics.threshold_exceeded, bool) + assert isinstance(metrics.slow_operations, list) + + async def test_performance_metrics_cache_hit_detection(self) -> None: + """Test that cache hits are correctly detected in metrics.""" + node = MockNode() + node.initialize_introspection( + node_id="cache-hit-metrics-node", + node_type="EFFECT", + event_bus=None, + ) + + # First call - cache miss + await node.get_introspection_data() + metrics_miss = node.get_performance_metrics() + assert metrics_miss is not None + assert metrics_miss.cache_hit is False + + # Second call - cache hit + await node.get_introspection_data() + metrics_hit = node.get_performance_metrics() + assert metrics_hit is not None + assert metrics_hit.cache_hit is True + + async def test_performance_metrics_method_count(self) -> None: + """Test that method count is correctly reported in metrics.""" + node = MockNode() + node.initialize_introspection( + node_id="method-count-metrics-node", + node_type="EFFECT", + event_bus=None, + ) + + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + # MockNode has at least 4 public methods (execute, health_check, + # handle_event, process_batch) + assert metrics.method_count >= 4 + + async def test_performance_metrics_to_dict(self) -> None: + """Test that to_dict() returns all fields.""" + node = MockNode() + node.initialize_introspection( + node_id="to-dict-node", + node_type="EFFECT", + event_bus=None, + ) + + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + metrics_dict = metrics.to_dict() + + assert isinstance(metrics_dict, dict) + assert "get_capabilities_ms" in metrics_dict + assert "discover_capabilities_ms" in metrics_dict + assert "get_endpoints_ms" in metrics_dict + assert "get_current_state_ms" in metrics_dict + assert "total_introspection_ms" in metrics_dict + assert "cache_hit" in metrics_dict + assert "method_count" in metrics_dict + assert "threshold_exceeded" in metrics_dict + assert "slow_operations" in metrics_dict + + async def test_performance_metrics_fresh_on_each_call(self) -> None: + """Test that performance metrics are fresh for each introspection call.""" + node = MockNode() + node.initialize_introspection( + node_id="fresh-metrics-node", + node_type="EFFECT", + event_bus=None, + cache_ttl=0.001, # Very short TTL to force cache refresh + ) + + # First call + await node.get_introspection_data() + metrics1 = node.get_performance_metrics() + assert metrics1 is not None + total_ms_1 = metrics1.total_introspection_ms + + # Wait for cache to expire + await asyncio.sleep(0.01) + + # Second call with fresh computation + await node.get_introspection_data() + metrics2 = node.get_performance_metrics() + assert metrics2 is not None + total_ms_2 = metrics2.total_introspection_ms + + # Metrics should be different (fresh computation) + # We can't guarantee exact timing, but both should be positive + assert total_ms_1 > 0 + assert total_ms_2 > 0 + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionMethodCountBenchmark: + """Performance benchmarks with varying method counts. + + These tests validate that introspection performance scales appropriately + with the number of methods on a node. The <50ms target should be maintained + even with a large number of methods due to class-level caching. + """ + + async def test_benchmark_minimal_methods_node(self) -> None: + """Benchmark introspection on a node with minimal methods.""" + + class MinimalMethodsNode(MixinNodeIntrospection): + """Node with just one operation method.""" + + async def execute(self, data: str) -> str: + return data + + node = MinimalMethodsNode() + node.initialize_introspection( + node_id="minimal-methods-node", + node_type="COMPUTE", + event_bus=None, + ) + + # Clear cache for accurate measurement + MixinNodeIntrospection._invalidate_class_method_cache(MinimalMethodsNode) + + times: list[float] = [] + for _ in range(20): + node._introspection_cache = None + node._introspection_cached_at = None + MixinNodeIntrospection._invalidate_class_method_cache(MinimalMethodsNode) + + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + avg_time = sum(times) / len(times) + max_time = max(times) + metrics = node.get_performance_metrics() + + print( + f"\nMinimal methods node ({metrics.method_count if metrics else 0} methods):" + ) + print(f" avg={avg_time:.2f}ms, max={max_time:.2f}ms") + + # Should be well under the threshold + threshold_ms = PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS * PERF_MULTIPLIER + assert avg_time < threshold_ms, ( + f"Minimal methods avg {avg_time:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + + async def test_benchmark_medium_methods_node(self) -> None: + """Benchmark introspection on a node with ~20 methods.""" + + class MediumMethodsNode(MixinNodeIntrospection): + """Node with ~20 operation methods.""" + + async def execute_task_01(self, d: str) -> str: + return d + + async def execute_task_02(self, d: str) -> str: + return d + + async def execute_task_03(self, d: str) -> str: + return d + + async def handle_event_01(self, d: str) -> str: + return d + + async def handle_event_02(self, d: str) -> str: + return d + + async def handle_event_03(self, d: str) -> str: + return d + + async def process_data_01(self, d: str) -> str: + return d + + async def process_data_02(self, d: str) -> str: + return d + + async def process_data_03(self, d: str) -> str: + return d + + async def run_operation_01(self, d: str) -> str: + return d + + async def run_operation_02(self, d: str) -> str: + return d + + async def run_operation_03(self, d: str) -> str: + return d + + async def invoke_action_01(self, d: str) -> str: + return d + + async def invoke_action_02(self, d: str) -> str: + return d + + async def invoke_action_03(self, d: str) -> str: + return d + + async def call_service_01(self, d: str) -> str: + return d + + async def call_service_02(self, d: str) -> str: + return d + + async def call_service_03(self, d: str) -> str: + return d + + # Additional utility methods + def validate_input(self, d: str) -> bool: + return True + + def transform_output(self, d: str) -> str: + return d + + node = MediumMethodsNode() + node.initialize_introspection( + node_id="medium-methods-node", + node_type="COMPUTE", + event_bus=None, + ) + + # Clear cache for accurate measurement + MixinNodeIntrospection._invalidate_class_method_cache(MediumMethodsNode) + + times: list[float] = [] + for _ in range(20): + node._introspection_cache = None + node._introspection_cached_at = None + MixinNodeIntrospection._invalidate_class_method_cache(MediumMethodsNode) + + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + avg_time = sum(times) / len(times) + max_time = max(times) + metrics = node.get_performance_metrics() + + print( + f"\nMedium methods node ({metrics.method_count if metrics else 0} methods):" + ) + print(f" avg={avg_time:.2f}ms, max={max_time:.2f}ms") + + # Should still be under the threshold + threshold_ms = PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS * PERF_MULTIPLIER + assert avg_time < threshold_ms, ( + f"Medium methods avg {avg_time:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + + async def test_benchmark_large_methods_node(self) -> None: + """Benchmark introspection on a node with ~50 methods.""" + + # Create a node class dynamically with many methods + class LargeMethodsNode(MixinNodeIntrospection): + """Node with ~50 methods to stress-test reflection performance.""" + + # Add 50 methods dynamically + for i in range(50): + # Alternate between different operation keywords + keywords = ["execute", "handle", "process", "run", "invoke"] + keyword = keywords[i % len(keywords)] + + async def method(self: LargeMethodsNode, data: str = "") -> str: + return data + + method.__name__ = f"{keyword}_operation_{i:02d}" + setattr(LargeMethodsNode, method.__name__, method) + + node = LargeMethodsNode() + node.initialize_introspection( + node_id="large-methods-node", + node_type="COMPUTE", + event_bus=None, + ) + + # Clear cache for accurate measurement + MixinNodeIntrospection._invalidate_class_method_cache(LargeMethodsNode) + + times: list[float] = [] + for _ in range(20): + node._introspection_cache = None + node._introspection_cached_at = None + MixinNodeIntrospection._invalidate_class_method_cache(LargeMethodsNode) + + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + avg_time = sum(times) / len(times) + max_time = max(times) + p95_time = sorted(times)[int(len(times) * 0.95)] + metrics = node.get_performance_metrics() + + print( + f"\nLarge methods node ({metrics.method_count if metrics else 0} methods):" + ) + print(f" avg={avg_time:.2f}ms, max={max_time:.2f}ms, p95={p95_time:.2f}ms") + + # Should still be under the threshold even with 50+ methods + threshold_ms = PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS * PERF_MULTIPLIER + assert avg_time < threshold_ms, ( + f"Large methods avg {avg_time:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + + async def test_benchmark_cache_hit_performance_50_methods(self) -> None: + """Benchmark cache hit performance with large method count.""" + + class LargeCacheNode(MixinNodeIntrospection): + pass + + # Add 50 methods + for i in range(50): + + async def method(self: LargeCacheNode, data: str = "") -> str: + return data + + method.__name__ = f"execute_task_{i:02d}" + setattr(LargeCacheNode, method.__name__, method) + + node = LargeCacheNode() + node.initialize_introspection( + node_id="large-cache-node", + node_type="COMPUTE", + event_bus=None, + ) + + # Warm cache + await node.get_introspection_data() + + # Measure cache hits + times: list[float] = [] + for _ in range(100): + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + avg_time = sum(times) / len(times) + max_time = max(times) + p99_time = sorted(times)[int(len(times) * 0.99)] + + print( + f"\nCache hit (50 methods): avg={avg_time:.3f}ms, max={max_time:.3f}ms, p99={p99_time:.3f}ms" + ) + + # Cache hits should be very fast regardless of method count + threshold_ms = PERF_THRESHOLD_CACHE_HIT_MS * PERF_MULTIPLIER + assert avg_time < threshold_ms, ( + f"Cache hit avg {avg_time:.3f}ms exceeds {threshold_ms:.1f}ms" + ) + + async def test_method_count_scaling_analysis(self) -> None: + """Analyze how introspection time scales with method count.""" + results: list[tuple[int, float]] = [] + + for method_count in [5, 10, 20, 30, 40, 50]: + # Create node class with specified method count + class ScalingTestNode(MixinNodeIntrospection): + pass + + for i in range(method_count): + + async def method(self: ScalingTestNode, data: str = "") -> str: + return data + + method.__name__ = f"execute_op_{i:02d}" + setattr(ScalingTestNode, method.__name__, method) + + node = ScalingTestNode() + node.initialize_introspection( + node_id=f"scaling-test-{method_count}", + node_type="COMPUTE", + event_bus=None, + ) + + # Clear cache and measure + MixinNodeIntrospection._invalidate_class_method_cache(ScalingTestNode) + + times: list[float] = [] + for _ in range(10): + node._introspection_cache = None + node._introspection_cached_at = None + MixinNodeIntrospection._invalidate_class_method_cache(ScalingTestNode) + + start = time.perf_counter() + await node.get_introspection_data() + times.append((time.perf_counter() - start) * 1000) + + avg_time = sum(times) / len(times) + results.append((method_count, avg_time)) + + # Clean up class from cache + MixinNodeIntrospection._invalidate_class_method_cache(ScalingTestNode) + + print("\n\nMethod Count Scaling Analysis:") + print("Methods | Avg Time (ms)") + print("--------|---------------") + for method_count, avg_time in results: + print(f" {method_count:3d} | {avg_time:.2f}") + + # All should be under threshold + threshold_ms = PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS * PERF_MULTIPLIER + for method_count, avg_time in results: + assert avg_time < threshold_ms, ( + f"{method_count} methods: {avg_time:.2f}ms exceeds {threshold_ms:.0f}ms" + ) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestMixinNodeIntrospectionThresholdDetection: + """Tests for threshold exceeded detection.""" + + async def test_threshold_not_exceeded_normal_operation(self) -> None: + """Test that thresholds are not marked exceeded in normal operation.""" + node = MockNode() + node.initialize_introspection( + node_id="threshold-normal-node", + node_type="EFFECT", + event_bus=None, + ) + + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + # Under normal operation with MockNode, thresholds should not be exceeded + # (unless running on very slow CI) + if metrics.total_introspection_ms < PERF_THRESHOLD_GET_INTROSPECTION_DATA_MS: + assert metrics.threshold_exceeded is False + assert len(metrics.slow_operations) == 0 + + async def test_slow_operations_list_populated_when_exceeded(self) -> None: + """Test that slow_operations is populated when threshold exceeded.""" + # This test verifies the structure is correct + # We can't reliably force slow operations in a unit test + node = MockNode() + node.initialize_introspection( + node_id="slow-ops-structure-node", + node_type="EFFECT", + event_bus=None, + ) + + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + # slow_operations should always be a list + assert isinstance(metrics.slow_operations, list) + + async def test_cache_hit_threshold_separate_from_total(self) -> None: + """Test that cache hit has its own performance threshold.""" + node = MockNode() + node.initialize_introspection( + node_id="cache-threshold-node", + node_type="EFFECT", + event_bus=None, + ) + + # First call - cache miss + await node.get_introspection_data() + + # Second call - cache hit + await node.get_introspection_data() + metrics = node.get_performance_metrics() + + assert metrics is not None + assert metrics.cache_hit is True + + # Cache hit should be very fast + # If it exceeds 1ms, there might be an issue + if metrics.total_introspection_ms < PERF_THRESHOLD_CACHE_HIT_MS: + assert "cache_hit" not in metrics.slow_operations