Repository navigation
feat(executor): implement missing workflow executor features [OMN-656] - #157
jonahgabriel merged 5 commits into
Conversation
Implement the 4 missing features for NodeOrchestrator Sequential Executor: - Add _build_workflow_context for step output aggregation - Add _validate_json_payload for early payload validation - Enhance _get_topological_order with priority-aware ordering using heapq - Add verify_workflow_integrity for contract hash verification - Add TypedDictWorkflowContext for type-safe context structure All features pass mypy --strict and existing unit tests.
WalkthroughAdds a TypedDict for workflow execution context and re-exports it; introduces deterministic SHA-256 workflow hashing and integrity verification, JSON payload validation, priority-clamped priority ordering, and workflow_context propagation into step execution. Model and metadata types gained hashing helpers/fields and tests were added/expanded. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes
Possibly related PRs
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/omnibase_core/utils/workflow_executor.py (2)
729-791: Add payload validation or update docstring.The docstring (Lines 743-744) claims the function raises
ModelOnexErrorfor non-JSON-serializable payloads, but the implementation doesn't call_validate_json_payload. Since_validate_json_payloadexists but is unused, consider:
- Add validation before constructing the action
- Or remove the "Raises" claim if validation is deferred
Apply this diff to add pre-flight validation:
def _create_action_for_step( step: ModelWorkflowStep, workflow_id: UUID, ) -> ModelAction: """ Create action for workflow step. Args: step: Workflow step to create action for workflow_id: Parent workflow ID Returns: ModelAction for step execution Raises: ModelOnexError: If payload is not JSON-serializable """ + # Build payload + payload = { + "workflow_id": str(workflow_id), + "step_id": str(step.step_id), + "step_name": step.step_name, + } + + # Validate JSON serializability + _validate_json_payload(payload, context=step.step_name) + # Map step type to action type action_type_map = {
820-826: Add None check for step.priority before clamping.Line 825 uses
min(step.priority, 10)which will raiseTypeErrorifstep.priorityisNone. WhileModelWorkflowStepmay have a default value, defensive coding suggests handling the None case explicitly.Apply this diff to add None handling:
step_order_key: dict[UUID, tuple[int, int]] = {} for idx, step in enumerate(workflow_steps): - clamped_priority = min(step.priority, 10) + clamped_priority = min(step.priority or 1, 10) step_order_key[step.step_id] = (clamped_priority, idx)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/omnibase_core/types/__init__.py(2 hunks)src/omnibase_core/types/typed_dict_workflow_context.py(1 hunks)src/omnibase_core/utils/workflow_executor.py(7 hunks)
🧰 Additional context used
📓 Path-based instructions (4)
**/*.{py,pyc,pyi}
📄 CodeRabbit inference engine (CLAUDE.md)
Always use Poetry for Python package management and task execution - never use pip or python directly
Files:
src/omnibase_core/types/__init__.pysrc/omnibase_core/types/typed_dict_workflow_context.pysrc/omnibase_core/utils/workflow_executor.py
src/omnibase_core/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/omnibase_core/**/*.py: Use ModelOnexError with EnumCoreErrorCode for all error handling, never generic Exception
Use @standard_error_handling decorator from omnibase_core.decorators.error_handling to eliminate try/catch boilerplate
Use protocol names (e.g., 'ProtocolEventBus') with container.get_service(), never concrete class names
Use EnumNodeKind for high-level architectural classification (EFFECT, COMPUTE, REDUCER, ORCHESTRATOR) and EnumNodeType for specific implementation types
Maintain 100% strict mypy compliance - all functions must have type annotations and zero mypy errors
Do NOT share node instances across threads without explicit synchronization - most ONEX components are not thread-safe by default
Implement mixin system using available mixins (MixinDiscoveryResponder, MixinEventHandler, MixinNodeExecutor, etc.) for cross-cutting concerns
Use ModelEventEnvelope for inter-service communication and event-driven processing patterns
Python version requirement: 3.12+
Structure project following the omnibase_core directory layout with clear separation of concerns (constants, container, decorators, discovery, enums, errors, events, infrastructure, logging, mixins, models, nodes, primitives, types, utils, validation)
Files:
src/omnibase_core/types/__init__.pysrc/omnibase_core/types/typed_dict_workflow_context.pysrc/omnibase_core/utils/workflow_executor.py
**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Import node classes from omnibase_core.nodes (NodeCompute, NodeEffect, NodeReducer, NodeOrchestrator), not from infrastructure submodules
Files:
src/omnibase_core/types/__init__.pysrc/omnibase_core/types/typed_dict_workflow_context.pysrc/omnibase_core/utils/workflow_executor.py
**/*.{py,pyi}
📄 CodeRabbit inference engine (CLAUDE.md)
Format code with black and sort imports with isort before committing
Files:
src/omnibase_core/types/__init__.pysrc/omnibase_core/types/typed_dict_workflow_context.pysrc/omnibase_core/utils/workflow_executor.py
🧬 Code graph analysis (2)
src/omnibase_core/types/__init__.py (1)
src/omnibase_core/types/typed_dict_workflow_context.py (1)
TypedDictWorkflowContext(12-29)
src/omnibase_core/utils/workflow_executor.py (3)
src/omnibase_core/types/typed_dict_workflow_context.py (1)
TypedDictWorkflowContext(12-29)src/omnibase_core/enums/enum_core_error_code.py (1)
EnumCoreErrorCode(9-163)src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (1)
ModelWorkflowDefinition(16-46)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Smoke Tests
- GitHub Check: claude-review
🔇 Additional comments (8)
src/omnibase_core/types/typed_dict_workflow_context.py (1)
12-29: LGTM - Well-structured TypedDict definition.The TypedDict provides clear type safety for workflow context with appropriate field types and comprehensive documentation.
src/omnibase_core/types/__init__.py (1)
182-182: LGTM - Consistent with module export patterns.The import and export additions follow the established conventions in this module.
Also applies to: 299-299
src/omnibase_core/utils/workflow_executor.py (6)
11-13: LGTM - Necessary imports for new features.The added imports support the hash computation, priority-aware topological ordering, and type-safe context building.
Also applies to: 37-37
839-860: LGTM - Correct priority-aware topological ordering.The heap-based Kahn's algorithm correctly prioritizes steps with lower priority values and earlier declaration order when dependencies are equal. The implementation properly handles the in-degree tracking and neighbor updates.
913-941: LGTM - Proper deterministic hash computation.The function correctly uses sorted keys and a default serializer to ensure consistent hashing across executions. SHA-256 is appropriate for integrity verification.
944-988: LGTM - Robust integrity verification implementation.The function properly handles optional hash verification, provides clear error messages on mismatch, and follows ONEX error handling patterns with
ModelOnexErrorandEnumCoreErrorCode.
992-998: LGTM - Appropriate public API exposure.Correctly exports
verify_workflow_integrityas a public function while keeping internal helpers private.
613-684: Verify that _build_workflow_context is integrated into execution flow.The function is well-implemented but appears unused. Neither
_execute_sequentialnor_execute_parallelcall this function or populate thestep_outputsdictionary that it expects. The execution functions initializestep_outputstracking but never capture actual outputs from steps.Run the following script to verify if this function is called anywhere in the codebase:
#!/bin/bash # Search for calls to _build_workflow_context rg -n '_build_workflow_context\s*\(' --type=py
PR Review: feat(executor): implement missing workflow executor features [OMN-656]SummaryThis PR adds 4 new utility functions to the workflow executor for step output aggregation, JSON payload validation, priority-aware topological ordering, and workflow integrity verification. The implementation is well-documented with comprehensive docstrings and examples. ✅ Strengths1. Code Quality & Documentation
2. Type Safety
3. Priority-Aware Ordering
4. Security - Integrity Verification
|
- Integrate _validate_json_payload into _create_action_for_step - Add 21 new unit tests for workflow executor functions: - TestBuildWorkflowContext (4 tests) - TestValidateJsonPayload (6 tests) - TestVerifyWorkflowIntegrity (3 tests) - TestComputeWorkflowHash (3 tests) - TestPriorityOrdering (5 tests) All 42 tests pass, mypy --strict passes.
Code Review - PR #157: Workflow Executor EnhancementsSummaryThis PR adds four important features to the workflow executor: step output aggregation, JSON payload validation, priority-aware topological ordering, and workflow integrity verification. The implementation is solid overall with excellent test coverage (477 new test lines). Here's my detailed feedback: ✅ Strengths1. Excellent Test Coverage
2. Type Safety
3. Clear Documentation
4. Fail-Fast Validation
🔍 Issues & Suggestions1. CRITICAL: Priority Behavior May Be Unintuitive
|
- Add "Used by" docstring to _validate_json_payload (addresses dead code concern) - Add None check for step.priority in _get_topological_order - Integrate _build_workflow_context in sequential/parallel execution - Add workflow_hash to result metadata for integrity verification - Add WorkflowStepExecutionContext.workflow_context parameter - Add 14 new tests for all changes (56 total tests now)
Code Review - PR #157: Workflow Executor Features✅ Overall AssessmentThis PR successfully implements 4 critical missing features for NodeOrchestrator's Sequential Executor. The implementation is well-structured, thoroughly tested, and follows omnibase_core conventions. Recommendation: Approve with minor suggestions below. 🎯 Strengths1. Excellent Test Coverage (1,211 new lines of tests)
2. Type Safety ✅
3. Documentation Quality 📚
4. Proper Error Handling
🔍 Code Quality Analysis_build_workflow_context ✅Purpose: Aggregate step outputs for inter-step data flow Strengths:
Observations:
_validate_json_payload ✅Purpose: Fail-fast validation for JSON serializability Strengths:
Minor Suggestion:
_get_topological_order (Priority-Aware) ✅Purpose: Deterministic step ordering with priority and declaration order Strengths:
Excellent Design: # Lower priority value = higher priority (min-heap behavior)
clamped_priority = min(step.priority, 10) if step.priority is not None else 1
step_order_key[step.step_id] = (clamped_priority, idx)Question: Why clamp to 1-10 when verify_workflow_integrity ✅Purpose: Tamper detection via SHA-256 hashing Strengths:
Security Consideration:
Performance Note:
🧪 Testing AnalysisTest Structure ✅
Coverage Gaps (Minor)
🔒 Security Review✅ No Security Concerns Identified
JSON Serialization Safety
⚡ Performance ConsiderationsAlgorithmic Complexity
All acceptable for typical workflow sizes (10-100 steps). Memory Usage
📋 RecommendationsCritical (Must Fix)None - all blocking issues resolved High Priority (Should Fix)
Low Priority (Nice to Have)
🎓 Adherence to CLAUDE.md
📊 Final Metrics
✅ Approval RecommendationAPPROVE - This PR is production-ready with minor documentation improvements suggested above. Rationale:
Suggested Next Steps:
🙏 AcknowledgmentExcellent work on this implementation! The attention to detail (deterministic hashing, priority-aware ordering, comprehensive tests) demonstrates strong engineering discipline. Correlation ID: |
…sh [OMN-656] - Clarify unused _context variable with TODO comment for future scaffolding - Document priority clamping rationale (1-1000 → 1-10 conversion) - Add 11 performance tests for large workflows (100+ steps) - Relax JSON validation to allow UUID/datetime with custom handler - Add workflow_hash field to contract metadata for persistence/caching
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.py (1)
43-50: LGTM! Optional workflow hash field for persistence and caching.The field is appropriately typed as
str | Nonewith a clear description. Consider whether to add a field validator to ensure the hash is a valid 64-character hex string when set, which would catch invalid hashes at model construction time rather than at runtime.Optional validation could be added:
from pydantic import field_validator @field_validator("workflow_hash") @classmethod def validate_hash_format(cls, v: str | None) -> str | None: if v is not None and (len(v) != 64 or not all(c in "0123456789abcdef" for c in v)): raise ValueError("workflow_hash must be a 64-character hex string") return vtests/unit/utils/test_workflow_executor.py (1)
988-1001: UUID permissive mode test has unused datetime import.Line 990 imports
datetimebut it's not used in this specific test method. However, it's used in the subsequent testtest_datetime_passes_in_permissive_modeat line 1005. Consider moving the import to the module level or to the test that uses it.def test_uuid_passes_in_permissive_mode(self) -> None: """Test UUID objects are accepted in default permissive mode (strict=False).""" - from datetime import datetime - from omnibase_core.utils.workflow_executor import _validate_json_payload
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py(2 hunks)src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.py(1 hunks)src/omnibase_core/utils/workflow_executor.py(19 hunks)tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py(1 hunks)tests/unit/utils/test_workflow_executor.py(2 hunks)
🧰 Additional context used
📓 Path-based instructions (5)
**/*.{py,pyc,pyi}
📄 CodeRabbit inference engine (CLAUDE.md)
Always use Poetry for Python package management and task execution - never use pip or python directly
Files:
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.pysrc/omnibase_core/models/contracts/subcontracts/model_workflow_definition.pytests/unit/utils/test_workflow_executor.pytests/unit/models/contracts/subcontracts/test_model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.py
src/omnibase_core/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/omnibase_core/**/*.py: Use ModelOnexError with EnumCoreErrorCode for all error handling, never generic Exception
Use @standard_error_handling decorator from omnibase_core.decorators.error_handling to eliminate try/catch boilerplate
Use protocol names (e.g., 'ProtocolEventBus') with container.get_service(), never concrete class names
Use EnumNodeKind for high-level architectural classification (EFFECT, COMPUTE, REDUCER, ORCHESTRATOR) and EnumNodeType for specific implementation types
Maintain 100% strict mypy compliance - all functions must have type annotations and zero mypy errors
Do NOT share node instances across threads without explicit synchronization - most ONEX components are not thread-safe by default
Implement mixin system using available mixins (MixinDiscoveryResponder, MixinEventHandler, MixinNodeExecutor, etc.) for cross-cutting concerns
Use ModelEventEnvelope for inter-service communication and event-driven processing patterns
Python version requirement: 3.12+
Structure project following the omnibase_core directory layout with clear separation of concerns (constants, container, decorators, discovery, enums, errors, events, infrastructure, logging, mixins, models, nodes, primitives, types, utils, validation)
Files:
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.pysrc/omnibase_core/models/contracts/subcontracts/model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.py
**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Import node classes from omnibase_core.nodes (NodeCompute, NodeEffect, NodeReducer, NodeOrchestrator), not from infrastructure submodules
Files:
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.pysrc/omnibase_core/models/contracts/subcontracts/model_workflow_definition.pytests/unit/utils/test_workflow_executor.pytests/unit/models/contracts/subcontracts/test_model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.py
**/*.{py,pyi}
📄 CodeRabbit inference engine (CLAUDE.md)
Format code with black and sort imports with isort before committing
Files:
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.pysrc/omnibase_core/models/contracts/subcontracts/model_workflow_definition.pytests/unit/utils/test_workflow_executor.pytests/unit/models/contracts/subcontracts/test_model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.py
tests/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
tests/**/*.py: Ensure minimum 60% test coverage for all code changes, tracked via pytest --cov
Use pytest with 4 parallel workers by default locally (via pytest-xdist -n4) and respect 60-second test timeout
Files:
tests/unit/utils/test_workflow_executor.pytests/unit/models/contracts/subcontracts/test_model_workflow_definition.py
🧬 Code graph analysis (3)
tests/unit/utils/test_workflow_executor.py (1)
src/omnibase_core/utils/workflow_executor.py (8)
_compute_workflow_hash(1045-1073)_build_workflow_context(676-747)_validate_json_payload(777-849)verify_workflow_integrity(1076-1120)_get_topological_order(932-992)_create_action_for_step(852-921)get_execution_order(280-355)_has_dependency_cycles(995-1042)
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py (4)
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (3)
ModelWorkflowDefinition(21-118)compute_workflow_hash(53-89)with_computed_hash(91-118)src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.py (1)
ModelWorkflowDefinitionMetadata(14-57)src/omnibase_core/models/primitives/model_semver.py (1)
ModelSemVer(24-143)src/omnibase_core/utils/workflow_executor.py (1)
_compute_workflow_hash(1045-1073)
src/omnibase_core/utils/workflow_executor.py (3)
src/omnibase_core/types/typed_dict_workflow_context.py (1)
TypedDictWorkflowContext(12-29)src/omnibase_core/enums/enum_core_error_code.py (1)
EnumCoreErrorCode(9-163)src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (1)
ModelWorkflowDefinition(21-118)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: claude-review
- GitHub Check: Smoke Tests
🔇 Additional comments (21)
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (2)
53-89: Well-structured deterministic hashing implementation.The implementation correctly:
- Excludes
workflow_hashto avoid circular dependency- Uses
sort_keys=Truefor deterministic JSON output- Uses
default=strto handle UUIDs and datetimesHowever, the docstring states it excludes "runtime metadata (timestamps, UUIDs generated at runtime)" but the code only explicitly excludes
workflow_hash. If any other runtime-generated fields exist or are added later, they would be included in the hash. Verify this aligns with the intended behavior.
91-118: Immutable pattern correctly applied for hash computation.The use of
model_copy(update=...)ensures the original instance remains unchanged while creating a new instance with the computed hash. This is the correct Pydantic pattern for immutable updates.tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py (3)
1-43: Comprehensive test fixture and imports.The fixture provides a well-structured baseline
ModelWorkflowDefinitionfor testing hash functionality. All necessary imports are present and used.
337-366: Important integration test validating contract vs executor hash behavior.This test documents the intentional design difference:
- Contract's
compute_workflow_hash()excludesworkflow_hashfield for persistence use cases- Executor's
_compute_workflow_hashincludes all fields for runtime integrityThe test correctly verifies both produce valid 64-char hex strings without asserting they must be equal, which aligns with the documented design intent.
399-434: Deduplication test validates content-based hashing.The test correctly demonstrates that identical workflow definitions produce identical hashes, enabling cache deduplication. The use of an inner factory function ensures clean, isolated test data.
tests/unit/utils/test_workflow_executor.py (4)
746-772: Good test coverage for workflow hash in execution results.This test validates the complete flow: hash is computed, stored in metadata, has correct format (64 hex chars), and matches the expected computed hash. This ensures integrity verification works end-to-end.
897-940: Thorough validation testing for JSON payload serializability.Tests cover key scenarios:
- Valid payloads pass
- Empty payloads pass
- Lambdas raise
ModelOnexErrorwith correct error code- Custom objects raise errors
- Error context includes payload keys and step context
The assertion on nested context structure (lines 936-940) correctly navigates the
ModelOnexErrorcontext hierarchy.
1738-1795: Defensive tests for None priority handling are well-designed.Using
MagicMockto simulate None priority values tests code resilience against potential future model changes. The tests verify that:
_get_topological_ordertreats None priority as default (priority 1)_create_action_for_stepdefaults None priority to 1This defensive coding approach aligns with robust software practices.
2029-2042: Performance test class is well-structured with clear documentation.The test patterns (linear chain, wide parallel, mixed, diamond) cover realistic workflow topologies. Performance targets (<5s for 100 steps, <100ms for topological sort) are reasonable and account for CI variability.
src/omnibase_core/utils/workflow_executor.py (12)
8-27: Excellent module-level documentation for priority clamping.The docstring clearly explains:
- The intentional design of step priority (1-1000) vs action priority (1-10)
- The clamping rule:
action_priority = min(step.priority, 10)- That this is expected behavior, not an error condition
- References to related architecture documentation
This prevents future confusion and aligns with the coding guidelines for comprehensive documentation.
146-147: Workflow hash computation and metadata storage correctly integrated.The hash is computed after validation passes (line 147) and stored in result metadata (line 176), enabling downstream integrity verification. This follows the pattern documented in the PR objectives.
Also applies to: 175-176
412-432: Workflow context building and step output tracking for sequential execution.The implementation correctly:
- Builds context from prior step outputs before each step
- Passes workflow_context to
WorkflowStepExecutionContext- Stores
action.payloadas step output for subsequent stepsNote: Using
action.payloadas the step output means downstream steps receive the action's payload structure, not actual execution results. This is consistent with the current design where the executor emits actions rather than executing steps directly.
549-559: TODO comment for future context usage in parallel execution.The TODO comment at lines 550-556 documents scaffolding for future enhancements. The context variable is prefixed with underscore (
_context) to indicate it's currently unused, which is the correct Python convention for intentionally unused variables. This is acceptable for incremental development.
676-747: Well-documented workflow context builder.
_build_workflow_contextis cleanly implemented with:
- Clear docstring with usage example
- UUID-to-string conversion for JSON compatibility
- Returns properly typed
TypedDictWorkflowContext- Aggregates completed steps and their outputs
750-774: Restrictive JSON default handler is a good security practice.
_json_default_for_workflowonly allowsUUIDanddatetimetypes, rejecting all other non-JSON-native types. This prevents accidental serialization of sensitive objects and provides clear error messages when unsupported types are encountered.
777-849: Comprehensive JSON payload validation with strict mode support.The implementation:
- Validates payloads are JSON-serializable before action creation
- Supports strict mode (only native JSON types) and permissive mode (UUID/datetime allowed)
- Raises
ModelOnexErrorwith helpful context (payload keys, step context)- Detailed docstring explains both modes with examples
This addresses the past review comment about unused validation - it's now called in
_create_action_for_stepat line 903.
890-921: Action creation with validated payloads.The implementation now:
- Builds the payload dict explicitly (lines 896-900)
- Validates JSON serializability via
_validate_json_payload(line 903)- Creates the
ModelActionwith the validated payloadThis addresses the past review comment about integrating payload validation into action creation.
932-992: Priority-aware topological ordering using heap-based Kahn's algorithm.The implementation correctly:
- Clamps priorities to 1-10 range (line 957)
- Uses
(clamped_priority, declaration_index, step_id)tuples for heap ordering- Lower priority value = higher importance (processed first)
- Falls back to declaration order for equal priorities
- Handles None priority defensively (defaults to 1)
This provides deterministic ordering that respects both dependencies and priority hints.
1045-1073: Executor hash computation includes all fields (intentionally different from contract hash).Unlike
ModelWorkflowDefinition.compute_workflow_hash()which excludesworkflow_hash, this executor-level function hashes the entire workflow definition including all fields. This is intentional:
- Contract hash: for persistence/caching (stable across hash-setting)
- Executor hash: for runtime integrity verification (complete snapshot)
The tests in
test_model_workflow_definition.py(lines 337-366) explicitly document and verify this design difference.
1076-1120: Integrity verification with proper null-safety and error context.
verify_workflow_integrity:
- Correctly skips verification when
expected_hash is None- Uses timing-safe comparison (string equality) for hash matching
- Raises
ModelOnexErrorwith comprehensive context (expected/actual hash, workflow name)- Error code
VALIDATION_ERRORis appropriate for tamper detection
1123-1130: Public API correctly updated to exposeverify_workflow_integrity.The
__all__list now includes the new public function, making it available for external consumers to verify workflow integrity before execution.
PR Review: Workflow Executor Features [OMN-656]SummaryThis PR adds critical missing features to the workflow executor with high quality implementation, comprehensive test coverage (1,886 added test lines), and excellent documentation. ✅ Strengths1. Exceptional Documentation
2. Outstanding Test Coverage80+ test cases covering:
Test file: 2,629 lines with detailed scenario coverage. 3. Type Safety
4. Security & Integrity
5. Priority-Aware Topological OrderingLines 932-992: Excellent heapq-based implementation:
6. Wave-Based Context SharingLines 585-589: Smart parallel execution isolation - steps in same wave can't see each other's outputs, but subsequent waves access prior outputs. 🔍 Areas for Improvement1. Duplicate Hash Computation LogicLocations: model_workflow_definition.py:54-90 and workflow_executor.py:1045-1073 Recommendation: Consolidate by delegating executor's _compute_workflow_hash to model's compute_workflow_hash method for consistency. 2. Unused Context VariableLocation: workflow_executor.py:557 _context created but never used in parallel path. TODO comment explains future scaffolding. Current approach acceptable with OMN-656 reference. 3. Test Coverage Gaps (Minor)
🛡️ Security AssessmentStrengths: Overall: Strong security posture with proper validation. 📊 PerformanceStrengths: No regressions expected. 🧪 Test CoverageTotal: 80+ test cases, 2,629 lines Coverage exceeds project 60% requirement. Exceptional quality. 📝 Code QualityCLAUDE.md Adherence: Best Practices: 🎯 RecommendationsHigh Priority:
Medium Priority: 🏁 Final VerdictStatus: ✅ APPROVED (with minor recommendations) Exceptional engineering quality:
Production-ready with minor improvement opportunities. Test coverage is exemplary. Reviewed by: Claude Code (Sonnet 4.5) |
…OMN-656] - Delegate _compute_workflow_hash() to model's compute_workflow_hash() for DRY consistency and reduced code duplication - Remove unused hashlib import from workflow_executor.py - Add TestModelWorkflowDefinitionFrozenBehavior tests for immutability - Add TestWorkflowSizeLimits tests for DoS prevention (7 tests): - Large workflow step count (100 steps) - Deep dependency chains (50 steps) - Fan-in patterns (20 predecessors) - Execution order determinism - Multiple fan-in points - Graceful validation failure - Varying step types
Code Review - PR #157: Workflow Executor EnhancementsSummaryThis PR adds critical missing features to the workflow executor: step output aggregation, payload validation, priority-aware ordering, and integrity verification. The implementation is well-structured with excellent test coverage (2,300+ lines of new tests). ✅ Strengths1. Excellent Documentation
2. Strong Type Safety
3. Robust Test Coverage
4. Security & Validation
5. Performance Optimization
🔍 Areas for Consideration1. Unused Context in Parallel Execution (Minor)Location: workflow_executor.py:542-557 The _context variable is created but not used in the parallel execution path. Recommendation:
Severity: 🟢 Low (intentional scaffolding) 2. Hash Computation Performance (Minor)Location: model_workflow_definition.py:54-90 Observation: Hash computation uses model_dump() which creates a full deep copy of the workflow definition, then serializes to JSON. Performance Impact:
Recommendations:
Severity: 🟢 Low (premature optimization concern) 3. Priority Clamping Visibility (Minor)Location: workflow_executor.py:889 Priority clamping from 1-1000 (step priority) to 1-10 (action priority) is silent. Recommendation:
Severity: 🟢 Low (design decision, not a bug) 4. JSON Serialization Edge Cases (Minor)Location: workflow_executor.py:749-769 Concern: _json_default_for_workflow only handles UUID and datetime. Other Pydantic models in payloads will fail. Recommendation:
Severity: 🟢 Low (intentional restriction) 5. Circular Dependency in Hash Computation (Info)Location: model_workflow_definition.py:82-83 Analysis: ✅ Well-handled: Safe guard prevents KeyError if metadata is missing. Severity: 🟢 None (well-designed) 🛡️ Security Review✅ Secure Practices
|
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py (1)
337-367: Consider clarifying the expected behavior difference between contract and executor hashes.The test
test_executor_hash_includes_all_fieldsdocuments that contract hash and executor hash may differ, but the assertions only verify both are valid SHA-256 strings without asserting they're equal or different. The comment (lines 358-362) explains this is intentional, but the test doesn't actually verify the "may differ" claim.If the intent is to document that hashes may differ, consider adding an explicit assertion or comment clarifying when they would differ:
# This is expected and intentional design # The important invariant is that each is deterministic and content-based assert isinstance(contract_hash, str) assert isinstance(executor_hash, str) assert len(contract_hash) == 64 assert len(executor_hash) == 64 + # Note: These hashes are currently equal because workflow_hash is None. + # They would differ if workflow_hash were set to a non-None value.tests/unit/utils/test_workflow_executor.py (1)
2237-2276: Performance test timing assertions may be flaky in CI.The 5-second timeout for 100 sequential steps is generous, but CI environments can have significant variability. Consider adding a note about potential flakiness or using a more generous threshold for CI.
The timing assertions (e.g.,
elapsed_time < 5.0) may cause intermittent failures in resource-constrained CI environments. Consider either:
- Adding a larger tolerance for CI
- Skipping timing assertions in CI via environment variable check
- Documenting expected behavior if tests become flaky
# Performance assertion: should complete in under 5 seconds # This is generous to account for CI variability + # If flaky in CI, consider increasing to 10.0 or skipping timing check assert elapsed_time < 5.0, (src/omnibase_core/utils/workflow_executor.py (1)
548-558: TODO comment references non-existent ticket format.The TODO comment references "OMN-656" but the codebase uses Linear tickets with format "OMN-XXX". The variable
_contextis created but only its assignment is tracked, not used further in the parallel path.The
_contextvariable is assigned but not used in the parallel execution path. Consider either:
- Using it (e.g., for timing/logging)
- Removing the scaffolding if not needed soon
- Using
_ =prefix convention if intentionally unused- # TODO(OMN-656): Use _context for step execution logic that needs prior outputs. + # TODO: Use step context for step execution logic that needs prior outputs. # Currently scaffolded for future use cases: # - Accessing prior wave outputs via _context.workflow_context # - Tracking step timing via _context.started_at/_context.completed_at # - Error context via _context.error # The sequential execution path uses context.completed_at; parallel path # will need similar integration when step execution becomes more complex. - _context = WorkflowStepExecutionContext( + _ = WorkflowStepExecutionContext( # noqa: F841 - scaffolded for future use step, workflow_id, completed_step_ids, wave_context )
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/omnibase_core/utils/workflow_executor.py(19 hunks)tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py(1 hunks)tests/unit/utils/test_workflow_executor.py(2 hunks)
🧰 Additional context used
📓 Path-based instructions (5)
**/*.{py,pyc,pyi}
📄 CodeRabbit inference engine (CLAUDE.md)
Always use Poetry for Python package management and task execution - never use pip or python directly
Files:
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.pytests/unit/utils/test_workflow_executor.py
**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
Import node classes from omnibase_core.nodes (NodeCompute, NodeEffect, NodeReducer, NodeOrchestrator), not from infrastructure submodules
Files:
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.pytests/unit/utils/test_workflow_executor.py
**/*.{py,pyi}
📄 CodeRabbit inference engine (CLAUDE.md)
Format code with black and sort imports with isort before committing
Files:
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.pysrc/omnibase_core/utils/workflow_executor.pytests/unit/utils/test_workflow_executor.py
tests/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
tests/**/*.py: Ensure minimum 60% test coverage for all code changes, tracked via pytest --cov
Use pytest with 4 parallel workers by default locally (via pytest-xdist -n4) and respect 60-second test timeout
Files:
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.pytests/unit/utils/test_workflow_executor.py
src/omnibase_core/**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
src/omnibase_core/**/*.py: Use ModelOnexError with EnumCoreErrorCode for all error handling, never generic Exception
Use @standard_error_handling decorator from omnibase_core.decorators.error_handling to eliminate try/catch boilerplate
Use protocol names (e.g., 'ProtocolEventBus') with container.get_service(), never concrete class names
Use EnumNodeKind for high-level architectural classification (EFFECT, COMPUTE, REDUCER, ORCHESTRATOR) and EnumNodeType for specific implementation types
Maintain 100% strict mypy compliance - all functions must have type annotations and zero mypy errors
Do NOT share node instances across threads without explicit synchronization - most ONEX components are not thread-safe by default
Implement mixin system using available mixins (MixinDiscoveryResponder, MixinEventHandler, MixinNodeExecutor, etc.) for cross-cutting concerns
Use ModelEventEnvelope for inter-service communication and event-driven processing patterns
Python version requirement: 3.12+
Structure project following the omnibase_core directory layout with clear separation of concerns (constants, container, decorators, discovery, enums, errors, events, infrastructure, logging, mixins, models, nodes, primitives, types, utils, validation)
Files:
src/omnibase_core/utils/workflow_executor.py
🧬 Code graph analysis (3)
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py (2)
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (3)
ModelWorkflowDefinition(21-118)compute_workflow_hash(53-89)with_computed_hash(91-118)src/omnibase_core/models/contracts/subcontracts/model_workflow_definition_metadata.py (1)
ModelWorkflowDefinitionMetadata(14-57)
src/omnibase_core/utils/workflow_executor.py (3)
src/omnibase_core/types/typed_dict_workflow_context.py (1)
TypedDictWorkflowContext(12-29)src/omnibase_core/enums/enum_core_error_code.py (1)
EnumCoreErrorCode(9-163)src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (2)
ModelWorkflowDefinition(21-118)compute_workflow_hash(53-89)
tests/unit/utils/test_workflow_executor.py (6)
src/omnibase_core/models/contracts/subcontracts/model_workflow_definition.py (1)
ModelWorkflowDefinition(21-118)src/omnibase_core/utils/workflow_executor.py (9)
_compute_workflow_hash(1044-1066)_build_workflow_context(675-746)_validate_json_payload(776-848)verify_workflow_integrity(1069-1113)_get_topological_order(931-991)_create_action_for_step(851-920)get_execution_order(279-354)validate_workflow_definition(180-276)_has_dependency_cycles(994-1041)src/omnibase_core/enums/enum_core_error_code.py (1)
EnumCoreErrorCode(9-163)src/omnibase_core/enums/enum_workflow_coordination.py (1)
EnumFailureRecoveryStrategy(40-46)src/omnibase_core/models/workflow/execution/model_declarative_workflow_step_context.py (1)
ModelDeclarativeWorkflowStepContext(15-45)src/omnibase_core/types/typed_dict_workflow_context.py (1)
TypedDictWorkflowContext(12-29)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: claude-review
- GitHub Check: Smoke Tests
🔇 Additional comments (23)
tests/unit/models/contracts/subcontracts/test_model_workflow_definition.py (4)
1-42: Well-structured test file with comprehensive coverage.The fixture and imports are correctly organized. The
simple_workflow_definitionfixture provides a clean reusable base for all tests with all required fields properly initialized.
45-80: Good coverage of workflow_hash field behaviors in metadata.Tests cover all three scenarios: default None, explicit set, and explicit None. The test at line 61 using
"a" * 64is a valid approach to simulate a SHA-256 hex string for testing purposes.
83-165: Thorough testing of compute_workflow_hash method.Tests correctly validate:
- SHA-256 format (64 hex characters)
- Determinism across multiple calls
- Different definitions produce different hashes
- The workflow_hash field is excluded from computation (preventing circular dependency)
399-434: Well-designed deduplication test.Using a factory function
create_workflow()to create two identical but separate instances is the correct approach to test that content-based hashing works for deduplication scenarios.tests/unit/utils/test_workflow_executor.py (7)
745-807: Good test coverage for workflow hash in metadata.Tests verify hash presence, format, determinism, and consistency across execution modes. The assertion that hashes are consistent regardless of execution mode (sequential vs parallel) correctly validates that the hash is based on the definition, not the execution.
810-894: Comprehensive tests for _build_workflow_context.Tests cover all scenarios: empty steps, single step, multiple steps, and partial outputs. The test at lines 876-894 correctly handles the edge case where not all completed steps have corresponding outputs.
897-1059: Thorough JSON payload validation tests.Excellent coverage including:
- Valid payloads (basic and empty)
- Invalid payloads (lambda, custom objects, nested)
- Permissive vs strict mode behavior
- UUID and datetime handling
- Error message verification with context
1061-1113: Well-structured integrity verification tests.Tests cover the three main scenarios: None hash (skip), matching hash (pass), and mismatched hash (error with context). The assertion structure properly navigates the nested context from ModelOnexError.
1738-1795: Defensive None priority handling tests are valuable.Using MagicMock to simulate None priority values tests defensive code paths that guard against future model changes. The docstrings clearly explain that this tests defensive handling rather than expected runtime behavior.
2029-2235: Performance test helper method is well-designed.The
_create_large_workflow_stepshelper supports multiple dependency patterns (linear, parallel, mixed, diamond) which enables comprehensive performance testing. The implementation correctly handles edge cases like small workflow sizes.
2632-2848: Size limit tests provide good boundary testing.Tests cover large step counts, deep dependency chains, wide fan-in, and determinism verification. The test at lines 2800-2848 for execution order determinism is particularly valuable for debugging reproducibility.
src/omnibase_core/utils/workflow_executor.py (12)
8-28: Excellent module-level documentation for priority clamping.The docstring clearly explains the design rationale for priority clamping (1-1000 → 1-10) and references the relevant architecture documentation. This prevents confusion about why priorities are clamped and documents it as intentional behavior.
145-176: Workflow hash computation and metadata integration is correct.The hash is computed after validation passes (line 146), ensuring only valid workflows get hashed. The hash is then added to result metadata (line 175) for integrity verification by consumers.
383-431: Workflow context integration in sequential execution is well-implemented.The context is properly built before each step execution (lines 412-414), passed to the step context (line 418), and step outputs are stored for subsequent steps (line 431). This enables data flow between sequential steps.
504-605: Parallel execution context handling is correct.Wave context is built once per wave (lines 586-588) before parallel execution, ensuring steps in the same wave cannot see each other's outputs (they run concurrently). Step outputs are stored after completion (line 605) for subsequent waves.
675-746: _build_workflow_context is well-documented and correctly implemented.The function properly converts UUIDs to strings for JSON compatibility (lines 734, 737-739) and returns a TypedDict with all required fields. The docstring includes a comprehensive example.
749-773: Custom JSON serializer correctly limits allowed types.The
_json_default_for_workflowfunction explicitly allows only UUID and datetime, raising TypeError for all other types. This is the correct behavior for catching programming errors early while supporting common workflow types.
776-848: Payload validation addresses the previously flagged issue.The
_validate_json_payloadfunction is now properly integrated and called from_create_action_for_step(line 902). This resolves the past review comment about the unused validation function.
889-920: Action creation with validation is correctly implemented.Priority clamping (line 892), payload construction (lines 895-899), validation (line 902), and action creation are all properly sequenced. The validation call ensures fail-fast behavior for non-serializable payloads.
931-991: Priority-aware topological ordering is correctly implemented.The implementation uses heapq for efficient priority-aware ordering with:
- Priority clamping to 1-10 (line 956)
- Declaration order as tiebreaker (line 957)
- Defensive None handling defaulting to 1 (line 956)
The heap tuple
(priority, idx, step_id)ensures correct ordering by priority first, then declaration order.
1044-1066: _compute_workflow_hash correctly delegates to model method.The thin wrapper delegates to
ModelWorkflowDefinition.compute_workflow_hash()for consistency. The docstring accurately describes the delegation and the exclusion of workflow_hash field.
1069-1113: verify_workflow_integrity is well-implemented.The function correctly:
- Skips verification when expected_hash is None (line 1097-1098)
- Computes actual hash and compares (lines 1101, 1104)
- Raises ModelOnexError with detailed context on mismatch (lines 1105-1112)
The context includes expected_hash, actual_hash, and workflow_name for debugging.
1116-1123: Public API correctly updated.The
__all__list now includesverify_workflow_integrity, making it part of the module's public API. This aligns with the PR objectives.
Summary
_build_workflow_contextfor step output aggregation between workflow steps_validate_json_payloadfor early payload validation during action creation_get_topological_orderwith priority-aware ordering using heapqverify_workflow_integrityfor contract hash verification (tamper detection)TypedDictWorkflowContextfor type-safe context structureTest plan
Linear ticket
OMN-656
Summary by CodeRabbit
New Features
Tests
✏️ Tip: You can customize this high-level summary in your review settings.