Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion docs/guides/THREADING.md
Original file line number Diff line number Diff line change
Expand Up @@ -1712,9 +1712,22 @@ class TestNodeComputeThreadSafety:
| ModelComputeInput/Output | Yes (frozen=True) | None needed |
| ModelReducerInput/Output | Yes (frozen=True) | None needed |
| ModelEffectInput/Output | Yes (frozen=True) | None needed |
| ModelOrchestratorInput/Output | Yes (frozen=True) | None needed |
| ModelOrchestratorInput | Partial* | See note below |
| ModelOrchestratorOutput | Yes (frozen=True) | None needed |
| EffectIOConfig models | Yes (frozen=True) | None needed |

**\*ModelOrchestratorInput Thread Safety Note**:

`ModelOrchestratorInput` is top-level frozen (`frozen=True`), but contains a mutable `metadata` field of type `ModelOrchestratorInputMetadata` (`frozen=False`). This means:

- Top-level fields (workflow_id, steps, execution_mode, etc.) cannot be reassigned
- The `metadata` object CAN be mutated in place (e.g., `input.metadata.source = "new"`)

**Mitigation**:
1. **Treat as read-only** after creation (recommended)
2. **Use model_copy()** to create modified copies: `input.model_copy(update={"metadata": new_metadata})`
3. **Use explicit locks** if mutation across threads is required

### Workflow Contract Models (v0.4.0+)

| Component | Thread-Safe? | Mitigation |
Expand Down
26 changes: 23 additions & 3 deletions src/omnibase_core/mixins/mixin_workflow_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,18 +7,22 @@
Typing: Strongly typed with strategic Any usage for mixin kwargs and configuration dicts.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any
from uuid import UUID

if TYPE_CHECKING:
from omnibase_core.models.workflow.execution.model_workflow_state_snapshot import (
ModelWorkflowStateSnapshot,
)
from omnibase_core.types.typed_dict_mixin_types import TypedDictWorkflowStepConfig

from omnibase_core.enums.enum_workflow_execution import EnumExecutionMode
from omnibase_core.models.contracts.model_workflow_step import ModelWorkflowStep
from omnibase_core.models.contracts.subcontracts.model_workflow_definition import (
ModelWorkflowDefinition,
)
from omnibase_core.models.workflow import ModelWorkflowStateSnapshot
from omnibase_core.types.type_workflow_context import WorkflowContextType
from omnibase_core.utils.workflow_executor import (
WorkflowExecutionResult,
Expand Down Expand Up @@ -116,6 +120,12 @@ def update_workflow_state(
state is replaced, not merged. Step IDs are stored as tuples
for immutability.
"""
# Lazy import to avoid circular import at module load time.
# The TYPE_CHECKING import provides type hints; this provides the runtime class.
from omnibase_core.models.workflow.execution.model_workflow_state_snapshot import (
ModelWorkflowStateSnapshot,
)

self._workflow_state = ModelWorkflowStateSnapshot(
workflow_id=workflow_id,
current_step_index=current_step_index,
Expand Down Expand Up @@ -181,6 +191,16 @@ async def execute_workflow_from_contract(
completed_ids = [UUID(step_id) for step_id in result.completed_steps]
failed_ids = [UUID(step_id) for step_id in result.failed_steps]

# Metadata is always present when result comes from execute_workflow().
# All internal execution paths (_execute_sequential, _execute_parallel,
# _execute_batch) explicitly create ModelWorkflowResultMetadata.
# The Optional type in ModelDeclarativeWorkflowResult exists for API
# flexibility (direct instantiation), not for the execute_workflow() path.
assert result.metadata is not None, (
"execute_workflow() must always return result with metadata; "
"this indicates a bug in workflow_executor.py"
)

self.update_workflow_state(
workflow_id=workflow_id,
current_step_index=len(completed_ids) + len(failed_ids),
Expand All @@ -190,7 +210,7 @@ async def execute_workflow_from_contract(
"execution_status": result.execution_status.value,
"execution_time_ms": result.execution_time_ms,
"actions_count": len(result.actions_emitted),
**result.metadata,
**result.metadata.model_dump(exclude_unset=True),
},
)

Expand Down Expand Up @@ -251,7 +271,7 @@ def get_workflow_execution_order(

def create_workflow_steps_from_config(
self,
steps_config: list["TypedDictWorkflowStepConfig"],
steps_config: list[TypedDictWorkflowStepConfig],
) -> list[ModelWorkflowStep]:
"""
Create ModelWorkflowStep instances from configuration dictionaries.
Expand Down
26 changes: 0 additions & 26 deletions src/omnibase_core/models/core/model_base_input_state.py

This file was deleted.

30 changes: 0 additions & 30 deletions src/omnibase_core/models/core/model_base_output_state.py

This file was deleted.

2 changes: 2 additions & 0 deletions src/omnibase_core/models/orchestrator/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@
from .model_orchestrator_context import ModelOrchestratorContext
from .model_orchestrator_graph import ModelOrchestratorGraph
from .model_orchestrator_input import ModelOrchestratorInput
from .model_orchestrator_input_metadata import ModelOrchestratorInputMetadata
from .model_orchestrator_output import ModelOrchestratorOutput
from .model_orchestrator_plan import ModelOrchestratorPlan
from .model_orchestrator_result import ModelOrchestratorResult
Expand All @@ -94,6 +95,7 @@
"ModelOrchestratorContext",
"ModelOrchestratorGraph",
"ModelOrchestratorInput",
"ModelOrchestratorInputMetadata",
"ModelOrchestratorOutput",
"ModelOrchestratorPlan",
"ModelOrchestratorResult",
Expand Down
3 changes: 3 additions & 0 deletions src/omnibase_core/models/orchestrator/model_action.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,3 +163,6 @@ class ModelAction(BaseModel):
use_enum_values=False,
from_attributes=True,
)


__all__ = ["ModelAction"]
Original file line number Diff line number Diff line change
Expand Up @@ -155,3 +155,6 @@ def get_stats(self) -> TypedDictLoadBalancerStats:
utilization=len(self.active_operations) / self.max_concurrent_operations,
total_operations=sum(self.operation_counts.values()),
)


__all__ = ["ModelLoadBalancer"]
Original file line number Diff line number Diff line change
Expand Up @@ -21,3 +21,6 @@ class ModelOrchestratorGraph(BaseModel):
graph_name: str = Field(default=..., description="Graph name")
nodes: list[ModelGraphNode] = Field(default_factory=list, description="Graph nodes")
edges: list[ModelGraphEdge] = Field(default_factory=list, description="Graph edges")


__all__ = ["ModelOrchestratorGraph"]
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,12 @@
parallelism, timeouts, and failure handling strategies.

Thread Safety:
ModelOrchestratorInput is immutable (frozen=True) and thread-safe.
All instances are read-only after creation, making them safe to share
across threads without synchronization.
ModelOrchestratorInput itself is frozen (frozen=True), meaning top-level
fields cannot be reassigned after creation. However, the `metadata` field
contains a mutable ModelOrchestratorInputMetadata object that can be
modified in place. If thread safety is required, either:
(a) Do not mutate metadata after creation, or
(b) Use appropriate synchronization when accessing metadata across threads.

Key Features:
- Multiple execution modes (SEQUENTIAL, PARALLEL, CONDITIONAL)
Expand Down Expand Up @@ -59,7 +62,9 @@
from pydantic import BaseModel, ConfigDict, Field

from omnibase_core.enums.enum_workflow_execution import EnumExecutionMode
from omnibase_core.models.common.model_schema_value import ModelSchemaValue
from omnibase_core.models.orchestrator.model_orchestrator_input_metadata import (
ModelOrchestratorInputMetadata,
)


class ModelOrchestratorInput(BaseModel):
Expand All @@ -70,6 +75,18 @@ class ModelOrchestratorInput(BaseModel):
configuration for execution modes, parallelism, timeouts, and failure
handling. Used by NodeOrchestrator to coordinate multi-step workflows.

Thread Safety:
This model is top-level frozen (frozen=True), meaning you cannot reassign
fields after creation. However, the `metadata` field contains a mutable
ModelOrchestratorInputMetadata object that CAN be modified in place.

Warning:
Do NOT mutate nested metadata across threads without synchronization.
If thread safety is required, either:
(a) Treat metadata as read-only after creation, or
(b) Use explicit locks when accessing metadata across threads, or
(c) Create new instances using `model_copy(update={"metadata": ...})`

Attributes:
workflow_id: Unique identifier for this workflow instance.
steps: List of workflow step definitions. Each step is a dictionary
Expand All @@ -88,7 +105,7 @@ class ModelOrchestratorInput(BaseModel):
operations. Defaults to False.
dependency_resolution_enabled: Whether to automatically resolve step
dependencies based on declared inputs/outputs. Defaults to True.
metadata: Additional context metadata for tracking and custom behavior.
metadata: Typed workflow metadata for observability, FSM control, and persistence.
timestamp: When this input was created. Auto-generated to current time.

Example:
Expand All @@ -103,8 +120,9 @@ class ModelOrchestratorInput(BaseModel):
... )
>>>
>>> # To "update" a frozen model, use model_copy
>>> original = ModelOrchestratorInput(workflow_id=uuid4(), steps=[], metadata={})
>>> updated = original.model_copy(update={"metadata": {"key": "value"}})
>>> original = ModelOrchestratorInput(workflow_id=uuid4(), steps=[])
>>> new_meta = ModelOrchestratorInputMetadata(source="updated")
>>> updated = original.model_copy(update={"metadata": new_meta})
"""

workflow_id: UUID = Field(..., description="Unique workflow identifier")
Expand Down Expand Up @@ -132,8 +150,9 @@ class ModelOrchestratorInput(BaseModel):
dependency_resolution_enabled: bool = Field(
default=True, description="Enable automatic dependency resolution"
)
metadata: dict[str, ModelSchemaValue] = Field(
default_factory=dict, description="Additional workflow metadata"
metadata: ModelOrchestratorInputMetadata = Field(
default_factory=ModelOrchestratorInputMetadata,
description="Typed workflow metadata for observability, FSM control, and persistence",
)
timestamp: datetime = Field(
default_factory=datetime.now, description="Workflow creation timestamp"
Expand All @@ -146,3 +165,6 @@ class ModelOrchestratorInput(BaseModel):
extra="forbid",
from_attributes=True,
)


__all__ = ["ModelOrchestratorInput"]
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
# SPDX-FileCopyrightText: 2025 OmniNode Team
# SPDX-License-Identifier: Apache-2.0
"""
Typed metadata model for orchestrator input.

This module provides ModelOrchestratorInputMetadata which replaces the
untyped dict[str, ModelSchemaValue] pattern in ModelOrchestratorInput
with strongly-typed fields for full type safety and IDE support.

Thread Safety:
ModelOrchestratorInputMetadata is mutable (frozen=False) as input
metadata may be modified during workflow execution preparation.
Instances should not be shared across threads without synchronization.

Key Features:
- Full type safety for all metadata fields
- IDE autocomplete and type checking support
- No dict-based escape hatches - all fields are explicitly typed
- Based on actual usage audit of integration tests and documentation

Example:
>>> from uuid import uuid4
>>> from omnibase_core.models.orchestrator import (
... ModelOrchestratorInputMetadata,
... )
>>>
>>> # Create metadata with tracing info
>>> metadata = ModelOrchestratorInputMetadata(
... source="api_gateway",
... environment="production",
... correlation_id=uuid4(),
... trigger="start",
... persist_result=True,
... )

See Also:
- omnibase_core.models.orchestrator.model_orchestrator_input: Parent model
- docs/guides/node-building/06_ORCHESTRATOR_NODE_TUTORIAL.md: Tutorial
"""

from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field


class ModelOrchestratorInputMetadata(BaseModel):
"""Typed metadata for orchestrator input.

Replaces dict[str, ModelSchemaValue] with strongly-typed fields.
All fields are based on actual usage audit of integration tests and docs.

Attributes:
source: Origin of the workflow request (e.g., "api_gateway", "scheduler").
environment: Execution environment (e.g., "test", "staging", "prod").
correlation_id: Request correlation ID for distributed tracing.
trigger: FSM trigger for state transitions. Defaults to "process".
persist_result: Whether to persist workflow results. Defaults to False.

Example:
>>> metadata = ModelOrchestratorInputMetadata(
... source="scheduler",
... environment="staging",
... trigger="process",
... )
>>> metadata.source
'scheduler'
"""

model_config = ConfigDict(
frozen=False, # Mutable - input metadata may be modified during execution
extra="forbid",
from_attributes=True,
)

# Observability/Tracing
source: str | None = Field(
default=None,
description="Origin of the workflow request",
)

environment: str | None = Field(
default=None,
description="Execution environment (test, staging, prod)",
)

correlation_id: UUID | None = Field(
default=None,
description="Request correlation ID for distributed tracing",
)

# FSM Control
trigger: str = Field(
default="process",
description="FSM trigger for state transitions",
)

# Persistence Control
persist_result: bool = Field(
default=False,
description="Whether to persist workflow results",
)


__all__ = ["ModelOrchestratorInputMetadata"]
Original file line number Diff line number Diff line change
Expand Up @@ -202,3 +202,6 @@ def convert_actions_emitted(
default=None,
description="Custom output fields for orchestrator-specific data",
)


__all__ = ["ModelOrchestratorOutput"]
Original file line number Diff line number Diff line change
Expand Up @@ -29,3 +29,6 @@ class ModelOrchestratorPlan(BaseModel):
custom_metadata: ModelCustomFields | None = Field(
default=None, description="Custom metadata fields"
)


__all__ = ["ModelOrchestratorPlan"]
Loading