fix(mixins): remove Any type annotations from all mixins (OMN-1178)

src/omnibase_core/mixins/mixin_caching.py

@@ -22,7 +22,6 @@ async def execute_compute(self, contract):
 
 import hashlib
 import json
-from typing import Any
 
 from omnibase_core.types.typed_dict_mixin_types import TypedDictCacheStats
 
@@ -40,14 +39,14 @@ class MixinCaching:
         _cache_data: In-memory cache storage (stub) - stores serializable values
     """
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
+    def __init__(self, *args: object, **kwargs: object) -> None:
         """Initialize caching mixin."""
         super().__init__(*args, **kwargs)
         self._cache_enabled = True
         # Cache stores arbitrary serializable values - use object for type flexibility
         self._cache_data: dict[str, object] = {}
 
-    def generate_cache_key(self, data: Any) -> str:
+    def generate_cache_key(self, data: object) -> str:
         """
         Generate a cache key from data.
 
@@ -65,7 +64,7 @@ def generate_cache_key(self, data: Any) -> str:
             # Fallback for non-serializable data
             return hashlib.sha256(str(data).encode()).hexdigest()
 
-    async def get_cached(self, cache_key: str) -> Any | None:
+    async def get_cached(self, cache_key: str) -> object | None:
         """
         Retrieve cached value.
 
@@ -83,7 +82,7 @@ async def get_cached(self, cache_key: str) -> Any | None:
 
     # stub-ok: Intentional stub - full caching backend (Redis/Memcached) in future phase
     async def set_cached(
-        self, cache_key: str, value: Any, ttl_seconds: int = 3600
+        self, cache_key: str, value: object, ttl_seconds: int = 3600
     ) -> None:
         """
         Store value in cache.

src/omnibase_core/mixins/mixin_canonical_serialization.py

@@ -41,7 +41,7 @@ def _strip_comment_prefix(
     Remove leading comment prefixes from each line of a block.
     Args:
         block: Multiline string block to process.
-        comment_prefixes: Tuple/list[Any]of prefix strings to remove from line starts.
+        comment_prefixes: Tuple/list of prefix strings to remove from line starts.
     Returns:
         Block with comment prefixes removed from each line.
     """
@@ -65,9 +65,9 @@ class MixinCanonicalYAMLSerializer(ProtocolCanonicalSerializer):
     All field normalization and placeholder logic is schema-driven, using NodeMetadataBlock.model_fields.
     No hardcoded field names or types.
 
-    NOTE: Field order is always as declared in NodeMetadataBlock.model_fields, never by dict[str, Any]or YAML loader order. This is required for perfect idempotency.
+    NOTE: Field order is always as declared in NodeMetadataBlock.model_fields, never by dict or YAML loader order. This is required for perfect idempotency.
 
-    - All nested collections (list[Any]s of dict[str, Any]s, dict[str, Any]s of dict[str, Any]s) are sorted by a stable key (e.g., 'name' or dict[str, Any]key).
+    - All nested collections (lists of dicts, dicts of dicts) are sorted by a stable key (e.g., 'name' or dict key).
     - All booleans are normalized to lowercase YAML ('true'/'false').
     - All numbers are formatted with consistent precision.
     """
@@ -91,7 +91,7 @@ def canonicalize_metadata_block(
         """
         Canonicalize a metadata block for deterministic YAML serialization and hash computation.
         Args:
-            metadata_block: A dict[str, Any]or NodeMetadataBlock instance (must implement model_dump(mode="json")).
+            metadata_block: A dict[str, object] or NodeMetadataBlock instance (must implement model_dump(mode="json")).
             volatile_fields: Fields to replace with protocol placeholder values.
             placeholder: Placeholder value for volatile fields.
             sort_keys: Whether to sort keys in YAML output.
@@ -112,7 +112,7 @@ def canonicalize_metadata_block(
             # Preserve original dict for exception handler
             original_dict = metadata_block.copy()
 
-            # Convert dict[str, Any]to NodeMetadataBlock, handling type conversions
+            # Convert dict to NodeMetadataBlock, handling type conversions
             if "entrypoint" in metadata_block and isinstance(
                 metadata_block["entrypoint"], str
             ):
@@ -218,12 +218,9 @@ def canonicalize_metadata_block(
             # Check for Union types (both typing.Union and PEP 604 | syntax)
             # Note: PEP 604 unions (str | None) don't have __origin__ via getattr,
             # so we use isinstance(annotation, types.UnionType) to detect them
-            is_union = (
-                origin is Union  # Handles typing.Union
-                or isinstance(
-                    annotation, types.UnionType
-                )  # Handles PEP 604 (str | None)
-            )
+            is_union = origin is Union or isinstance(  # Handles typing.Union
+                annotation, types.UnionType
+            )  # Handles PEP 604 (str | None)
             if is_union and hasattr(annotation, "__args__"):
                 args = annotation.__args__
                 if str in args:
@@ -257,7 +254,7 @@ def canonicalize_metadata_block(
                 v = field.default if field.default is not None else ""
                 normalized_dict[k] = v
                 continue
-            # Normalize list[Any]fields
+            # Normalize list fields
             if k in list_fields and (v is None or v == "null"):
                 v = field.default if field.default is not None else []
                 normalized_dict[k] = v
@@ -268,10 +265,10 @@ def canonicalize_metadata_block(
             # Normalize numbers
             if isinstance(v, float):
                 v = format(v, ".15g")
-            # Sort list[Any]s of dict[str, Any]s
+            # Sort lists of dicts
             if isinstance(v, list) and v and all(isinstance(x, dict) for x in v):
                 v = sorted(v, key=lambda d: d["name"])
-            # Sort dict[str, Any]s
+            # Sort dicts
             if isinstance(v, dict):
                 v = dict(sorted(v.items()))
             # If still None, use default if available

src/omnibase_core/mixins/mixin_cli_handler.py

@@ -9,7 +9,6 @@
 import json
 import sys
 from pathlib import Path
-from typing import Any
 
 from omnibase_core.enums.enum_core_error_code import EnumCoreErrorCode
 from omnibase_core.enums.enum_log_level import EnumLogLevel as LogLevel
@@ -46,7 +45,7 @@ def process(self, input_state: MyInputState) -> MyOutputState:
             sys.exit(tool.main())
     """
 
-    def __init__(self, **kwargs: Any) -> None:
+    def __init__(self, **kwargs: object) -> None:
         """Initialize the CLI handler mixin."""
         super().__init__(**kwargs)
 
@@ -335,7 +334,7 @@ def _create_input_state(self, data: SerializedDict) -> InputStateT:
                 return cast("InputStateT", input_class(**data))
 
         # Fallback - return data as-is
-        return data  # type: ignore[return-value]
+        return data  # type: ignore[return-value]  # Fallback when input state class unavailable; dict substitutes for InputStateT
 
     def _handle_introspection(self) -> int:
         """Handle introspection request."""

src/omnibase_core/mixins/mixin_compute_execution.py

@@ -36,7 +36,6 @@
 """
 
 from collections import Counter
-from typing import Any
 from uuid import UUID, uuid4
 
 from omnibase_core.models.compute.model_compute_execution_context import (
@@ -92,7 +91,7 @@ class MixinComputeExecution:
     async def execute_contract_pipeline(
         self,
         contract: ModelComputeSubcontract,
-        input_data: Any,  # Any: pipeline accepts dict, Pydantic model, or JSON-compatible data
+        input_data: object,  # object: pipeline accepts dict, Pydantic model, or JSON-compatible data
         correlation_id: UUID | None = None,
     ) -> ModelComputePipelineResult:
         """

src/omnibase_core/mixins/mixin_contract_metadata.py

@@ -1,5 +1,4 @@
 from collections.abc import Mapping
-from typing import Any
 
 """
 Contract Metadata Mixin for ONEX Tool Nodes.
@@ -39,7 +38,7 @@ def __init__(self, **kwargs) -> None:
                 print(f"Tool: {self.node_name} v{self.node_version}")
     """
 
-    def __init__(self, **kwargs: Any) -> None:
+    def __init__(self, **kwargs: object) -> None:
         """Initialize the contract metadata mixin."""
         super().__init__(**kwargs)
 

src/omnibase_core/mixins/mixin_contract_state_reducer.py

@@ -9,7 +9,6 @@
 """
 
 from pathlib import Path
-from typing import Any
 
 from omnibase_core.enums.enum_core_error_code import EnumCoreErrorCode
 from omnibase_core.enums.enum_log_level import EnumLogLevel as LogLevel
@@ -39,7 +38,7 @@ def process(self, input_state):
                 return self.process_action_with_transitions(input_state)
     """
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
+    def __init__(self, *args: object, **kwargs: object) -> None:
         """Initialize contract state reducer mixin."""
         super().__init__(*args, **kwargs)
 
@@ -166,7 +165,7 @@ def _load_state_transitions(self) -> list[ModelStateTransition]:
             self._transitions_loaded = True
             return []
 
-    def process_action_with_transitions(self, input_state: Any) -> Any:
+    def process_action_with_transitions(self, input_state: object) -> object:
         """
         Process action using contract-driven state transitions.
 
@@ -180,7 +179,22 @@ def process_action_with_transitions(self, input_state: Any) -> Any:
         """
         try:
             tool_name = getattr(self, "node_name", "unknown_tool")
-            action_name = getattr(input_state.action, "action_name", "unknown_action")
+
+            # Safely access action attribute - handle None input_state
+            action: object | None = (
+                getattr(input_state, "action", None)
+                if input_state is not None
+                else None
+            )
+
+            # Safely access action_name - handle None, missing attribute, or empty string
+            # Note: Empty strings are intentionally treated as "unknown_action" since
+            # action_name must be a non-empty string per FSM model validation rules
+            action_name: str = "unknown_action"
+            if action is not None:
+                raw_action_name: object = getattr(action, "action_name", None)
+                if isinstance(raw_action_name, str) and raw_action_name:
+                    action_name = raw_action_name
 
             emit_log_event(
                 LogLevel.INFO,
@@ -195,12 +209,18 @@ def process_action_with_transitions(self, input_state: Any) -> Any:
             transitions = self._load_state_transitions()
 
             # Find transitions triggered by this action
+            # Safely handle potentially None triggers (defensive check)
             applicable_transitions = [
-                t for t in transitions if action_name in t.triggers
+                t
+                for t in transitions
+                if getattr(t, "triggers", None) and action_name in t.triggers
             ]
 
             # Apply transitions in priority order
-            applicable_transitions.sort(key=lambda t: t.priority, reverse=True)
+            # Use getattr for safe priority access with default of 0
+            applicable_transitions.sort(
+                key=lambda t: getattr(t, "priority", 0), reverse=True
+            )
 
             for transition in applicable_transitions:
                 self._apply_transition(transition, input_state)
@@ -212,7 +232,7 @@ def process_action_with_transitions(self, input_state: Any) -> Any:
             # Fallback: create basic success response
             return self._create_default_output_state(input_state)
 
-        except (RuntimeError, ValueError) as e:
+        except (AttributeError, RuntimeError, ValueError) as e:
             tool_name = getattr(self, "node_name", "unknown_tool")
             emit_log_event(
                 LogLevel.ERROR,
@@ -226,7 +246,7 @@ def process_action_with_transitions(self, input_state: Any) -> Any:
             ) from e
 
     def _apply_transition(
-        self, transition: ModelStateTransition, input_state: Any
+        self, transition: ModelStateTransition, input_state: object
     ) -> None:
         """
         Apply a single state transition.
@@ -236,75 +256,83 @@ def _apply_transition(
             input_state: Current input state
         """
         tool_name = getattr(self, "node_name", "unknown_tool")
+        # Defensive access to transition attributes to prevent AttributeError
+        transition_name = getattr(transition, "name", "unknown_transition")
+        transition_type = getattr(transition, "transition_type", None)
 
         try:
-            if transition.transition_type == EnumTransitionType.SIMPLE:
+            if transition_type == EnumTransitionType.SIMPLE:
                 self._apply_simple_transition(transition, input_state)
-            elif transition.transition_type == EnumTransitionType.TOOL_BASED:
+            elif transition_type == EnumTransitionType.TOOL_BASED:
                 self._apply_tool_based_transition(transition, input_state)
-            elif transition.transition_type == EnumTransitionType.CONDITIONAL:
+            elif transition_type == EnumTransitionType.CONDITIONAL:
                 self._apply_conditional_transition(transition, input_state)
             else:
                 emit_log_event(
                     LogLevel.WARNING,
-                    f"Unsupported transition type: {transition.transition_type}",
+                    f"Unsupported transition type: {transition_type}",
                     {
                         "tool_name": tool_name,
-                        "transition_name": transition.name,
+                        "transition_name": transition_name,
                     },
                 )
-        except (RuntimeError, ValueError) as e:
+        except (AttributeError, RuntimeError, ValueError) as e:
             emit_log_event(
                 LogLevel.ERROR,
-                f"Failed to apply transition {transition.name}: {e!s}",
+                f"Failed to apply transition {transition_name}: {e!s}",
                 {
                     "tool_name": tool_name,
-                    "transition_name": transition.name,
+                    "transition_name": transition_name,
                     "error": str(e),
                 },
             )
 
     def _apply_simple_transition(
         self,
         transition: ModelStateTransition,
-        input_state: Any,
+        input_state: object,
     ) -> None:
         """Apply simple field update transition."""
         # Simple transitions update state fields using template expressions
         # For now, just log the transition (actual field updates would require state management)
         tool_name = getattr(self, "node_name", "unknown_tool")
+        transition_name = getattr(transition, "name", "unknown_transition")
 
         emit_log_event(
             LogLevel.DEBUG,
-            f"Applied simple transition: {transition.name}",
+            f"Applied simple transition: {transition_name}",
             {
                 "tool_name": tool_name,
-                "transition_name": transition.name,
+                "transition_name": transition_name,
                 "transition_type": "simple",
             },
         )
 
     def _apply_tool_based_transition(
         self,
         transition: ModelStateTransition,
-        input_state: Any,
+        input_state: object,
     ) -> None:
         """Apply tool-based transition by delegating to specified tool."""
         tool_name = getattr(self, "node_name", "unknown_tool")
+        transition_name = getattr(transition, "name", "unknown_transition")
 
-        if not transition.tool_config:
+        # Defensive access to tool_config to prevent AttributeError
+        tool_config = getattr(transition, "tool_config", None)
+        if not tool_config:
             return
 
-        target_tool_name = transition.tool_config.tool_display_name or str(
-            transition.tool_config.tool_id
-        )
+        # Safely extract tool identifiers with fallback defaults
+        tool_display_name = getattr(tool_config, "tool_display_name", None)
+        tool_id = getattr(tool_config, "tool_id", None)
+        target_tool_name = tool_display_name or str(tool_id) if tool_id else "unknown"
 
         emit_log_event(
             LogLevel.DEBUG,
-            f"Applied tool-based transition: {transition.name} -> {target_tool_name}",
+            f"Applied tool-based transition: {transition_name} -> {target_tool_name}",
             {
                 "tool_name": tool_name,
-                "transition_name": transition.name,
+                "transition_name": transition_name,
                 "transition_type": "tool_based",
                 "target_tool": target_tool_name,
             },
@@ -313,23 +341,24 @@ def _apply_tool_based_transition(
     def _apply_conditional_transition(
         self,
         transition: ModelStateTransition,
-        input_state: Any,
+        input_state: object,
     ) -> None:
         """Apply conditional transition based on state conditions."""
         tool_name = getattr(self, "node_name", "unknown_tool")
+        transition_name = getattr(transition, "name", "unknown_transition")
 
         emit_log_event(
             LogLevel.DEBUG,
-            f"Applied conditional transition: {transition.name}",
+            f"Applied conditional transition: {transition_name}",
             {
                 "tool_name": tool_name,
-                "transition_name": transition.name,
+                "transition_name": transition_name,
                 "transition_type": "conditional",
             },
         )
 
     def _create_default_output_state(
-        self, input_state: Any
+        self, input_state: object
     ) -> TypedDictDefaultOutputState:
         """Create a default output state when no main tool is available."""
         # This is a fallback - each tool should implement proper processing