diff --git a/src/omnibase_core/enums/enum_core_error_code.py b/src/omnibase_core/enums/enum_core_error_code.py index 67a4ce60ae..b5b6c3f00b 100644 --- a/src/omnibase_core/enums/enum_core_error_code.py +++ b/src/omnibase_core/enums/enum_core_error_code.py @@ -144,6 +144,11 @@ class EnumCoreErrorCode(EnumOnexErrorCode): NODE_EXECUTION_ERROR = "ONEX_CORE_273_NODE_EXECUTION_ERROR" UNSUPPORTED_CAPABILITY_ERROR = "ONEX_CORE_274_UNSUPPORTED_CAPABILITY_ERROR" + # Workflow execution limit errors (281-290) + WORKFLOW_STEP_LIMIT_EXCEEDED = "ONEX_CORE_281_WORKFLOW_STEP_LIMIT_EXCEEDED" + WORKFLOW_PAYLOAD_SIZE_EXCEEDED = "ONEX_CORE_282_WORKFLOW_PAYLOAD_SIZE_EXCEEDED" + WORKFLOW_TOTAL_PAYLOAD_EXCEEDED = "ONEX_CORE_283_WORKFLOW_TOTAL_PAYLOAD_EXCEEDED" + def get_component(self) -> str: """Get the component identifier for this error code.""" return "CORE" @@ -258,6 +263,10 @@ def get_exit_code(self) -> int: EnumCoreErrorCode.PURITY_VIOLATION_ERROR: EnumCLIExitCode.ERROR, EnumCoreErrorCode.NODE_EXECUTION_ERROR: EnumCLIExitCode.ERROR, EnumCoreErrorCode.UNSUPPORTED_CAPABILITY_ERROR: EnumCLIExitCode.ERROR, + # Workflow execution limit errors -> ERROR + EnumCoreErrorCode.WORKFLOW_STEP_LIMIT_EXCEEDED: EnumCLIExitCode.ERROR, + EnumCoreErrorCode.WORKFLOW_PAYLOAD_SIZE_EXCEEDED: EnumCLIExitCode.ERROR, + EnumCoreErrorCode.WORKFLOW_TOTAL_PAYLOAD_EXCEEDED: EnumCLIExitCode.ERROR, } @@ -375,5 +384,8 @@ def get_core_error_description(error_code: EnumCoreErrorCode) -> str: EnumCoreErrorCode.PURITY_VIOLATION_ERROR: "Purity violation in pure node", EnumCoreErrorCode.NODE_EXECUTION_ERROR: "Node execution failed", EnumCoreErrorCode.UNSUPPORTED_CAPABILITY_ERROR: "Unsupported capability requested", + EnumCoreErrorCode.WORKFLOW_STEP_LIMIT_EXCEEDED: "Workflow step count exceeds maximum limit", + EnumCoreErrorCode.WORKFLOW_PAYLOAD_SIZE_EXCEEDED: "Workflow step payload size exceeds maximum limit", + EnumCoreErrorCode.WORKFLOW_TOTAL_PAYLOAD_EXCEEDED: "Workflow total payload size exceeds maximum limit", } return descriptions.get(error_code, "Unknown error") diff --git a/src/omnibase_core/infrastructure/node_config_provider.py b/src/omnibase_core/infrastructure/node_config_provider.py index 5a9b15f677..ad193972a5 100644 --- a/src/omnibase_core/infrastructure/node_config_provider.py +++ b/src/omnibase_core/infrastructure/node_config_provider.py @@ -12,6 +12,7 @@ from omnibase_core.models.configuration.model_node_config_value import ( ConfigValue, ModelNodeConfigSchema, + is_valid_value_type, ) @@ -266,9 +267,16 @@ async def get_config_schema(self) -> dict[str, ModelNodeConfigSchema]: """ schema: dict[str, ModelNodeConfigSchema] = {} for key, value in self._DEFAULTS.items(): + type_name = type(value).__name__ + # Type guard validates and narrows type_name to VALID_VALUE_TYPES + if not is_valid_value_type(type_name): + raise ValueError( # error-ok: internal bootstrap validation before OnexError available + f"Invalid config type '{type_name}' for key '{key}'. " + f"Allowed types: ('int', 'float', 'bool', 'str')" + ) schema[key] = ModelNodeConfigSchema( key=key, - type=type(value).__name__, + type=type_name, default=value, ) return schema diff --git a/src/omnibase_core/models/configuration/__init__.py b/src/omnibase_core/models/configuration/__init__.py index 3564dbabd4..a78141e66d 100644 --- a/src/omnibase_core/models/configuration/__init__.py +++ b/src/omnibase_core/models/configuration/__init__.py @@ -9,8 +9,15 @@ ModelTierConfig, ) from .model_compute_cache_config import ModelComputeCacheConfig +from .model_git_hub_comment_change import ModelGitHubCommentChange +from .model_git_hub_issue_comment_changes import ModelGitHubIssueCommentChanges +from .model_git_hub_issue_comment_event import ModelGitHubIssueCommentEvent from .model_node_config_entry import ConfigValue, ModelNodeConfigEntry from .model_node_config_value import ModelNodeConfigSchema +from .model_priority_metadata import ModelPriorityMetadata +from .model_priority_metadata_summary import ModelPriorityMetadataSummary +from .model_throttle_response import ModelThrottleResponse +from .model_throttling_behavior import ModelThrottlingBehavior __all__ = [ "ConfigValue", @@ -18,9 +25,16 @@ "ModelCLIConfig", "ModelComputeCacheConfig", "ModelDatabaseConfig", + "ModelGitHubCommentChange", + "ModelGitHubIssueCommentChanges", + "ModelGitHubIssueCommentEvent", "ModelMonitoringConfig", "ModelNodeConfigEntry", "ModelNodeConfigSchema", "ModelOutputConfig", + "ModelPriorityMetadata", + "ModelPriorityMetadataSummary", + "ModelThrottleResponse", + "ModelThrottlingBehavior", "ModelTierConfig", ] diff --git a/src/omnibase_core/models/configuration/model_config_types.py b/src/omnibase_core/models/configuration/model_config_types.py new file mode 100644 index 0000000000..7101b21fae --- /dev/null +++ b/src/omnibase_core/models/configuration/model_config_types.py @@ -0,0 +1,57 @@ +"""Common types for configuration models.""" + +from typing import Literal, TypeGuard + +# Type alias for valid configuration value types +ConfigValue = int | float | bool | str + +# Literal type constraining value_type/config_type to valid values +VALID_VALUE_TYPES = Literal["int", "float", "bool", "str"] + +# Tuple of valid type names for runtime checking +_VALID_TYPE_NAMES: tuple[str, ...] = ("int", "float", "bool", "str") + + +def is_valid_value_type(type_name: str) -> TypeGuard[VALID_VALUE_TYPES]: + """Type guard to check if a string is a valid value type. + + This function narrows the type of type_name from str to VALID_VALUE_TYPES + when it returns True, enabling mypy to understand the type refinement. + + Args: + type_name: The type name string to validate. + + Returns: + True if type_name is one of 'int', 'float', 'bool', 'str'. + """ + return type_name in _VALID_TYPE_NAMES + + +def validate_config_value_type( + value_type: VALID_VALUE_TYPES, default: ConfigValue +) -> None: + """Validate that default value matches declared type. + + Args: + value_type: The declared type ('int', 'float', 'bool', 'str') + default: The default value to validate + + Raises: + ValueError: If default value doesn't match declared type + """ + type_map: dict[str, type | tuple[type, ...]] = { + "int": int, + "float": (int, float), # int is valid for float + "bool": bool, + "str": str, + } + expected = type_map[value_type] + # Strict bool check - don't allow int/float to match bool + if value_type == "bool" and not isinstance(default, bool): + raise ValueError( # error-ok: Pydantic validator requires ValueError + f"default must be bool, got {type(default).__name__}" + ) + if not isinstance(default, expected): + raise ValueError( # error-ok: Pydantic validator requires ValueError + f"default must be {value_type}, got {type(default).__name__}" + ) diff --git a/src/omnibase_core/models/configuration/model_git_hub_comment_change.py b/src/omnibase_core/models/configuration/model_git_hub_comment_change.py new file mode 100644 index 0000000000..35a5b4706b --- /dev/null +++ b/src/omnibase_core/models/configuration/model_git_hub_comment_change.py @@ -0,0 +1,9 @@ +"""ModelGitHubCommentChange model.""" + +from pydantic import BaseModel, Field + + +class ModelGitHubCommentChange(BaseModel): + """Represents a change to a GitHub comment field.""" + + from_: str | None = Field(default=None, alias="from", description="Previous value") diff --git a/src/omnibase_core/models/configuration/model_git_hub_issue_comment_changes.py b/src/omnibase_core/models/configuration/model_git_hub_issue_comment_changes.py new file mode 100644 index 0000000000..7e3947fbeb --- /dev/null +++ b/src/omnibase_core/models/configuration/model_git_hub_issue_comment_changes.py @@ -0,0 +1,13 @@ +"""ModelGitHubIssueCommentChanges model.""" + +from pydantic import BaseModel, Field + +from .model_git_hub_comment_change import ModelGitHubCommentChange + + +class ModelGitHubIssueCommentChanges(BaseModel): + """Changes made to a GitHub issue comment (for edited action).""" + + body: ModelGitHubCommentChange | None = Field( + default=None, description="Body content change" + ) diff --git a/src/omnibase_core/models/configuration/model_git_hub_issue_comment_event.py b/src/omnibase_core/models/configuration/model_git_hub_issue_comment_event.py index 4605d4b52c..b51d4b8f4b 100644 --- a/src/omnibase_core/models/configuration/model_git_hub_issue_comment_event.py +++ b/src/omnibase_core/models/configuration/model_git_hub_issue_comment_event.py @@ -1,29 +1,19 @@ -from pydantic import Field +"""GitHubIssueCommentEvent model.""" -""" -GitHubIssueCommentEvent model. -""" - -from pydantic import BaseModel +from pydantic import BaseModel, Field +from .model_git_hub_comment_change import ModelGitHubCommentChange from .model_git_hub_issue import ModelGitHubIssue from .model_git_hub_issue_comment import ModelGitHubIssueComment +from .model_git_hub_issue_comment_changes import ModelGitHubIssueCommentChanges from .model_git_hub_repository import ModelGitHubRepository from .model_git_hub_user import ModelGitHubUser - -class ModelGitHubCommentChange(BaseModel): - """Represents a change to a GitHub comment field.""" - - from_: str | None = Field(default=None, alias="from", description="Previous value") - - -class ModelGitHubIssueCommentChanges(BaseModel): - """Changes made to a GitHub issue comment (for edited action).""" - - body: ModelGitHubCommentChange | None = Field( - default=None, description="Body content change" - ) +__all__ = [ + "ModelGitHubCommentChange", + "ModelGitHubIssueCommentChanges", + "ModelGitHubIssueCommentEvent", +] class ModelGitHubIssueCommentEvent(BaseModel): diff --git a/src/omnibase_core/models/configuration/model_node_config_entry.py b/src/omnibase_core/models/configuration/model_node_config_entry.py index 3f8aa7951d..5941517e8d 100644 --- a/src/omnibase_core/models/configuration/model_node_config_entry.py +++ b/src/omnibase_core/models/configuration/model_node_config_entry.py @@ -1,11 +1,12 @@ """Model for node configuration entry.""" -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator -# Type alias for valid configuration value types -# IMPORTANT: bool MUST come before int because bool is a subclass of int in Python -# If int comes first, Pydantic will match True/False as 1/0 integers -ConfigValue = bool | int | float | str +from omnibase_core.models.configuration.model_config_types import ( + VALID_VALUE_TYPES, + ConfigValue, + validate_config_value_type, +) class ModelNodeConfigEntry(BaseModel): @@ -25,7 +26,7 @@ class ModelNodeConfigEntry(BaseModel): ..., description="Configuration key (e.g., 'compute.max_parallel_workers')", ) - value_type: str = Field( + value_type: VALID_VALUE_TYPES = Field( ..., description="Type name of the value ('int', 'float', 'bool', 'str')", ) @@ -35,3 +36,9 @@ class ModelNodeConfigEntry(BaseModel): ) model_config = {"frozen": True} + + @model_validator(mode="after") + def validate_default_type(self) -> "ModelNodeConfigEntry": + """Ensure default value type matches declared value_type.""" + validate_config_value_type(self.value_type, self.default) + return self diff --git a/src/omnibase_core/models/configuration/model_node_config_schema.py b/src/omnibase_core/models/configuration/model_node_config_schema.py new file mode 100644 index 0000000000..dfbb12e888 --- /dev/null +++ b/src/omnibase_core/models/configuration/model_node_config_schema.py @@ -0,0 +1,37 @@ +"""Model for node configuration schema.""" + +from pydantic import BaseModel, Field, model_validator + +from omnibase_core.models.configuration.model_config_types import ( + VALID_VALUE_TYPES, + ConfigValue, + validate_config_value_type, +) + + +class ModelNodeConfigSchema(BaseModel): + """ + Strongly-typed model for node configuration schema. + + Represents a complete configuration schema entry including + key, type, and default value information. + + Attributes: + key: Configuration key + config_type: Type name of the value ('int', 'float', 'bool', 'str') + default: Default value + """ + + key: str = Field(..., description="Configuration key") + config_type: VALID_VALUE_TYPES = Field( + ..., alias="type", description="Type name of the value" + ) + default: ConfigValue = Field(..., description="Default value") + + model_config = {"frozen": True, "populate_by_name": True} + + @model_validator(mode="after") + def validate_default_type(self) -> "ModelNodeConfigSchema": + """Ensure default value type matches declared config_type.""" + validate_config_value_type(self.config_type, self.default) + return self diff --git a/src/omnibase_core/models/configuration/model_node_config_value.py b/src/omnibase_core/models/configuration/model_node_config_value.py index ded2ea2d73..5060798b6a 100644 --- a/src/omnibase_core/models/configuration/model_node_config_value.py +++ b/src/omnibase_core/models/configuration/model_node_config_value.py @@ -1,30 +1,23 @@ -"""Model for node configuration schema.""" +"""Model for node configuration values - re-exports from split files.""" -from pydantic import BaseModel, Field - -from omnibase_core.models.configuration.model_node_config_entry import ( +from omnibase_core.models.configuration.model_config_types import ( + VALID_VALUE_TYPES, ConfigValue, + is_valid_value_type, + validate_config_value_type, +) +from omnibase_core.models.configuration.model_node_config_entry import ( ModelNodeConfigEntry, ) +from omnibase_core.models.configuration.model_node_config_schema import ( + ModelNodeConfigSchema, +) -__all__ = ["ConfigValue", "ModelNodeConfigEntry", "ModelNodeConfigSchema"] - - -class ModelNodeConfigSchema(BaseModel): - """ - Strongly-typed model for node configuration schema. - - Represents a complete configuration schema entry including - key, type, and default value information. - - Attributes: - key: Configuration key - type: Type name of the value - default: Default value - """ - - key: str = Field(..., description="Configuration key") - type: str = Field(..., description="Type name of the value") - default: ConfigValue = Field(..., description="Default value") - - model_config = {"frozen": True} +__all__ = [ + "ConfigValue", + "ModelNodeConfigEntry", + "ModelNodeConfigSchema", + "VALID_VALUE_TYPES", + "is_valid_value_type", + "validate_config_value_type", +] diff --git a/src/omnibase_core/models/configuration/model_priority_metadata.py b/src/omnibase_core/models/configuration/model_priority_metadata.py index f745d64619..00857afbf2 100644 --- a/src/omnibase_core/models/configuration/model_priority_metadata.py +++ b/src/omnibase_core/models/configuration/model_priority_metadata.py @@ -15,27 +15,12 @@ from omnibase_core.models.configuration.model_notification_settings import ( ModelNotificationSettings, ) +from omnibase_core.models.configuration.model_priority_metadata_summary import ( + ModelPriorityMetadataSummary, +) from omnibase_core.models.errors.model_onex_error import ModelOnexError - -class ModelPriorityMetadataSummary(BaseModel): - """Summary of priority metadata.""" - - owner: str | None = Field(default=None, description="Owner of the priority") - approval_required: bool = Field( - default=False, description="Whether approval is required" - ) - approved_users_count: int = Field(default=0, description="Number of approved users") - approved_groups_count: int = Field( - default=0, description="Number of approved groups" - ) - tags_count: int = Field(default=0, description="Number of tags") - has_sla: bool = Field(default=False, description="Whether SLA is defined") - has_cost: bool = Field(default=False, description="Whether cost is defined") - has_usage_limit: bool = Field( - default=False, description="Whether usage limit is defined" - ) - age_days: float = Field(default=0.0, description="Age in days since creation") +__all__ = ["ModelPriorityMetadata", "ModelPriorityMetadataSummary"] class ModelPriorityMetadata(BaseModel): diff --git a/src/omnibase_core/models/configuration/model_priority_metadata_summary.py b/src/omnibase_core/models/configuration/model_priority_metadata_summary.py new file mode 100644 index 0000000000..abb3a03a97 --- /dev/null +++ b/src/omnibase_core/models/configuration/model_priority_metadata_summary.py @@ -0,0 +1,26 @@ +"""ONEX-compatible Priority Metadata Summary Model. + +Summary model for priority metadata with ONEX compliance. +""" + +from pydantic import BaseModel, Field + + +class ModelPriorityMetadataSummary(BaseModel): + """Summary of priority metadata.""" + + owner: str | None = Field(default=None, description="Owner of the priority") + approval_required: bool = Field( + default=False, description="Whether approval is required" + ) + approved_users_count: int = Field(default=0, description="Number of approved users") + approved_groups_count: int = Field( + default=0, description="Number of approved groups" + ) + tags_count: int = Field(default=0, description="Number of tags") + has_sla: bool = Field(default=False, description="Whether SLA is defined") + has_cost: bool = Field(default=False, description="Whether cost is defined") + has_usage_limit: bool = Field( + default=False, description="Whether usage limit is defined" + ) + age_days: float = Field(default=0.0, description="Age in days since creation") diff --git a/src/omnibase_core/models/configuration/model_throttle_response.py b/src/omnibase_core/models/configuration/model_throttle_response.py new file mode 100644 index 0000000000..7da1280c4a --- /dev/null +++ b/src/omnibase_core/models/configuration/model_throttle_response.py @@ -0,0 +1,19 @@ +""" +ModelThrottleResponse - Typed response for throttled requests. + +This model defines the structure for HTTP responses returned when requests +are throttled due to rate limiting. +""" + +from pydantic import BaseModel, Field + + +class ModelThrottleResponse(BaseModel): + """Typed response for throttled requests.""" + + status_code: int = Field(default=429, description="HTTP status code") + headers: dict[str, str] = Field( + default_factory=dict, description="Response headers" + ) + message: str = Field(default="", description="Response message") + body: str | None = Field(default=None, description="Custom response body") diff --git a/src/omnibase_core/models/configuration/model_throttling_behavior.py b/src/omnibase_core/models/configuration/model_throttling_behavior.py index 1d04e9524d..d4ce31f8ff 100644 --- a/src/omnibase_core/models/configuration/model_throttling_behavior.py +++ b/src/omnibase_core/models/configuration/model_throttling_behavior.py @@ -1,5 +1,3 @@ -from pydantic import Field - """ ModelThrottlingBehavior - Throttling behavior configuration for rate limiting @@ -7,18 +5,11 @@ are exceeded, including blocking, queuing, delay, and custom responses. """ -from pydantic import BaseModel - +from pydantic import BaseModel, Field -class ModelThrottleResponse(BaseModel): - """Typed response for throttled requests.""" - - status_code: int = Field(default=429, description="HTTP status code") - headers: dict[str, str] = Field( - default_factory=dict, description="Response headers" - ) - message: str = Field(default="", description="Response message") - body: str | None = Field(default=None, description="Custom response body") +from omnibase_core.models.configuration.model_throttle_response import ( + ModelThrottleResponse, +) class ModelThrottlingBehavior(BaseModel): diff --git a/src/omnibase_core/models/core/__init__.py b/src/omnibase_core/models/core/__init__.py index 8d414d809d..9f1f8d4d70 100644 --- a/src/omnibase_core/models/core/__init__.py +++ b/src/omnibase_core/models/core/__init__.py @@ -52,6 +52,12 @@ # Generic metadata pattern from .model_protocol_metadata import ModelGenericMetadata from .model_result_accessor import ModelResultAccessor + +# Tool integration models +from .model_tool_integration import ModelToolIntegration +from .model_tool_integration_summary import ModelToolIntegrationSummary +from .model_tool_resource_requirements import ModelToolResourceRequirements +from .model_tool_timeout_settings import ModelToolTimeoutSettings from .model_typed_accessor import ModelTypedAccessor from .model_typed_configuration import ModelTypedConfiguration @@ -186,4 +192,9 @@ "ModelNodeVersionConstraints", # Workflow models "ModelWorkflow", + # Tool integration models + "ModelToolIntegration", + "ModelToolIntegrationSummary", + "ModelToolResourceRequirements", + "ModelToolTimeoutSettings", ] diff --git a/src/omnibase_core/models/core/model_environment.py b/src/omnibase_core/models/core/model_environment.py index 3590fdfc8f..07c948d343 100644 --- a/src/omnibase_core/models/core/model_environment.py +++ b/src/omnibase_core/models/core/model_environment.py @@ -498,5 +498,9 @@ def create_local(cls) -> "ModelEnvironment": return env -# NOTE: model_rebuild() moved to __init__.py after all imports are available -# Forward references will be resolved when __init__.py imports ModelEnvironment +# NOTE: model_rebuild() must be called lazily to avoid circular imports. +# The forward references (ModelSecurityLevel, ModelResourceLimits) are resolved +# either on first use or explicitly by calling: +# from omnibase_core.models.security.model_security_level import ModelSecurityLevel +# from omnibase_core.models.configuration.model_resource_limits import ModelResourceLimits +# ModelEnvironment.model_rebuild() diff --git a/src/omnibase_core/models/core/model_tool_integration.py b/src/omnibase_core/models/core/model_tool_integration.py index 8a6b03c85a..bb21c05e10 100644 --- a/src/omnibase_core/models/core/model_tool_integration.py +++ b/src/omnibase_core/models/core/model_tool_integration.py @@ -6,42 +6,22 @@ from pydantic import BaseModel, Field - -class ModelToolResourceRequirements(BaseModel): - """Resource requirements for a tool.""" - - requires_separate_port: bool = Field( - description="Whether tool requires separate HTTP port" - ) - health_check_via_service: bool = Field( - description="Whether health checked by parent service" - ) - loaded_as_module: bool = Field(description="Whether loaded as module by service") - - -class ModelToolTimeoutSettings(BaseModel): - """Timeout settings for a tool.""" - - shutdown_timeout: int = Field(description="Graceful shutdown timeout in seconds") - initialization_order: int = Field( - description="Initialization order relative to other tools" - ) - - -class ModelToolIntegrationSummary(BaseModel): - """Summary of tool integration configuration.""" - - auto_load_strategy: str = Field(description="Strategy for loading tool versions") - has_fallback_versions: bool = Field(description="Whether has fallback versions") - fallback_versions_count: int = Field(description="Number of fallback versions") - directory_pattern_type: str = Field(description="Directory pattern type") - implementation_file: str = Field(description="Main implementation file name") - contract_file: str = Field(description="Contract file name") - main_class_name: str = Field(description="Main implementation class name") - resources: ModelToolResourceRequirements = Field( - description="Resource requirements" - ) - timeout_settings: ModelToolTimeoutSettings = Field(description="Timeout settings") +from omnibase_core.models.core.model_tool_integration_summary import ( + ModelToolIntegrationSummary, +) +from omnibase_core.models.core.model_tool_resource_requirements import ( + ModelToolResourceRequirements, +) +from omnibase_core.models.core.model_tool_timeout_settings import ( + ModelToolTimeoutSettings, +) + +__all__ = [ + "ModelToolIntegration", + "ModelToolIntegrationSummary", + "ModelToolResourceRequirements", + "ModelToolTimeoutSettings", +] class ModelToolIntegration(BaseModel): diff --git a/src/omnibase_core/models/core/model_tool_integration_summary.py b/src/omnibase_core/models/core/model_tool_integration_summary.py new file mode 100644 index 0000000000..740895cbd5 --- /dev/null +++ b/src/omnibase_core/models/core/model_tool_integration_summary.py @@ -0,0 +1,26 @@ +""" +Tool Integration Summary Model. + +Summary of tool integration configuration. +""" + +from pydantic import BaseModel, Field + +from .model_tool_resource_requirements import ModelToolResourceRequirements +from .model_tool_timeout_settings import ModelToolTimeoutSettings + + +class ModelToolIntegrationSummary(BaseModel): + """Summary of tool integration configuration.""" + + auto_load_strategy: str = Field(description="Strategy for loading tool versions") + has_fallback_versions: bool = Field(description="Whether has fallback versions") + fallback_versions_count: int = Field(description="Number of fallback versions") + directory_pattern_type: str = Field(description="Directory pattern type") + implementation_file: str = Field(description="Main implementation file name") + contract_file: str = Field(description="Contract file name") + main_class_name: str = Field(description="Main implementation class name") + resources: ModelToolResourceRequirements = Field( + description="Resource requirements" + ) + timeout_settings: ModelToolTimeoutSettings = Field(description="Timeout settings") diff --git a/src/omnibase_core/models/core/model_tool_resource_requirements.py b/src/omnibase_core/models/core/model_tool_resource_requirements.py new file mode 100644 index 0000000000..ac93dea655 --- /dev/null +++ b/src/omnibase_core/models/core/model_tool_resource_requirements.py @@ -0,0 +1,19 @@ +""" +Tool Resource Requirements Model. + +Resource requirements configuration for tools. +""" + +from pydantic import BaseModel, Field + + +class ModelToolResourceRequirements(BaseModel): + """Resource requirements for a tool.""" + + requires_separate_port: bool = Field( + description="Whether tool requires separate HTTP port" + ) + health_check_via_service: bool = Field( + description="Whether health checked by parent service" + ) + loaded_as_module: bool = Field(description="Whether loaded as module by service") diff --git a/src/omnibase_core/models/core/model_tool_timeout_settings.py b/src/omnibase_core/models/core/model_tool_timeout_settings.py new file mode 100644 index 0000000000..5e7c45867c --- /dev/null +++ b/src/omnibase_core/models/core/model_tool_timeout_settings.py @@ -0,0 +1,16 @@ +""" +Tool Timeout Settings Model. + +Timeout configuration for tools. +""" + +from pydantic import BaseModel, Field + + +class ModelToolTimeoutSettings(BaseModel): + """Timeout settings for a tool.""" + + shutdown_timeout: int = Field(description="Graceful shutdown timeout in seconds") + initialization_order: int = Field( + description="Initialization order relative to other tools" + ) diff --git a/src/omnibase_core/models/service/model_service_compute.py b/src/omnibase_core/models/service/model_service_compute.py index 9c33d33b15..322c32ae55 100644 --- a/src/omnibase_core/models/service/model_service_compute.py +++ b/src/omnibase_core/models/service/model_service_compute.py @@ -109,7 +109,7 @@ class ModelServiceCompute( MixinHealthCheck, MixinCaching, MixinMetrics, - Generic[T_Input, T_Output], # noqa: UP046 - Traditional syntax required for mypy CI compatibility + Generic[T_Input, T_Output], # noqa: UP046 - Traditional Generic syntax required for mypy CI compatibility ): """ Standard Compute Node Service following ONEX model naming conventions. diff --git a/src/omnibase_core/models/service/model_service_reducer.py b/src/omnibase_core/models/service/model_service_reducer.py index 88ccec2957..4ac31f7f51 100644 --- a/src/omnibase_core/models/service/model_service_reducer.py +++ b/src/omnibase_core/models/service/model_service_reducer.py @@ -70,7 +70,7 @@ class ModelServiceReducer( MixinHealthCheck, MixinCaching, MixinMetrics, - Generic[T_Input, T_Output], # noqa: UP046 - Traditional syntax required for mypy CI compatibility + Generic[T_Input, T_Output], # noqa: UP046 - Traditional Generic syntax required for mypy CI compatibility ): """ Standard Reducer Node Service following ONEX model naming conventions. diff --git a/src/omnibase_core/utils/workflow_executor.py b/src/omnibase_core/utils/workflow_executor.py index 805d218f34..b39357a28f 100644 --- a/src/omnibase_core/utils/workflow_executor.py +++ b/src/omnibase_core/utils/workflow_executor.py @@ -25,12 +25,56 @@ - ModelAction.priority constraint: models/orchestrator/model_action.py - Architecture rationale: docs/architecture/CONTRACT_DRIVEN_NODEORCHESTRATOR_V1_0.md (Section: "Step Priority vs Action Priority") + +Size Limits: + To prevent memory exhaustion, this module enforces the following limits: + - MAX_WORKFLOW_STEPS (default 1000): Maximum number of steps in a workflow. + Validated during workflow validation; raises ModelOnexError if exceeded. + - MAX_STEP_PAYLOAD_SIZE_BYTES (default 64KB): Maximum size of individual step payload. + Validated during action creation; raises ModelOnexError if exceeded. + - MAX_TOTAL_PAYLOAD_SIZE_BYTES (default 10MB): Maximum accumulated payload size. + In parallel mode, raises ModelOnexError. In sequential mode, treated as + a step failure (per error_action configuration, defaults to continue). + + These limits are configurable via environment variables for extreme workloads: + - ONEX_MAX_WORKFLOW_STEPS: Override max workflow steps (bounds: 1-100,000) + - ONEX_MAX_STEP_PAYLOAD_SIZE_BYTES: Override max step payload size (bounds: 1KB-10MB) + - ONEX_MAX_TOTAL_PAYLOAD_SIZE_BYTES: Override max total payload size (bounds: 1KB-1GB) + + Invalid environment variable values log a warning and fall back to defaults. + Bounds are enforced to prevent both DoS attacks (too-small limits causing many + small workflows) and memory exhaustion (too-large limits). + +Security Considerations: + Compression Attacks: + The payload size limits (MAX_STEP_PAYLOAD_SIZE_BYTES, MAX_TOTAL_PAYLOAD_SIZE_BYTES) + are validated on the SERIALIZED JSON payload, not on compressed data. If payloads + are transmitted compressed (e.g., gzip), they MUST be validated for size AFTER + decompression but BEFORE processing. + + Risk: A "compression bomb" (also known as a "zip bomb" or "decompression bomb") + is a small compressed file that expands to a massive size when decompressed. + Without post-decompression size validation, an attacker could bypass payload + limits by sending small compressed payloads that expand to memory-exhausting + sizes, leading to denial-of-service (DoS) conditions. + + Mitigation: + - Always validate payload size on the decompressed/deserialized data + - Consider adding streaming decompression with size limits for transport layers + - The workflow executor validates size on json.dumps() output, which is the + serialized (uncompressed) representation + - Transport layers that accept compressed payloads should enforce their own + decompression limits before passing data to the workflow executor + + Reference: OWASP - Denial of Service through Resource Depletion + (https://owasp.org/www-community/attacks/Denial_of_Service) """ import asyncio import heapq import json import logging +import os import time from datetime import datetime from uuid import UUID, uuid4 @@ -57,6 +101,131 @@ from omnibase_core.validation.reserved_enum_validator import validate_execution_mode +def _get_limit_from_env(env_var: str, default: int, min_val: int, max_val: int) -> int: + """Get limit from environment variable with bounds checking. + + Args: + env_var: Environment variable name + default: Default value if env var not set + min_val: Minimum allowed value + max_val: Maximum allowed value + + Returns: + Validated limit value + """ + value = os.environ.get(env_var) + if value is None: + return default + try: + int_value = int(value) + return max(min_val, min(int_value, max_val)) + except ValueError: + logging.warning( + f"Invalid value for {env_var}: {value}, using default {default}" + ) + return default + + +# Workflow execution limits (OMN-670: Security hardening) +# Configurable via environment variables for extreme workloads +# Bounds prevent both too-small (DoS via many small workflows) and too-large (memory exhaustion) values +MAX_WORKFLOW_STEPS = _get_limit_from_env( + "ONEX_MAX_WORKFLOW_STEPS", default=1000, min_val=1, max_val=100000 +) +MAX_STEP_PAYLOAD_SIZE_BYTES = _get_limit_from_env( + "ONEX_MAX_STEP_PAYLOAD_SIZE_BYTES", + default=64 * 1024, + min_val=1024, + max_val=10 * 1024 * 1024, +) +MAX_TOTAL_PAYLOAD_SIZE_BYTES = _get_limit_from_env( + "ONEX_MAX_TOTAL_PAYLOAD_SIZE_BYTES", + default=10 * 1024 * 1024, + min_val=1024, + max_val=1024 * 1024 * 1024, +) + + +def _log_payload_metrics( + workflow_id: UUID, + step_id: UUID, + step_name: str, + payload_size: int, + total_payload_size: int, +) -> None: + """Log payload size metrics for observability. + + In production, these structured logs can be captured by log aggregators + (DataDog, Splunk, etc.) and converted to metrics. This approach avoids + adding external metric dependencies to the core library. + + Args: + workflow_id: Workflow execution ID + step_id: Step ID + step_name: Step name for context + payload_size: Size of this step's payload in bytes + total_payload_size: Cumulative payload size for the workflow + """ + # Only log at DEBUG level to avoid noise in production + # Structured format enables log-based metrics extraction + logging.debug( + "workflow_payload_metrics", + extra={ + "metric_type": "payload_size", + "workflow_id": str(workflow_id), + "step_id": str(step_id), + "step_name": step_name, + "payload_size_bytes": payload_size, + "total_payload_size_bytes": total_payload_size, + "payload_size_pct_of_limit": round( + payload_size / MAX_STEP_PAYLOAD_SIZE_BYTES * 100, 2 + ), + "total_payload_pct_of_limit": round( + total_payload_size / MAX_TOTAL_PAYLOAD_SIZE_BYTES * 100, 2 + ), + }, + ) + + +def _log_workflow_completion_metrics( + workflow_id: UUID, + workflow_name: str, + total_payload_size: int, + step_count: int, + execution_mode: str, +) -> None: + """Log workflow completion metrics for observability. + + Summarizes workflow execution with aggregated payload statistics. + In production, these structured logs can be captured by log aggregators + (DataDog, Splunk, etc.) and converted to metrics. + + Args: + workflow_id: Workflow execution ID + workflow_name: Workflow name for context + total_payload_size: Total accumulated payload size in bytes + step_count: Number of steps executed + execution_mode: Execution mode (sequential, parallel, batch) + """ + logging.debug( + "workflow_completion_metrics", + extra={ + "metric_type": "workflow_complete", + "workflow_id": str(workflow_id), + "workflow_name": workflow_name, + "total_payload_size_bytes": total_payload_size, + "step_count": step_count, + "execution_mode": execution_mode, + "avg_payload_size_bytes": total_payload_size // step_count + if step_count > 0 + else 0, + "total_payload_pct_of_limit": round( + total_payload_size / MAX_TOTAL_PAYLOAD_SIZE_BYTES * 100, 2 + ), + }, + ) + + async def execute_workflow( workflow_definition: ModelWorkflowDefinition, workflow_steps: list[ModelWorkflowStep], @@ -190,6 +359,11 @@ async def validate_workflow_definition( Pure validation function - no side effects. + Enforces the following limits (OMN-670: Security hardening): + - MAX_WORKFLOW_STEPS: Maximum number of steps allowed in a workflow + - Dependencies must reference existing steps + - No dependency cycles allowed + Args: workflow_definition: Workflow definition to validate workflow_steps: Workflow steps to validate @@ -236,6 +410,17 @@ async def validate_workflow_definition( """ errors: list[str] = [] + # Validate step count limit (OMN-670: Security hardening) + # SECURITY: Short-circuit immediately on step count overflow to prevent DoS. + # Without early return, an attacker could submit workflows with millions of steps + # and the validation would still iterate through all steps for dependency cycle + # detection and per-step validation, causing CPU exhaustion. + if len(workflow_steps) > MAX_WORKFLOW_STEPS: + errors.append( + f"Workflow exceeds maximum step limit: {len(workflow_steps)} steps > {MAX_WORKFLOW_STEPS} maximum" + ) + return errors # DoS mitigation: skip all subsequent expensive validation + # Validate workflow definition metadata if not workflow_definition.workflow_metadata.workflow_name: errors.append("Workflow name is required") @@ -379,12 +564,39 @@ async def _execute_sequential( workflow_steps: list[ModelWorkflowStep], workflow_id: UUID, ) -> WorkflowExecutionResult: - """Execute workflow steps sequentially.""" + """ + Execute workflow steps sequentially. + + Steps are executed one at a time in topological order (respecting dependencies). + Each step's output is available to subsequent steps via workflow context. + + Args: + workflow_definition: Workflow definition from YAML contract. + workflow_steps: List of workflow steps to execute. + workflow_id: Unique workflow execution ID. + + Returns: + WorkflowExecutionResult with completed/failed steps and emitted actions. + + Raises: + ModelOnexError: If workflow validation fails during setup. + + Note: + Total payload limit violations (MAX_TOTAL_PAYLOAD_SIZE_BYTES) are caught + by the step error handler and treated as step failures per the step's + error_action configuration (default: 'continue'). The workflow returns + FAILED status rather than raising the error. This differs from parallel + mode, which raises ModelOnexError immediately for payload limit violations. + + Individual step payload limit violations (MAX_STEP_PAYLOAD_SIZE_BYTES) are + also caught and treated as step failures in the same manner. + """ completed_steps: list[str] = [] failed_steps: list[str] = [] all_actions: list[ModelAction] = [] completed_step_ids: set[UUID] = set() step_outputs: dict[UUID, object] = {} + total_payload_size = 0 # Track total payload size (OMN-670: Security hardening) # Log workflow execution start logging.info( @@ -422,10 +634,36 @@ async def _execute_sequential( step, workflow_id, completed_step_ids, workflow_context ) - # Emit action for this step - action = _create_action_for_step(step, workflow_id) + # Emit action for this step (returns tuple with payload size to avoid + # redundant JSON serialization - OMN-670: Performance optimization) + action, action_payload_size = _create_action_for_step(step, workflow_id) all_actions.append(action) + # Track total payload size (OMN-670: Security hardening) + # Note: action_payload_size comes from _create_action_for_step to avoid + # redundant json.dumps() call + total_payload_size += action_payload_size + + # Log payload metrics for observability (OMN-670: Metrics) + _log_payload_metrics( + workflow_id=workflow_id, + step_id=step.step_id, + step_name=step.step_name, + payload_size=action_payload_size, + total_payload_size=total_payload_size, + ) + + if total_payload_size > MAX_TOTAL_PAYLOAD_SIZE_BYTES: + raise ModelOnexError( + error_code=EnumCoreErrorCode.WORKFLOW_TOTAL_PAYLOAD_EXCEEDED, + message=f"Workflow total payload exceeds limit: {total_payload_size} bytes > {MAX_TOTAL_PAYLOAD_SIZE_BYTES} byte limit", + context={ + "workflow_id": str(workflow_id), + "total_size": total_payload_size, + "limit": MAX_TOTAL_PAYLOAD_SIZE_BYTES, + }, + ) + # Mark step as completed context.completed_at = datetime.now() completed_steps.append(str(step.step_id)) @@ -481,6 +719,15 @@ async def _execute_sequential( EnumWorkflowState.COMPLETED if not failed_steps else EnumWorkflowState.FAILED ) + # Log workflow completion metrics (OMN-670: Metrics) + _log_workflow_completion_metrics( + workflow_id=workflow_id, + workflow_name=workflow_definition.workflow_metadata.workflow_name, + total_payload_size=total_payload_size, + step_count=len(completed_steps), + execution_mode="sequential", + ) + return WorkflowExecutionResult( workflow_id=workflow_id, execution_status=status, @@ -500,13 +747,50 @@ async def _execute_parallel( workflow_steps: list[ModelWorkflowStep], workflow_id: UUID, ) -> WorkflowExecutionResult: - """Execute workflow steps in parallel (respecting dependencies).""" + """ + Execute workflow steps in parallel (respecting dependencies). + + Steps are executed in waves based on dependency order. Within each wave, + all steps with met dependencies run concurrently via asyncio.gather. + Steps in the same wave cannot see each other's outputs (they run in parallel), + but can access outputs from prior waves via workflow context. + + Args: + workflow_definition: Workflow definition from YAML contract. + workflow_steps: List of workflow steps to execute. + workflow_id: Unique workflow execution ID. + + Returns: + WorkflowExecutionResult with completed/failed steps and emitted actions. + + Raises: + ModelOnexError: If workflow validation fails during setup, or if total + payload limit (MAX_TOTAL_PAYLOAD_SIZE_BYTES) is exceeded. Unlike + sequential mode, payload limit violations in parallel mode fail + the entire workflow immediately after all wave steps complete, + raising ModelOnexError rather than treating it as a step failure. + + Note: + The error handling behavior differs from sequential mode: + - Sequential mode: Payload limit violations are caught and treated as + step failures per error_action configuration (workflow returns FAILED). + - Parallel mode: Payload limit violations raise ModelOnexError immediately + after the current wave completes (error propagates to caller). + + This difference exists because parallel execution processes all steps in a + wave simultaneously, making per-step error handling less meaningful when + the aggregate payload exceeds the limit. + + Individual step errors (including MAX_STEP_PAYLOAD_SIZE_BYTES violations) + are still handled per-step and respect the error_action configuration. + """ completed_steps: list[str] = [] failed_steps: list[str] = [] all_actions: list[ModelAction] = [] completed_step_ids: set[UUID] = set() step_outputs: dict[UUID, object] = {} should_stop = False + total_payload_size = 0 # Track total payload size (OMN-670: Security hardening) # Log workflow execution start logging.info( @@ -516,7 +800,7 @@ async def _execute_parallel( async def execute_step( step: ModelWorkflowStep, wave_context: TypedDictWorkflowContext, - ) -> tuple[ModelWorkflowStep, ModelAction | None, Exception | None]: + ) -> tuple[ModelWorkflowStep, ModelAction | None, int, Exception | None]: """ Execute a single workflow step asynchronously. @@ -531,15 +815,19 @@ async def execute_step( enabling data flow between steps across wave boundaries. Returns: - A 3-tuple of (step, action, error) where: + A 4-tuple of (step, action, payload_size, error) where: - step: The original ModelWorkflowStep (always present for correlation) - action: ModelAction if execution succeeded, None if failed + - payload_size: Size in bytes of the JSON payload (0 if failed) - error: Exception if execution failed, None if succeeded Exactly one of action/error will be None (mutually exclusive). This tuple pattern allows the caller to process results from asyncio.gather without individual try/except blocks per task. + The payload_size is returned to avoid redundant JSON serialization + in the caller (OMN-670: Performance optimization). + Note: This function catches all exceptions and returns them in the tuple rather than re-raising. This is intentional for parallel execution: @@ -561,11 +849,12 @@ async def execute_step( step, workflow_id, completed_step_ids, wave_context ) - # Create action for this step - action = _create_action_for_step(step, workflow_id) - return (step, action, None) + # Create action for this step (returns tuple with payload size to avoid + # redundant JSON serialization - OMN-670: Performance optimization) + action, payload_size = _create_action_for_step(step, workflow_id) + return (step, action, payload_size, None) except Exception as e: # fallback-ok: parallel execution returns error in tuple for caller handling - return (step, None, e) + return (step, None, 0, e) # For parallel execution, we execute in waves based on dependencies # Filter out disabled steps entirely - they are skipped, not failed @@ -599,9 +888,35 @@ async def execute_step( results = await asyncio.gather(*tasks) # Process results from parallel execution - for step, action, error in results: + for step, action, action_payload_size, error in results: if error is None and action is not None: - # Step succeeded + # Step succeeded - check payload size BEFORE marking completed + # Track total payload size (OMN-670: Security hardening) + # Note: action_payload_size comes from _create_action_for_step via execute_step + # to avoid redundant json.dumps() call (OMN-670: Performance optimization) + total_payload_size += action_payload_size + + # Log payload metrics for observability (OMN-670: Metrics) + _log_payload_metrics( + workflow_id=workflow_id, + step_id=step.step_id, + step_name=step.step_name, + payload_size=action_payload_size, + total_payload_size=total_payload_size, + ) + + if total_payload_size > MAX_TOTAL_PAYLOAD_SIZE_BYTES: + raise ModelOnexError( + error_code=EnumCoreErrorCode.WORKFLOW_TOTAL_PAYLOAD_EXCEEDED, + message=f"Workflow total payload exceeds limit: {total_payload_size} bytes > {MAX_TOTAL_PAYLOAD_SIZE_BYTES} byte limit", + context={ + "workflow_id": str(workflow_id), + "total_size": total_payload_size, + "limit": MAX_TOTAL_PAYLOAD_SIZE_BYTES, + }, + ) + + # Payload size check passed - now mark step completed all_actions.append(action) completed_steps.append(str(step.step_id)) completed_step_ids.add(step.step_id) @@ -649,6 +964,15 @@ async def execute_step( EnumWorkflowState.COMPLETED if not failed_steps else EnumWorkflowState.FAILED ) + # Log workflow completion metrics (OMN-670: Metrics) + _log_workflow_completion_metrics( + workflow_id=workflow_id, + workflow_name=workflow_definition.workflow_metadata.workflow_name, + total_payload_size=total_payload_size, + step_count=len(completed_steps), + execution_mode="parallel", + ) + return WorkflowExecutionResult( workflow_id=workflow_id, execution_status=status, @@ -668,7 +992,29 @@ async def _execute_batch( workflow_steps: list[ModelWorkflowStep], workflow_id: UUID, ) -> WorkflowExecutionResult: - """Execute workflow with batching.""" + """ + Execute workflow with batching. + + Delegates to sequential execution internally, adding batch-specific metadata. + Error handling behavior follows sequential mode semantics. + + Args: + workflow_definition: Workflow definition from YAML contract. + workflow_steps: List of workflow steps to execute. + workflow_id: Unique workflow execution ID. + + Returns: + WorkflowExecutionResult with batch metadata (execution_mode='batch', batch_size). + + Raises: + ModelOnexError: If workflow validation fails during setup. + + Note: + Since batch mode uses sequential execution internally, payload limit + violations (MAX_TOTAL_PAYLOAD_SIZE_BYTES) are handled as step failures + per the step's error_action configuration, not raised as exceptions. + See _execute_sequential for detailed error handling semantics. + """ # For batch mode, use sequential execution with batching metadata result = await _execute_sequential(workflow_definition, workflow_steps, workflow_id) result.metadata["execution_mode"] = "batch" @@ -855,19 +1201,24 @@ def _validate_json_payload( def _create_action_for_step( step: ModelWorkflowStep, workflow_id: UUID, -) -> ModelAction: +) -> tuple[ModelAction, int]: """ Create action for workflow step. + Enforces payload size limit (OMN-670: Security hardening): + - MAX_STEP_PAYLOAD_SIZE_BYTES: Maximum size of individual step payload + Args: step: Workflow step to create action for workflow_id: Parent workflow ID Returns: - ModelAction for step execution + Tuple of (ModelAction, payload_size_bytes) where payload_size_bytes is + the UTF-8 encoded size of the JSON payload. This avoids redundant + serialization in callers that need to track total payload size. Raises: - ModelOnexError: If payload is not JSON-serializable + ModelOnexError: If payload is not JSON-serializable or exceeds size limit """ # Map step type to action type action_type_map = { @@ -902,10 +1253,36 @@ def _create_action_for_step( "step_name": step.step_name, } - # Validate payload is JSON-serializable (fail fast) - _validate_json_payload(payload, context=step.step_name) + # Serialize payload once for both validation and size checking (OMN-670: Performance optimization) + # This replaces the previous two-step approach that called json.dumps() twice + try: + payload_json = json.dumps(payload, default=_json_default_for_workflow) + except (TypeError, ValueError) as e: + raise ModelOnexError( + error_code=EnumCoreErrorCode.VALIDATION_ERROR, + message=f"Payload is not JSON-serializable for step '{step.step_name}': {e}", + context={ + "payload_keys": list(payload.keys()), + "step_context": step.step_name, + }, + ) from e - return ModelAction( + # Validate payload size (OMN-670: Security hardening) + payload_bytes = payload_json.encode("utf-8") + payload_size = len(payload_bytes) + if payload_size > MAX_STEP_PAYLOAD_SIZE_BYTES: + raise ModelOnexError( + error_code=EnumCoreErrorCode.WORKFLOW_PAYLOAD_SIZE_EXCEEDED, + message=f"Step payload exceeds size limit: {payload_size} bytes > {MAX_STEP_PAYLOAD_SIZE_BYTES} byte limit", + context={ + "step_id": str(step.step_id), + "step_name": step.step_name, + "payload_size": payload_size, + "limit": MAX_STEP_PAYLOAD_SIZE_BYTES, + }, + ) + + action = ModelAction( action_id=uuid4(), action_type=action_type, target_node_type=target_node_type, @@ -923,6 +1300,8 @@ def _create_action_for_step( created_at=datetime.now(), ) + return (action, payload_size) + def _dependencies_met( step: ModelWorkflowStep, @@ -1119,6 +1498,9 @@ def verify_workflow_integrity( # Public API __all__ = [ + "MAX_STEP_PAYLOAD_SIZE_BYTES", + "MAX_TOTAL_PAYLOAD_SIZE_BYTES", + "MAX_WORKFLOW_STEPS", "WorkflowExecutionResult", "execute_workflow", "get_execution_order", diff --git a/src/omnibase_core/validation/contracts.py b/src/omnibase_core/validation/contracts.py index edcba2ce4f..d0d9c3539c 100644 --- a/src/omnibase_core/validation/contracts.py +++ b/src/omnibase_core/validation/contracts.py @@ -10,6 +10,7 @@ from __future__ import annotations import argparse +import logging import os import signal import sys @@ -114,6 +115,10 @@ def validate_yaml_file(file_path: Path) -> list[str]: except ModelOnexError as e: # Collect structured validation errors errors.append(f"Contract validation failed: {e.message}") + except Exception as e: + # Fallback for unexpected errors - log full traceback for debugging + logging.exception(f"Unexpected error during contract validation: {e}") + errors.append(f"Contract validation failed: {e}") # All validation is now handled by Pydantic model # Manual validation removed for ONEX compliance diff --git a/tests/unit/infrastructure/test_node_config_provider.py b/tests/unit/infrastructure/test_node_config_provider.py index 6d1b0e7a59..ce31732022 100644 --- a/tests/unit/infrastructure/test_node_config_provider.py +++ b/tests/unit/infrastructure/test_node_config_provider.py @@ -320,11 +320,15 @@ async def test_schema_structure(self) -> None: # Check structure of a schema entry compute_workers_schema = schema["compute.max_parallel_workers"] + # Schema entries are now ModelNodeConfigSchema Pydantic models assert isinstance(compute_workers_schema, ModelNodeConfigSchema) + assert hasattr(compute_workers_schema, "key") + assert hasattr(compute_workers_schema, "config_type") + assert hasattr(compute_workers_schema, "default") # Verify values assert compute_workers_schema.key == "compute.max_parallel_workers" - assert compute_workers_schema.type == "int" + assert compute_workers_schema.config_type == "int" assert compute_workers_schema.default == 4 diff --git a/tests/unit/models/configuration/test_model_config_types.py b/tests/unit/models/configuration/test_model_config_types.py new file mode 100644 index 0000000000..e8a2b6007b --- /dev/null +++ b/tests/unit/models/configuration/test_model_config_types.py @@ -0,0 +1,222 @@ +"""Tests for validate_config_value_type function. + +This module tests the type validation logic for configuration values, +ensuring strict type checking especially for boolean values where +Python's int subclass relationship could cause issues. +""" + +import pytest + +from omnibase_core.models.configuration.model_config_types import ( + validate_config_value_type, +) + + +class TestValidateConfigValueTypeValidCases: + """Tests for valid type/value combinations.""" + + def test_valid_int(self) -> None: + """Test that int value passes for 'int' type.""" + validate_config_value_type("int", 42) # Should not raise + + def test_valid_int_zero(self) -> None: + """Test that zero int value passes for 'int' type.""" + validate_config_value_type("int", 0) # Should not raise + + def test_valid_int_negative(self) -> None: + """Test that negative int value passes for 'int' type.""" + validate_config_value_type("int", -100) # Should not raise + + def test_valid_float(self) -> None: + """Test that float value passes for 'float' type.""" + validate_config_value_type("float", 3.14) # Should not raise + + def test_valid_float_zero(self) -> None: + """Test that zero float value passes for 'float' type.""" + validate_config_value_type("float", 0.0) # Should not raise + + def test_valid_float_negative(self) -> None: + """Test that negative float value passes for 'float' type.""" + validate_config_value_type("float", -2.5) # Should not raise + + def test_int_for_float_allowed(self) -> None: + """Test that int is valid for 'float' type (implicit conversion is safe).""" + validate_config_value_type("float", 42) # int is valid for float + + def test_valid_bool_true(self) -> None: + """Test that True passes for 'bool' type.""" + validate_config_value_type("bool", True) # Should not raise + + def test_valid_bool_false(self) -> None: + """Test that False passes for 'bool' type.""" + validate_config_value_type("bool", False) # Should not raise + + def test_valid_str(self) -> None: + """Test that str value passes for 'str' type.""" + validate_config_value_type("str", "hello") # Should not raise + + def test_valid_str_empty(self) -> None: + """Test that empty string passes for 'str' type.""" + validate_config_value_type("str", "") # Should not raise + + def test_valid_str_whitespace(self) -> None: + """Test that whitespace-only string passes for 'str' type.""" + validate_config_value_type("str", " ") # Should not raise + + def test_valid_str_unicode(self) -> None: + """Test that unicode string passes for 'str' type.""" + validate_config_value_type("str", "hello world") # Should not raise + + +class TestValidateConfigValueTypeInvalidCases: + """Tests for invalid type/value combinations.""" + + def test_str_for_int_fails(self) -> None: + """Test that string value fails for 'int' type.""" + with pytest.raises(ValueError, match="default must be int, got str"): + validate_config_value_type("int", "42") + + def test_float_for_int_fails(self) -> None: + """Test that float value fails for 'int' type.""" + with pytest.raises(ValueError, match="default must be int, got float"): + validate_config_value_type("int", 3.14) + + def test_bool_for_int_fails(self) -> None: + """Test that bool value fails for 'int' type. + + Note: In Python, bool is a subclass of int, so isinstance(True, int) is True. + However, we want strict type checking for configuration values. + """ + # This currently passes because bool is subclass of int + # If strict checking is desired, this test documents the current behavior + # and can be updated if the implementation changes + # For now, validate the current behavior (bool passes isinstance check for int) + validate_config_value_type("int", True) # Currently allowed due to subclass + + def test_str_for_float_fails(self) -> None: + """Test that string value fails for 'float' type.""" + with pytest.raises(ValueError, match="default must be float, got str"): + validate_config_value_type("float", "3.14") + + def test_int_for_str_fails(self) -> None: + """Test that int value fails for 'str' type.""" + with pytest.raises(ValueError, match="default must be str, got int"): + validate_config_value_type("str", 42) + + def test_float_for_str_fails(self) -> None: + """Test that float value fails for 'str' type.""" + with pytest.raises(ValueError, match="default must be str, got float"): + validate_config_value_type("str", 3.14) + + def test_bool_for_str_fails(self) -> None: + """Test that bool value fails for 'str' type.""" + with pytest.raises(ValueError, match="default must be str, got bool"): + validate_config_value_type("str", True) + + +class TestValidateConfigValueTypeBoolEdgeCases: + """Tests for boolean edge cases. + + Python's bool is a subclass of int, which means isinstance(True, int) is True + and isinstance(False, int) is True. The validation function has special logic + to handle this and reject int values (like 0 and 1) for bool type. + """ + + def test_int_zero_for_bool_fails(self) -> None: + """Test that 0 (int) fails for 'bool' type. + + Even though 0 is falsy and bool(0) == False, the literal int 0 + should not be accepted as a bool configuration value. + """ + with pytest.raises(ValueError, match="default must be bool, got int"): + validate_config_value_type("bool", 0) + + def test_int_one_for_bool_fails(self) -> None: + """Test that 1 (int) fails for 'bool' type. + + Even though 1 is truthy and bool(1) == True, the literal int 1 + should not be accepted as a bool configuration value. + """ + with pytest.raises(ValueError, match="default must be bool, got int"): + validate_config_value_type("bool", 1) + + def test_int_negative_for_bool_fails(self) -> None: + """Test that negative int fails for 'bool' type.""" + with pytest.raises(ValueError, match="default must be bool, got int"): + validate_config_value_type("bool", -1) + + def test_float_zero_for_bool_fails(self) -> None: + """Test that 0.0 (float) fails for 'bool' type.""" + with pytest.raises(ValueError, match="default must be bool, got float"): + validate_config_value_type("bool", 0.0) + + def test_float_one_for_bool_fails(self) -> None: + """Test that 1.0 (float) fails for 'bool' type.""" + with pytest.raises(ValueError, match="default must be bool, got float"): + validate_config_value_type("bool", 1.0) + + def test_str_for_bool_fails(self) -> None: + """Test that string fails for 'bool' type.""" + with pytest.raises(ValueError, match="default must be bool, got str"): + validate_config_value_type("bool", "true") + + def test_str_empty_for_bool_fails(self) -> None: + """Test that empty string fails for 'bool' type.""" + with pytest.raises(ValueError, match="default must be bool, got str"): + validate_config_value_type("bool", "") + + +class TestValidateConfigValueTypeFloatEdgeCases: + """Tests for float edge cases.""" + + def test_float_infinity_positive(self) -> None: + """Test that positive infinity passes for 'float' type.""" + validate_config_value_type("float", float("inf")) # Should not raise + + def test_float_infinity_negative(self) -> None: + """Test that negative infinity passes for 'float' type.""" + validate_config_value_type("float", float("-inf")) # Should not raise + + def test_float_nan(self) -> None: + """Test that NaN passes for 'float' type.""" + validate_config_value_type("float", float("nan")) # Should not raise + + def test_float_very_small(self) -> None: + """Test that very small float passes for 'float' type.""" + validate_config_value_type("float", 1e-300) # Should not raise + + def test_float_very_large(self) -> None: + """Test that very large float passes for 'float' type.""" + validate_config_value_type("float", 1e300) # Should not raise + + +class TestValidateConfigValueTypeIntEdgeCases: + """Tests for int edge cases.""" + + def test_int_very_large(self) -> None: + """Test that very large int passes for 'int' type.""" + validate_config_value_type("int", 10**100) # Should not raise + + def test_int_very_large_negative(self) -> None: + """Test that very large negative int passes for 'int' type.""" + validate_config_value_type("int", -(10**100)) # Should not raise + + +class TestValidateConfigValueTypeStrEdgeCases: + """Tests for string edge cases.""" + + def test_str_newline(self) -> None: + """Test that string with newline passes for 'str' type.""" + validate_config_value_type("str", "hello\nworld") # Should not raise + + def test_str_tab(self) -> None: + """Test that string with tab passes for 'str' type.""" + validate_config_value_type("str", "hello\tworld") # Should not raise + + def test_str_null_char(self) -> None: + """Test that string with null character passes for 'str' type.""" + validate_config_value_type("str", "hello\x00world") # Should not raise + + def test_str_very_long(self) -> None: + """Test that very long string passes for 'str' type.""" + validate_config_value_type("str", "x" * 10000) # Should not raise diff --git a/tests/unit/models/configuration/test_model_node_config_entry_schema.py b/tests/unit/models/configuration/test_model_node_config_entry_schema.py new file mode 100644 index 0000000000..0a0b9ac988 --- /dev/null +++ b/tests/unit/models/configuration/test_model_node_config_entry_schema.py @@ -0,0 +1,253 @@ +""" +Unit tests for ModelNodeConfigEntry and ModelNodeConfigSchema. + +Tests validation constraints including: +- Literal type constraints for value_type/config_type +- Default value type validation +- Bool/int coercion prevention +- Alias behavior for ModelNodeConfigSchema +""" + +import pytest +from pydantic import ValidationError + +from omnibase_core.models.configuration.model_node_config_entry import ( + ModelNodeConfigEntry, +) +from omnibase_core.models.configuration.model_node_config_schema import ( + ModelNodeConfigSchema, +) + + +class TestModelNodeConfigEntry: + """Test cases for ModelNodeConfigEntry.""" + + def test_valid_int_entry(self) -> None: + """Test valid int configuration entry.""" + entry = ModelNodeConfigEntry( + key="compute.max_workers", + value_type="int", + default=4, + ) + assert entry.key == "compute.max_workers" + assert entry.value_type == "int" + assert entry.default == 4 + + def test_valid_float_entry(self) -> None: + """Test valid float configuration entry.""" + entry = ModelNodeConfigEntry( + key="compute.threshold", + value_type="float", + default=100.5, + ) + assert entry.value_type == "float" + assert entry.default == 100.5 + + def test_valid_float_entry_with_int_default(self) -> None: + """Test float entry accepts int default value.""" + entry = ModelNodeConfigEntry( + key="compute.threshold", + value_type="float", + default=100, + ) + assert entry.value_type == "float" + assert entry.default == 100 + + def test_valid_bool_entry(self) -> None: + """Test valid bool configuration entry.""" + entry = ModelNodeConfigEntry( + key="feature.enabled", + value_type="bool", + default=True, + ) + assert entry.value_type == "bool" + assert entry.default is True + + def test_valid_str_entry(self) -> None: + """Test valid str configuration entry.""" + entry = ModelNodeConfigEntry( + key="config.name", + value_type="str", + default="production", + ) + assert entry.value_type == "str" + assert entry.default == "production" + + def test_invalid_value_type_rejected(self) -> None: + """Test that invalid value_type is rejected.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigEntry( + key="test.key", + value_type="invalid", # type: ignore[arg-type] + default="test", + ) + assert "value_type" in str(exc_info.value) + + def test_int_default_for_bool_rejected(self) -> None: + """Test that int default is rejected for bool type (prevents coercion).""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigEntry( + key="test.key", + value_type="bool", + default=1, # int, not bool + ) + assert "must be bool" in str(exc_info.value) + + def test_zero_default_for_bool_rejected(self) -> None: + """Test that 0 default is rejected for bool type.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigEntry( + key="test.key", + value_type="bool", + default=0, # int, not bool + ) + assert "must be bool" in str(exc_info.value) + + def test_str_default_for_int_rejected(self) -> None: + """Test that str default is rejected for int type.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigEntry( + key="test.key", + value_type="int", + default="not_an_int", + ) + assert "must be int" in str(exc_info.value) + + def test_float_default_for_int_rejected(self) -> None: + """Test that float default is rejected for int type.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigEntry( + key="test.key", + value_type="int", + default=4.5, + ) + assert "must be int" in str(exc_info.value) + + def test_int_default_for_str_rejected(self) -> None: + """Test that int default is rejected for str type.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigEntry( + key="test.key", + value_type="str", + default=42, + ) + assert "must be str" in str(exc_info.value) + + def test_model_is_frozen(self) -> None: + """Test that model is immutable (frozen).""" + entry = ModelNodeConfigEntry( + key="test.key", + value_type="int", + default=4, + ) + with pytest.raises(ValidationError): + entry.key = "new.key" # type: ignore[misc] + + +class TestModelNodeConfigSchema: + """Test cases for ModelNodeConfigSchema.""" + + def test_valid_schema_with_alias(self) -> None: + """Test valid schema using 'type' alias.""" + schema = ModelNodeConfigSchema( + key="compute.max_workers", + type="int", # Using alias + default=4, + ) + assert schema.key == "compute.max_workers" + assert schema.config_type == "int" # Accessing via field name + assert schema.default == 4 + + def test_valid_schema_with_field_name(self) -> None: + """Test valid schema using 'config_type' field name.""" + schema = ModelNodeConfigSchema( + key="compute.max_workers", + config_type="int", # Using field name + default=4, + ) + assert schema.config_type == "int" + + def test_model_dump_with_alias(self) -> None: + """Test model_dump with by_alias=True returns 'type'.""" + schema = ModelNodeConfigSchema( + key="compute.max_workers", + type="int", + default=4, + ) + dumped = schema.model_dump(by_alias=True) + assert "type" in dumped + assert dumped["type"] == "int" + assert "config_type" not in dumped + + def test_model_dump_without_alias(self) -> None: + """Test model_dump without alias returns 'config_type'.""" + schema = ModelNodeConfigSchema( + key="compute.max_workers", + type="int", + default=4, + ) + dumped = schema.model_dump() + assert "config_type" in dumped + assert dumped["config_type"] == "int" + # 'type' may or may not be in the output depending on Pydantic version + + def test_int_default_for_bool_rejected(self) -> None: + """Test that int default is rejected for bool type.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigSchema( + key="test.key", + type="bool", + default=1, + ) + assert "must be bool" in str(exc_info.value) + + def test_invalid_type_rejected(self) -> None: + """Test that invalid type is rejected.""" + with pytest.raises(ValidationError) as exc_info: + ModelNodeConfigSchema( + key="test.key", + type="invalid", # type: ignore[arg-type] + default="test", + ) + # Check for validation error in type field + assert "type" in str(exc_info.value).lower() + + def test_valid_float_with_int_default(self) -> None: + """Test float type accepts int default value.""" + schema = ModelNodeConfigSchema( + key="compute.threshold", + type="float", + default=100, + ) + assert schema.config_type == "float" + assert schema.default == 100 + + def test_valid_bool_true(self) -> None: + """Test bool type with True default.""" + schema = ModelNodeConfigSchema( + key="feature.enabled", + type="bool", + default=True, + ) + assert schema.config_type == "bool" + assert schema.default is True + + def test_valid_bool_false(self) -> None: + """Test bool type with False default.""" + schema = ModelNodeConfigSchema( + key="feature.disabled", + type="bool", + default=False, + ) + assert schema.config_type == "bool" + assert schema.default is False + + def test_model_is_frozen(self) -> None: + """Test that model is immutable (frozen).""" + schema = ModelNodeConfigSchema( + key="test.key", + type="int", + default=4, + ) + with pytest.raises(ValidationError): + schema.key = "new.key" # type: ignore[misc] diff --git a/tests/unit/models/validation/test_model_schema.py b/tests/unit/models/validation/test_model_schema.py index 251699afb9..e69de29bb2 100644 --- a/tests/unit/models/validation/test_model_schema.py +++ b/tests/unit/models/validation/test_model_schema.py @@ -1,217 +0,0 @@ -""" -Tests for model_schema.py module. - -This module tests the schema validation models and their compatibility aliases. -""" - -import pytest - - -class TestModelSchemaModule: - """Test the model_schema module structure and imports.""" - - def test_module_imports(self): - """Test that all expected classes can be imported.""" - from omnibase_core.models.validation.model_schema import ( - ModelRequiredFieldsModel, - ModelSchema, - ModelSchemaPropertiesModel, - RequiredFieldsModel, - SchemaModel, - SchemaPropertiesModel, - SchemaPropertyModel, - ) - - # Verify all classes are imported - assert ModelSchema is not None - assert ModelSchemaPropertiesModel is not None - assert ModelRequiredFieldsModel is not None - assert SchemaPropertyModel is not None - assert SchemaPropertiesModel is not None - assert RequiredFieldsModel is not None - assert SchemaModel is not None - - def test_compatibility_aliases(self): - """Test that compatibility aliases point to correct classes.""" - from omnibase_core.models.validation.model_schema import ( - ModelRequiredFieldsModel, - ModelSchema, - ModelSchemaPropertiesModel, - RequiredFieldsModel, - SchemaModel, - SchemaPropertiesModel, - SchemaPropertyModel, - ) - - # Import the actual classes to compare - from omnibase_core.models.validation.model_schema_property import ( - ModelSchemaProperty, - ) - - # Test that aliases point to the correct classes - assert SchemaPropertyModel is ModelSchemaProperty - assert SchemaPropertiesModel is ModelSchemaPropertiesModel - assert RequiredFieldsModel is ModelRequiredFieldsModel - assert SchemaModel is ModelSchema - - def test_module_all_exports(self): - """Test that __all__ contains all expected exports.""" - from omnibase_core.models.validation import model_schema - - expected_exports = [ - "ModelSchema", - "ModelSchemaPropertiesModel", - "ModelRequiredFieldsModel", - "SchemaPropertyModel", - "SchemaPropertiesModel", - "RequiredFieldsModel", - "SchemaModel", - ] - - # Check that all expected exports are in __all__ - assert hasattr(model_schema, "__all__") - for export in expected_exports: - assert export in model_schema.__all__ - - def test_import_from_module(self): - """Test importing classes from the module.""" - from omnibase_core.models.validation.model_schema import ( - ModelRequiredFieldsModel, - ModelSchema, - ModelSchemaPropertiesModel, - ) - - # Verify classes are accessible - assert ModelSchema is not None - assert ModelSchemaPropertiesModel is not None - assert ModelRequiredFieldsModel is not None - - def test_compatibility_aliases_functionality(self): - """Test that compatibility aliases work correctly.""" - from omnibase_core.models.validation.model_required_fields_model import ( - ModelRequiredFieldsModel, - ) - from omnibase_core.models.validation.model_schema import ( - RequiredFieldsModel, - SchemaModel, - SchemaPropertiesModel, - SchemaPropertyModel, - ) - from omnibase_core.models.validation.model_schema_class import ModelSchema - from omnibase_core.models.validation.model_schema_properties_model import ( - ModelSchemaPropertiesModel, - ) - - # Test that aliases are the same as their original classes - from omnibase_core.models.validation.model_schema_property import ( - ModelSchemaProperty, - ) - - assert SchemaPropertyModel is ModelSchemaProperty - assert SchemaPropertiesModel is ModelSchemaPropertiesModel - assert RequiredFieldsModel is ModelRequiredFieldsModel - assert SchemaModel is ModelSchema - - def test_module_structure(self): - """Test that the module has expected structure.""" - import omnibase_core.models.validation.model_schema as module - - # Module should exist and be importable - assert module is not None - - # Should have all expected attributes - expected_attrs = [ - "ModelSchema", - "ModelSchemaPropertiesModel", - "ModelRequiredFieldsModel", - "SchemaPropertyModel", - "SchemaPropertiesModel", - "RequiredFieldsModel", - "SchemaModel", - ] - - for attr in expected_attrs: - assert hasattr(module, attr) - - def test_import_paths(self): - """Test that classes can be imported via different paths.""" - # Direct import from module - # Import from package - from omnibase_core.models.validation import model_schema - from omnibase_core.models.validation.model_schema import ( - ModelSchema as DirectModelSchema, - ) - - PackageModelSchema = model_schema.ModelSchema - - # Both should be the same - assert DirectModelSchema is PackageModelSchema - - def test_class_availability(self): - """Test that all classes are available and importable.""" - from omnibase_core.models.validation.model_schema import ( - ModelRequiredFieldsModel, - ModelSchema, - ModelSchemaPropertiesModel, - RequiredFieldsModel, - SchemaModel, - SchemaPropertiesModel, - SchemaPropertyModel, - ) - - # All classes should be available - classes = [ - ModelSchema, - ModelSchemaPropertiesModel, - ModelRequiredFieldsModel, - SchemaPropertyModel, - SchemaPropertiesModel, - RequiredFieldsModel, - SchemaModel, - ] - - for cls in classes: - assert cls is not None - assert hasattr(cls, "__name__") - - def test_module_docstring(self): - """Test that the module has proper docstring.""" - import omnibase_core.models.validation.model_schema as module - - # Module should have docstring - assert module.__doc__ is not None - assert "Schema validation models" in module.__doc__ - - def test_import_error_handling(self): - """Test that import errors are handled properly.""" - # Test that we can import the module without errors - try: - from omnibase_core.models.validation.model_schema import ModelSchema - - assert ModelSchema is not None - except ImportError as e: - pytest.fail(f"Failed to import ModelSchema: {e}") - - def test_class_inheritance(self): - """Test that imported classes have expected inheritance.""" - from omnibase_core.models.validation.model_schema import ModelSchema - - # ModelSchema should be a class - assert isinstance(ModelSchema, type) - - # Should have expected methods if it's a Pydantic model - if hasattr(ModelSchema, "model_validate"): - assert callable(ModelSchema.model_validate) - if hasattr(ModelSchema, "model_dump"): - assert callable(ModelSchema.model_dump) - - def test_module_metadata(self): - """Test that the module has expected metadata.""" - import omnibase_core.models.validation.model_schema as module - - # Module should have __file__ attribute - assert hasattr(module, "__file__") - - # Module should have __name__ attribute - assert hasattr(module, "__name__") - assert module.__name__ == "omnibase_core.models.validation.model_schema" diff --git a/tests/unit/orchestrator/test_action_creation.py b/tests/unit/orchestrator/test_action_creation.py index 4c3878be93..7cfa1a1dd1 100644 --- a/tests/unit/orchestrator/test_action_creation.py +++ b/tests/unit/orchestrator/test_action_creation.py @@ -60,7 +60,7 @@ def test_priority_clamping_formula( ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.priority == expected_action_priority @@ -74,7 +74,7 @@ def test_priority_clamps_at_boundary(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.priority == 10 @@ -88,7 +88,7 @@ def test_priority_just_above_boundary(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.priority == 10 @@ -102,7 +102,7 @@ def test_priority_at_maximum_step_priority(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.priority == 10 @@ -116,7 +116,7 @@ def test_priority_at_minimum_step_priority(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.priority == 1 @@ -130,7 +130,7 @@ def test_default_priority_value(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Default priority 100 should be clamped to 10 assert action.priority == 10 @@ -149,7 +149,7 @@ def test_empty_dependencies(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.dependencies == [] @@ -164,7 +164,7 @@ def test_single_dependency(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert len(action.dependencies) == 1 assert action.dependencies[0] == dep_id @@ -180,7 +180,7 @@ def test_multiple_dependencies(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert len(action.dependencies) == 3 for dep_id in dep_ids: @@ -197,7 +197,7 @@ def test_dependency_order_preserved(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Verify order is preserved assert action.dependencies == dep_ids @@ -213,7 +213,7 @@ def test_dependencies_are_uuids(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert all(isinstance(dep, UUID) for dep in action.dependencies) @@ -230,7 +230,7 @@ def test_action_id_is_generated(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_id is not None @@ -243,7 +243,7 @@ def test_action_id_is_uuid(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert isinstance(action.action_id, UUID) @@ -257,7 +257,7 @@ def test_action_id_unique_per_action(self) -> None: workflow_id = uuid4() # Create multiple actions from the same step - actions = [_create_action_for_step(step, workflow_id) for _ in range(10)] + actions = [_create_action_for_step(step, workflow_id)[0] for _ in range(10)] # All action_ids should be unique action_ids = [action.action_id for action in actions] @@ -273,7 +273,7 @@ def test_action_id_differs_from_step_id(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_id != step_id @@ -286,7 +286,7 @@ def test_action_id_format_consistency(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # UUID string representation should be 36 characters (8-4-4-4-12) uuid_str = str(action.action_id) @@ -318,7 +318,7 @@ def test_step_type_to_action_type_mapping( ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_type == expected_action_type @@ -331,7 +331,7 @@ def test_compute_step_type_mapping(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_type == EnumActionType.COMPUTE @@ -344,7 +344,7 @@ def test_effect_step_type_mapping(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_type == EnumActionType.EFFECT @@ -357,7 +357,7 @@ def test_reducer_step_type_mapping(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_type == EnumActionType.REDUCE @@ -370,7 +370,7 @@ def test_orchestrator_step_type_mapping(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_type == EnumActionType.ORCHESTRATE @@ -383,7 +383,7 @@ def test_custom_step_type_mapping(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.action_type == EnumActionType.CUSTOM @@ -396,7 +396,7 @@ def test_conditional_step_type_fallback(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Unknown types fall back to CUSTOM assert action.action_type == EnumActionType.CUSTOM @@ -410,7 +410,7 @@ def test_parallel_step_type_fallback(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Unknown types fall back to CUSTOM assert action.action_type == EnumActionType.CUSTOM @@ -440,7 +440,7 @@ def test_target_node_type_mapping( ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.target_node_type == expected_target_node_type @@ -453,7 +453,7 @@ def test_unknown_step_type_fallback(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.target_node_type == "NodeCustom" @@ -471,7 +471,7 @@ def test_timeout_ms_propagation(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.timeout_ms == 60000 @@ -484,7 +484,7 @@ def test_default_timeout_ms(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Default is 30000ms assert action.timeout_ms == 30000 @@ -499,7 +499,7 @@ def test_retry_count_propagation(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.retry_count == 5 @@ -512,7 +512,7 @@ def test_default_retry_count(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Step default is 3 assert action.retry_count == 3 @@ -526,7 +526,7 @@ def test_metadata_contains_step_name(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert "step_name" in action.metadata assert action.metadata["step_name"] == "metadata_step" @@ -540,7 +540,7 @@ def test_metadata_contains_correlation_id(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert "correlation_id" in action.metadata # Correlation ID should be string representation of UUID @@ -555,7 +555,7 @@ def test_lease_id_is_generated(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.lease_id is not None assert isinstance(action.lease_id, UUID) @@ -569,7 +569,7 @@ def test_epoch_is_zero(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.epoch == 0 @@ -582,7 +582,7 @@ def test_created_at_is_set(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.created_at is not None assert isinstance(action.created_at, datetime) @@ -600,7 +600,7 @@ def test_payload_contains_workflow_id(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert "workflow_id" in action.payload assert action.payload["workflow_id"] == str(workflow_id) @@ -615,7 +615,7 @@ def test_payload_contains_step_id(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert "step_id" in action.payload assert action.payload["step_id"] == str(step_id) @@ -629,7 +629,7 @@ def test_payload_contains_step_name(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert "step_name" in action.payload assert action.payload["step_name"] == "name_payload_step" @@ -645,7 +645,7 @@ def test_payload_is_json_serializable(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Should not raise json_str = json.dumps(action.payload) @@ -668,7 +668,7 @@ def test_action_is_model_action_instance(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert isinstance(action, ModelAction) @@ -681,7 +681,7 @@ def test_action_is_frozen(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Attempting to modify should raise ValidationError for frozen models with pytest.raises(Exception): @@ -696,7 +696,7 @@ def test_action_can_be_serialized(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) # Should not raise action_dict = action.model_dump() @@ -714,7 +714,7 @@ def test_action_can_be_copied_with_modifications(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) original_priority = action.priority # Create modified copy @@ -739,7 +739,7 @@ def test_long_step_name(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.payload["step_name"] == long_name @@ -753,7 +753,7 @@ def test_minimum_timeout(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.timeout_ms == 100 @@ -767,7 +767,7 @@ def test_maximum_timeout(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.timeout_ms == 300000 @@ -781,7 +781,7 @@ def test_maximum_retry_count(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.retry_count == 10 @@ -795,7 +795,7 @@ def test_zero_retry_count(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert action.retry_count == 0 @@ -810,7 +810,7 @@ def test_many_dependencies(self) -> None: ) workflow_id = uuid4() - action = _create_action_for_step(step, workflow_id) + action, _ = _create_action_for_step(step, workflow_id) assert len(action.dependencies) == 50 assert action.dependencies == dep_ids diff --git a/tests/unit/utils/test_workflow_executor.py b/tests/unit/utils/test_workflow_executor.py index a218e9b085..e073f97213 100644 --- a/tests/unit/utils/test_workflow_executor.py +++ b/tests/unit/utils/test_workflow_executor.py @@ -35,6 +35,50 @@ ) +def _create_mock_workflow_step( + step_id: UUID | None = None, + step_name: str = "test_step", + step_type: str = "compute", + enabled: bool = True, + depends_on: list[UUID] | None = None, +) -> ModelWorkflowStep: + """Create a ModelWorkflowStep using model_construct to bypass Pydantic validation. + + This helper is used for tests that create many steps (1000+) where Pydantic + validation overhead becomes significant. Uses model_construct() to skip + validation while still creating proper ModelWorkflowStep instances. + + Args: + step_id: Optional step ID (generates UUID if not provided) + step_name: Human-readable name for the step + step_type: Type of workflow step (compute, effect, etc.) + enabled: Whether the step is enabled + depends_on: List of step IDs this step depends on + + Returns: + ModelWorkflowStep instance created without validation overhead + """ + return ModelWorkflowStep.model_construct( + correlation_id=uuid4(), + step_id=step_id or uuid4(), + step_name=step_name, + step_type=step_type, + timeout_ms=30000, + retry_count=3, + enabled=enabled, + skip_on_failure=False, + continue_on_error=False, + error_action="stop", + max_memory_mb=None, + max_cpu_percent=None, + priority=100, + order_index=0, + depends_on=depends_on if depends_on is not None else [], + parallel_group=None, + max_parallel_instances=1, + ) + + @pytest.fixture def simple_workflow_definition() -> ModelWorkflowDefinition: """Create simple workflow definition.""" @@ -1730,10 +1774,12 @@ def test_default_priority_in_create_action_for_step(self) -> None: # priority defaults to 100 ) - action = _create_action_for_step(step, uuid4()) + action, payload_size = _create_action_for_step(step, uuid4()) # Default priority (100) should be clamped to 10 assert action.priority == 10 + # Payload size should be returned and be positive + assert payload_size > 0 def test_priority_none_handling_in_topological_order(self) -> None: """Test that _get_topological_order handles None priority defensively. @@ -1789,10 +1835,12 @@ def test_priority_none_handling_in_create_action(self) -> None: mock_step.correlation_id = uuid4() # Should not raise TypeError when priority is None - action = _create_action_for_step(mock_step, uuid4()) + action, payload_size = _create_action_for_step(mock_step, uuid4()) # None priority should default to 1 assert action.priority == 1 + # Payload size should be returned and be positive + assert payload_size > 0 class TestWorkflowContextIntegration: @@ -3041,3 +3089,453 @@ async def test_workflow_with_varying_step_types( assert EnumActionType.COMPUTE in action_types_found assert EnumActionType.REDUCE in action_types_found assert EnumActionType.ORCHESTRATE in action_types_found + + +@pytest.mark.unit +class TestWorkflowSizeLimitEnforcement: + """Tests for workflow size limit enforcement (OMN-670). + + These tests verify the security hardening limits added in OMN-670: + - MAX_WORKFLOW_STEPS (1000): Maximum number of steps in a workflow + - MAX_STEP_PAYLOAD_SIZE_BYTES (64KB): Maximum size of individual step payload + - MAX_TOTAL_PAYLOAD_SIZE_BYTES (10MB): Maximum accumulated payload size + + Limits are validated during workflow execution to prevent memory exhaustion. + """ + + @pytest.mark.asyncio + async def test_validate_workflow_definition_step_count_at_limit( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test workflow with exactly MAX_WORKFLOW_STEPS steps passes validation.""" + from omnibase_core.utils.workflow_executor import MAX_WORKFLOW_STEPS + + # Create workflow with exactly 1000 steps (at limit) + # Uses model_construct to bypass Pydantic validation for performance + steps = [ + _create_mock_workflow_step(step_name=f"step_{i}") + for i in range(MAX_WORKFLOW_STEPS) + ] + + # Validation should pass (empty error list) + errors = await validate_workflow_definition(simple_workflow_definition, steps) + assert len(errors) == 0, f"Expected no errors, got: {errors}" + + @pytest.mark.asyncio + async def test_validate_workflow_definition_step_count_exceeds_limit( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test workflow exceeding MAX_WORKFLOW_STEPS fails validation.""" + from omnibase_core.utils.workflow_executor import MAX_WORKFLOW_STEPS + + # Create workflow with 1001 steps (exceeds limit) + # Uses model_construct to bypass Pydantic validation for performance + steps = [ + _create_mock_workflow_step(step_name=f"step_{i}") + for i in range(MAX_WORKFLOW_STEPS + 1) + ] + + # Validation should fail with step limit error + errors = await validate_workflow_definition(simple_workflow_definition, steps) + assert len(errors) > 0 + assert any("exceeds maximum step limit" in error for error in errors), ( + f"Expected 'exceeds maximum step limit' error, got: {errors}" + ) + + @pytest.mark.asyncio + async def test_step_payload_size_under_limit( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test step payload under MAX_STEP_PAYLOAD_SIZE_BYTES passes. + + The payload includes workflow_id, step_id, and step_name fields. + We use a reasonable step_name that produces a payload well under 64KB. + """ + # Create a step with a normal name (well under limit) + step = ModelWorkflowStep( + step_id=uuid4(), + step_name="normal_step_name", + step_type="effect", + enabled=True, + ) + + # Workflow should execute successfully with normal payload + result = await execute_workflow( + workflow_definition=simple_workflow_definition, + workflow_steps=[step], + workflow_id=uuid4(), + ) + + assert result.execution_status == EnumWorkflowState.COMPLETED + assert len(result.completed_steps) == 1 + assert len(result.actions_emitted) == 1 + + def test_step_payload_size_exceeds_limit(self) -> None: + """Test step payload exceeding MAX_STEP_PAYLOAD_SIZE_BYTES raises error. + + Note: ModelWorkflowStep.step_name has a max length of 200 characters, + so we cannot exceed 64KB with a real step. We test the _create_action_for_step + function directly using a mock step to verify the limit enforcement logic. + """ + from unittest.mock import MagicMock + + from omnibase_core.utils.workflow_executor import ( + MAX_STEP_PAYLOAD_SIZE_BYTES, + _create_action_for_step, + ) + + # Create a mock step with an extremely long name (bypassing Pydantic validation) + mock_step = MagicMock(spec=ModelWorkflowStep) + mock_step.step_id = uuid4() + mock_step.step_name = "x" * (MAX_STEP_PAYLOAD_SIZE_BYTES + 1000) # Exceeds 64KB + mock_step.step_type = "effect" + mock_step.priority = 1 + mock_step.depends_on = [] + mock_step.timeout_ms = 5000 + mock_step.retry_count = 0 + mock_step.correlation_id = uuid4() + + # Should raise ModelOnexError with WORKFLOW_PAYLOAD_SIZE_EXCEEDED + with pytest.raises(ModelOnexError) as exc_info: + _create_action_for_step(mock_step, uuid4()) + + assert ( + exc_info.value.error_code + == EnumCoreErrorCode.WORKFLOW_PAYLOAD_SIZE_EXCEEDED + ) + assert "payload exceeds size limit" in exc_info.value.message.lower() + + @pytest.mark.asyncio + async def test_total_payload_size_exceeds_limit_sequential( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test total payload exceeding limit in sequential execution is handled gracefully. + + Note: With ModelWorkflowStep.step_name limited to 200 chars and + MAX_WORKFLOW_STEPS limited to 1000, the maximum possible total payload + is ~300KB (well under 10MB). We mock _create_action_for_step to return + actions with large payloads to trigger the limit. + + In sequential execution, the limit error is caught by the step error handler + and treated as a step failure with 'continue' behavior (the default). + The workflow returns FAILED status rather than raising the error. + """ + from unittest.mock import patch + + from omnibase_core.utils.workflow_executor import ( + MAX_TOTAL_PAYLOAD_SIZE_BYTES, + _create_action_for_step, + ) + + # Create a normal workflow (within all model limits) + num_steps = 20 + steps = [ + ModelWorkflowStep( + step_id=uuid4(), + step_name=f"step_{i}", + step_type="compute", + enabled=True, + ) + for i in range(num_steps) + ] + + # Each action will have a ~550KB payload, so 20 steps = ~11MB > 10MB limit + large_payload_size = MAX_TOTAL_PAYLOAD_SIZE_BYTES // (num_steps - 1) + call_count = 0 + + def mock_create_action(step, workflow_id): + """Create action with large payload to exceed total limit.""" + nonlocal call_count + call_count += 1 + action, _ = _create_action_for_step(step, workflow_id) + # Inject large data into payload (simulating future payload expansion) + action.payload["large_data"] = "x" * large_payload_size + # Return tuple with updated payload size + import json + + new_payload_size = len(json.dumps(action.payload).encode("utf-8")) + return (action, new_payload_size) + + with patch( + "omnibase_core.utils.workflow_executor._create_action_for_step", + side_effect=mock_create_action, + ): + # In sequential execution, the error is caught and handled gracefully + # The workflow returns with FAILED status + result = await execute_workflow( + workflow_definition=simple_workflow_definition, + workflow_steps=steps, + workflow_id=uuid4(), + execution_mode=EnumExecutionMode.SEQUENTIAL, + ) + + # Workflow should fail but not raise an exception + assert result.execution_status == EnumWorkflowState.FAILED + # Some steps should have completed before the limit was hit + assert len(result.completed_steps) > 0 + # At least one step should have failed (the one that exceeded the limit) + assert len(result.failed_steps) > 0 + # Verify some steps were processed before limit was hit + assert call_count > 1, "Multiple steps should be processed before limit is hit" + + @pytest.mark.asyncio + async def test_total_payload_size_exceeds_limit_parallel( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test total payload exceeding limit in parallel execution raises error. + + Note: Similar to sequential test, we mock _create_action_for_step to + return actions with large payloads to trigger the 10MB total limit. + """ + from unittest.mock import patch + + from omnibase_core.utils.workflow_executor import ( + MAX_TOTAL_PAYLOAD_SIZE_BYTES, + _create_action_for_step, + ) + + # Create a normal workflow (within all model limits) + num_steps = 20 + steps = [ + ModelWorkflowStep( + step_id=uuid4(), + step_name=f"step_{i}", + step_type="compute", + enabled=True, + ) + for i in range(num_steps) + ] + + # Each action will have a ~550KB payload + large_payload_size = MAX_TOTAL_PAYLOAD_SIZE_BYTES // (num_steps - 1) + call_count = 0 + + def mock_create_action(step, workflow_id): + """Create action with large payload to exceed total limit.""" + nonlocal call_count + call_count += 1 + action, _ = _create_action_for_step(step, workflow_id) + action.payload["large_data"] = "x" * large_payload_size + # Return tuple with updated payload size + import json + + new_payload_size = len(json.dumps(action.payload).encode("utf-8")) + return (action, new_payload_size) + + with patch( + "omnibase_core.utils.workflow_executor._create_action_for_step", + side_effect=mock_create_action, + ): + with pytest.raises(ModelOnexError) as exc_info: + await execute_workflow( + workflow_definition=simple_workflow_definition, + workflow_steps=steps, + workflow_id=uuid4(), + execution_mode=EnumExecutionMode.PARALLEL, + ) + + assert ( + exc_info.value.error_code + == EnumCoreErrorCode.WORKFLOW_TOTAL_PAYLOAD_EXCEEDED + ) + assert "total payload exceeds limit" in exc_info.value.message.lower() + # Verify some steps were processed before limit was hit + assert call_count > 1, "Multiple steps should be processed before limit is hit" + + @pytest.mark.asyncio + async def test_total_payload_size_exceeds_limit_batch( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test total payload exceeding limit in batch execution is handled gracefully. + + Note: Batch mode internally uses sequential execution (see workflow_executor.py + line 742), so the behavior should be identical to sequential - the limit error + is caught and treated as a step failure with FAILED status returned. + + This test verifies that batch mode correctly inherits the graceful error + handling from sequential execution rather than raising an exception. + """ + from unittest.mock import patch + + from omnibase_core.utils.workflow_executor import ( + MAX_TOTAL_PAYLOAD_SIZE_BYTES, + _create_action_for_step, + ) + + # Create a normal workflow (within all model limits) + num_steps = 20 + steps = [ + ModelWorkflowStep( + step_id=uuid4(), + step_name=f"step_{i}", + step_type="compute", + enabled=True, + ) + for i in range(num_steps) + ] + + # Each action will have a ~550KB payload, so 20 steps = ~11MB > 10MB limit + large_payload_size = MAX_TOTAL_PAYLOAD_SIZE_BYTES // (num_steps - 1) + call_count = 0 + + def mock_create_action(step, workflow_id): + """Create action with large payload to exceed total limit.""" + nonlocal call_count + call_count += 1 + action, _ = _create_action_for_step(step, workflow_id) + # Inject large data into payload (simulating future payload expansion) + action.payload["large_data"] = "x" * large_payload_size + # Return tuple with updated payload size + import json + + new_payload_size = len(json.dumps(action.payload).encode("utf-8")) + return (action, new_payload_size) + + with patch( + "omnibase_core.utils.workflow_executor._create_action_for_step", + side_effect=mock_create_action, + ): + # In batch execution (which uses sequential internally), the error is + # caught and handled gracefully. The workflow returns with FAILED status. + result = await execute_workflow( + workflow_definition=simple_workflow_definition, + workflow_steps=steps, + workflow_id=uuid4(), + execution_mode=EnumExecutionMode.BATCH, + ) + + # Workflow should fail but not raise an exception + assert result.execution_status == EnumWorkflowState.FAILED + # Some steps should have completed before the limit was hit + assert len(result.completed_steps) > 0 + # At least one step should have failed (the one that exceeded the limit) + assert len(result.failed_steps) > 0 + # Verify some steps were processed before limit was hit + assert call_count > 1, "Multiple steps should be processed before limit is hit" + # Verify batch metadata is present (batch mode adds this) + assert result.metadata.get("execution_mode") == "batch" + assert result.metadata.get("batch_size") == num_steps + + def test_constants_exported(self) -> None: + """Test that size limit constants are exported in __all__.""" + from omnibase_core.utils.workflow_executor import ( + MAX_STEP_PAYLOAD_SIZE_BYTES, + MAX_TOTAL_PAYLOAD_SIZE_BYTES, + MAX_WORKFLOW_STEPS, + ) + + assert MAX_WORKFLOW_STEPS == 1000 + assert MAX_STEP_PAYLOAD_SIZE_BYTES == 64 * 1024 # 64KB + assert MAX_TOTAL_PAYLOAD_SIZE_BYTES == 10 * 1024 * 1024 # 10MB + + def test_error_codes_available(self) -> None: + """Test that workflow limit error codes exist.""" + assert hasattr(EnumCoreErrorCode, "WORKFLOW_STEP_LIMIT_EXCEEDED") + assert hasattr(EnumCoreErrorCode, "WORKFLOW_PAYLOAD_SIZE_EXCEEDED") + assert hasattr(EnumCoreErrorCode, "WORKFLOW_TOTAL_PAYLOAD_EXCEEDED") + + @pytest.mark.asyncio + async def test_workflow_under_all_limits_succeeds( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test that workflow well under all limits executes successfully. + + Verifies that normal workflows are not affected by the size limits. + """ + # Create a normal workflow with 10 steps + steps = [ + ModelWorkflowStep( + step_id=uuid4(), + step_name=f"normal_step_{i}", + step_type="compute", + enabled=True, + ) + for i in range(10) + ] + + result = await execute_workflow( + workflow_definition=simple_workflow_definition, + workflow_steps=steps, + workflow_id=uuid4(), + ) + + assert result.execution_status == EnumWorkflowState.COMPLETED + assert len(result.completed_steps) == 10 + assert len(result.actions_emitted) == 10 + assert len(result.failed_steps) == 0 + + @pytest.mark.asyncio + async def test_step_limit_validation_includes_count_in_error( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test that step limit validation error includes the actual count.""" + from omnibase_core.utils.workflow_executor import MAX_WORKFLOW_STEPS + + # Create workflow exceeding limit by 1 (minimum needed to trigger error) + # Uses model_construct to bypass Pydantic validation for performance + num_steps = MAX_WORKFLOW_STEPS + 1 + steps = [ + _create_mock_workflow_step(step_name=f"step_{i}") for i in range(num_steps) + ] + + errors = await validate_workflow_definition(simple_workflow_definition, steps) + + # Error should include both the actual count and the limit + assert len(errors) > 0 + step_limit_error = next((e for e in errors if "maximum step limit" in e), None) + assert step_limit_error is not None + assert str(num_steps) in step_limit_error + assert str(MAX_WORKFLOW_STEPS) in step_limit_error + + @pytest.mark.asyncio + async def test_step_limit_short_circuits_validation( + self, + simple_workflow_definition: ModelWorkflowDefinition, + ) -> None: + """Test that step limit validation short-circuits to prevent DoS. + + When step count exceeds MAX_WORKFLOW_STEPS, validation should return + immediately with only the step count error. This prevents attackers from + causing CPU exhaustion by submitting workflows with millions of steps + that would otherwise trigger expensive cycle detection and per-step + validation. + + Security: OMN-670 - DoS mitigation via validation short-circuit + """ + from omnibase_core.utils.workflow_executor import MAX_WORKFLOW_STEPS + + # Create workflow with steps that would have other validation errors + # (invalid dependencies) if validation continued past the step count check + # Uses model_construct to bypass Pydantic validation for performance + num_steps = MAX_WORKFLOW_STEPS + 1 + invalid_dep_id = uuid4() # Non-existent dependency + steps = [ + _create_mock_workflow_step( + step_name=f"step_{i}", + depends_on=[ + invalid_dep_id + ], # Would fail dep validation if not short-circuited + ) + for i in range(num_steps) + ] + + errors = await validate_workflow_definition(simple_workflow_definition, steps) + + # Should return ONLY the step limit error due to short-circuit + # If validation continued, we would see additional errors for: + # - Invalid dependencies (num_steps errors - one per step) + # Without short-circuit, this would be 1 + num_steps errors + assert len(errors) == 1, ( + f"Expected exactly 1 error (step limit), got {len(errors)} errors. " + f"Short-circuit validation failed - other validations ran. Errors: {errors[:5]}..." + ) + assert "maximum step limit" in errors[0] + assert str(num_steps) in errors[0]