orchestrata_ultimate.orchestrator.intent_workflow

Reference › orchestrata_ultimate › orchestrator › intent_workflow

A small engine for running several intents as one workflow. No other reference page imports it. It builds on the IntentLifecycleManager for status changes and on RedisStateManager for workflow state, and its steps hold Intent records.

A WorkflowStep has an id, an intent, depends_on ids, a FailureStrategy (retry, rollback, continue or fail; the default is fail), max_retries (3) and timeout_seconds (300). IntentWorkflow.define builds a WorkflowDef with an execution mode, and execute marks the workflow running in Redis, then runs it sequentially, in parallel, or as a DAG, and records the final status and duration. An unknown mode fails the result. The DAG mode repeatedly picks steps whose dependencies have completed and fails with "Deadlock: no ready steps" if none is ready; a failed step either stops the workflow (FAIL) or is treated as done (CONTINUE). _execute_step stores the step's intent and walks it through the lifecycle statuses to COMPLETED using the lifecycle manager, or to FAILED on error; _retry_step handles retries. Control methods are cancel, pause, resume, get_status and list_workflows, which can filter by WorkflowStatus. The result is a WorkflowResult with per-step results, error, timestamps and duration.

At a glance

Class diagram

Dependency diagram

API

Classes

WorkflowStatus (enum)

Bases: StrEnum · Decorators: traces(spec='003', story='US-008', reqs=['REQ-003-US-008-007']) · Line 25

Workflow execution status.

FailureStrategy (enum)

Bases: StrEnum · Decorators: traces(spec='003', story='US-008', reqs=['REQ-003-US-008-004', 'REQ-003-US-008-016']) · Line 37

How to handle step failure.

WorkflowStep (dataclass)

Decorators: traces(spec='003', story='US-008', reqs=['REQ-003-US-008-001', 'REQ-003-US-008-012']), dataclass · Line 53

A single step in a workflow.

WorkflowDef (dataclass)

Decorators: traces(spec='003', story='US-008', reqs=['REQ-003-US-008-001', 'REQ-003-US-008-012']), dataclass · Line 79

Workflow definition with steps and execution mode.

WorkflowResult (dataclass)

Decorators: traces(spec='003', story='US-008', reqs=['REQ-003-US-008-010', 'REQ-003-US-008-022']), dataclass · Line 106

Result of workflow execution.

IntentWorkflow

Decorators: traces(spec='003', story='US-008', reqs=['REQ-003-US-008-002', 'REQ-003-US-008-003', 'REQ-003-US-008-004', 'REQ-003-US-… · Line 143

Workflow engine for multi-step intent execution.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/intent_workflow.py

"""Intent workflow engine for multi-step intent execution.

Supports sequential, parallel, and DAG-based execution modes
with failure strategies, retries, and state persistence.
"""

import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.infra.redis_state import RedisStateManager
from orchestrata_ultimate.orchestrator.intent_lifecycle import IntentLifecycleManager
from orchestrata_ultimate.orchestrator.intent_types import Intent, IntentStatus

logger = logging.getLogger(__name__)


@traces(spec="003", story="US-008", reqs=["REQ-003-US-008-007"])
class WorkflowStatus(StrEnum):
    """Workflow execution status."""

    PENDING = "pending"
    RUNNING = "running"
    PAUSED = "paused"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"


@traces(spec="003", story="US-008", reqs=["REQ-003-US-008-004", "REQ-003-US-008-016"])
class FailureStrategy(StrEnum):
    """How to handle step failure."""

    RETRY = "retry"
    ROLLBACK = "rollback"
    CONTINUE = "continue"
    FAIL = "fail"


# Redis key prefixes
WORKFLOW_KEY = "orchestrata:workflow"
WORKFLOW_TTL = 86400  # 24 hours


@traces(spec="003", story="US-008", reqs=["REQ-003-US-008-001", "REQ-003-US-008-012"])
@dataclass
class WorkflowStep:
    """A single step in a workflow."""

    id: str
    intent: Intent
    depends_on: list[str] = field(default_factory=list)
    failure_strategy: FailureStrategy = FailureStrategy.FAIL
    max_retries: int = 3
    timeout_seconds: int = 300
    retry_count: int = 0

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-022"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "intent_id": self.intent.id,
            "depends_on": self.depends_on,
            "failure_strategy": self.failure_strategy.value,
            "max_retries": self.max_retries,
            "timeout_seconds": self.timeout_seconds,
            "retry_count": self.retry_count,
        }


@traces(spec="003", story="US-008", reqs=["REQ-003-US-008-001", "REQ-003-US-008-012"])
@dataclass
class WorkflowDef:
    """Workflow definition with steps and execution mode."""

    id: str
    steps: list[WorkflowStep]
    execution_mode: str = "sequential"  # "sequential", "parallel", "dag"
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-022"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "steps": [s.to_dict() for s in self.steps],
            "execution_mode": self.execution_mode,
            "created_at": self.created_at.isoformat(),
        }

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-012"])
    def get_step(self, step_id: str) -> WorkflowStep | None:
        for s in self.steps:
            if s.id == step_id:
                return s
        return None


@traces(spec="003", story="US-008", reqs=["REQ-003-US-008-010", "REQ-003-US-008-022"])
@dataclass
class WorkflowResult:
    """Result of workflow execution."""

    workflow_id: str
    status: WorkflowStatus
    step_results: dict[str, str] = field(default_factory=dict)
    started_at: datetime | None = None
    completed_at: datetime | None = None
    error: str | None = None
    duration_ms: float = 0.0

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-022"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "workflow_id": self.workflow_id,
            "status": self.status.value,
            "step_results": self.step_results,
            "started_at": self.started_at.isoformat() if self.started_at else None,
            "completed_at": (
                self.completed_at.isoformat() if self.completed_at else None
            ),
            "error": self.error,
            "duration_ms": self.duration_ms,
        }


@traces(
    spec="003",
    story="US-008",
    reqs=[
        "REQ-003-US-008-002",
        "REQ-003-US-008-003",
        "REQ-003-US-008-004",
        "REQ-003-US-008-005",
        "REQ-003-US-008-011",
    ],
)
class IntentWorkflow:
    """Workflow engine for multi-step intent execution.

    Manages workflow lifecycle, step ordering, failure handling,
    and state persistence via Redis.
    """

    def __init__(
        self,
        lifecycle: IntentLifecycleManager,
        redis: RedisStateManager,
    ) -> None:
        self._lifecycle = lifecycle
        self._redis = redis

    @property
    def client(self) -> Any:
        return self._redis.client

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-026"])
    async def define(
        self,
        steps: list[WorkflowStep],
        mode: str = "sequential",
    ) -> WorkflowDef:
        """Define a new workflow."""
        workflow = WorkflowDef(
            id=str(uuid.uuid4()),
            steps=steps,
            execution_mode=mode,
        )

        # Persist workflow definition
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow.id}:def",
            mapping={"data": json.dumps(workflow.to_dict())},
        )
        self.client.expire(f"{WORKFLOW_KEY}:{workflow.id}:def", WORKFLOW_TTL)

        # Initialize state
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow.id}:state",
            mapping={
                "status": WorkflowStatus.PENDING.value,
                "current_step": "",
                "started_at": "",
            },
        )
        self.client.expire(f"{WORKFLOW_KEY}:{workflow.id}:state", WORKFLOW_TTL)

        return workflow

    @traces(
        spec="003", story="US-008", reqs=["REQ-003-US-008-032", "REQ-003-US-008-035"]
    )
    async def execute(self, workflow_def: WorkflowDef) -> WorkflowResult:
        """Execute a workflow definition."""
        start_time = time.time()
        now = datetime.now(UTC)

        result = WorkflowResult(
            workflow_id=workflow_def.id,
            status=WorkflowStatus.RUNNING,
            started_at=now,
        )

        # Update state to RUNNING
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow_def.id}:state",
            mapping={
                "status": WorkflowStatus.RUNNING.value,
                "started_at": now.isoformat(),
            },
        )

        if workflow_def.execution_mode == "sequential":
            result = await self._execute_sequential(workflow_def, result)
        elif workflow_def.execution_mode == "parallel":
            result = await self._execute_parallel(workflow_def, result)
        elif workflow_def.execution_mode == "dag":
            result = await self._execute_dag(workflow_def, result)
        else:
            result.status = WorkflowStatus.FAILED
            result.error = f"Unknown execution mode: {workflow_def.execution_mode}"

        result.completed_at = datetime.now(UTC)
        result.duration_ms = (time.time() - start_time) * 1000

        # Persist final state
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow_def.id}:state",
            mapping={
                "status": result.status.value,
                "completed_at": result.completed_at.isoformat(),
                "duration_ms": str(result.duration_ms),
            },
        )

        return result

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-027"])
    async def _execute_sequential(
        self, workflow_def: WorkflowDef, result: WorkflowResult
    ) -> WorkflowResult:
        """Execute steps sequentially."""
        for step in workflow_def.steps:
            # Update current step
            self.client.hset(
                f"{WORKFLOW_KEY}:{workflow_def.id}:state",
                "current_step",
                step.id,
            )

            step_result = await self._execute_step(step)
            result.step_results[step.id] = step_result

            if step_result in (
                IntentStatus.FAILED.value,
                IntentStatus.TIMED_OUT.value,
            ):
                if step.failure_strategy == FailureStrategy.FAIL:
                    result.status = WorkflowStatus.FAILED
                    result.error = f"Step {step.id} failed"
                    return result
                elif step.failure_strategy == FailureStrategy.RETRY:
                    retried = await self._retry_step(step)
                    result.step_results[step.id] = retried
                    if retried in (
                        IntentStatus.FAILED.value,
                        IntentStatus.TIMED_OUT.value,
                    ):
                        result.status = WorkflowStatus.FAILED
                        result.error = f"Step {step.id} failed after retries"
                        return result
                elif step.failure_strategy == FailureStrategy.CONTINUE:
                    continue

        result.status = WorkflowStatus.COMPLETED
        return result

    @traces(
        spec="003",
        story="US-008",
        reqs=["REQ-003-US-008-003", "REQ-003-US-008-015", "REQ-003-US-008-028"],
    )
    async def _execute_parallel(
        self, workflow_def: WorkflowDef, result: WorkflowResult
    ) -> WorkflowResult:
        """Execute all steps (no dependency ordering)."""
        for step in workflow_def.steps:
            step_result = await self._execute_step(step)
            result.step_results[step.id] = step_result

        all_ok = all(
            v == IntentStatus.COMPLETED.value for v in result.step_results.values()
        )
        result.status = WorkflowStatus.COMPLETED if all_ok else WorkflowStatus.FAILED
        if not all_ok:
            failed = [
                k
                for k, v in result.step_results.items()
                if v != IntentStatus.COMPLETED.value
            ]
            result.error = f"Steps failed: {', '.join(failed)}"
        return result

    @traces(
        spec="003",
        story="US-008",
        reqs=[
            "REQ-003-US-008-002",
            "REQ-003-US-008-008",
            "REQ-003-US-008-013",
            "REQ-003-US-008-014",
            "REQ-003-US-008-020",
            "REQ-003-US-008-033",
        ],
    )
    async def _execute_dag(
        self, workflow_def: WorkflowDef, result: WorkflowResult
    ) -> WorkflowResult:
        """Execute steps respecting dependency graph."""
        completed_steps = set()
        remaining = list(workflow_def.steps)

        while remaining:
            ready = [
                s
                for s in remaining
                if all(dep in completed_steps for dep in s.depends_on)
            ]
            if not ready:
                result.status = WorkflowStatus.FAILED
                result.error = "Deadlock: no ready steps"
                return result

            for step in ready:
                step_result = await self._execute_step(step)
                result.step_results[step.id] = step_result
                if step_result == IntentStatus.COMPLETED.value:
                    completed_steps.add(step.id)
                elif step.failure_strategy == FailureStrategy.FAIL:
                    result.status = WorkflowStatus.FAILED
                    result.error = f"Step {step.id} failed"
                    return result
                elif step.failure_strategy == FailureStrategy.CONTINUE:
                    completed_steps.add(step.id)
                remaining.remove(step)

        result.status = WorkflowStatus.COMPLETED
        return result

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-014"])
    async def _execute_step(self, step: WorkflowStep) -> str:
        """Execute a single workflow step via the lifecycle manager."""
        try:
            self._lifecycle.store_intent(step.intent)
            self._lifecycle.transition(step.intent, IntentStatus.VALIDATED)
            self._lifecycle.transition(step.intent, IntentStatus.ROUTING)
            self._lifecycle.transition(step.intent, IntentStatus.DISPATCHED)
            self._lifecycle.transition(step.intent, IntentStatus.EXECUTING)
            self._lifecycle.transition(step.intent, IntentStatus.COMPLETED)
            return IntentStatus.COMPLETED.value
        except Exception as exc:
            try:
                self._lifecycle.transition(
                    step.intent, IntentStatus.FAILED, reason=str(exc)
                )
            except Exception as transition_exc:
                logger.warning(
                    "Failed to transition step intent %s to FAILED: %s",
                    getattr(step.intent, "id", "unknown"),
                    transition_exc,
                )
            return IntentStatus.FAILED.value

    @traces(
        spec="003",
        story="US-008",
        reqs=[
            "REQ-003-US-008-004",
            "REQ-003-US-008-009",
            "REQ-003-US-008-016",
            "REQ-003-US-008-021",
            "REQ-003-US-008-029",
            "REQ-003-US-008-034",
        ],
    )
    async def _retry_step(self, step: WorkflowStep) -> str:
        """Retry a failed step up to max_retries."""
        for attempt in range(step.max_retries):
            step.retry_count += 1
            result = await self._execute_step(step)
            if result == IntentStatus.COMPLETED.value:
                return result
        return IntentStatus.FAILED.value

    @traces(
        spec="003",
        story="US-008",
        reqs=["REQ-003-US-008-006", "REQ-003-US-008-018", "REQ-003-US-008-031"],
    )
    async def cancel(self, workflow_id: str) -> None:
        """Cancel a running workflow."""
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow_id}:state",
            "status",
            WorkflowStatus.CANCELLED.value,
        )

    @traces(
        spec="003",
        story="US-008",
        reqs=[
            "REQ-003-US-008-005",
            "REQ-003-US-008-007",
            "REQ-003-US-008-017",
            "REQ-003-US-008-019",
            "REQ-003-US-008-030",
        ],
    )
    async def get_status(self, workflow_id: str) -> WorkflowResult | None:
        """Get current workflow status."""
        raw = self.client.hgetall(f"{WORKFLOW_KEY}:{workflow_id}:state")
        if not raw:
            return None

        def decode(v: Any) -> str:
            return v.decode() if isinstance(v, bytes) else v

        data = {decode(k): decode(v) for k, v in raw.items()}
        status = WorkflowStatus(data.get("status", "pending"))

        return WorkflowResult(
            workflow_id=workflow_id,
            status=status,
            started_at=(
                datetime.fromisoformat(data["started_at"])
                if data.get("started_at")
                else None
            ),
            completed_at=(
                datetime.fromisoformat(data["completed_at"])
                if data.get("completed_at")
                else None
            ),
            duration_ms=float(data.get("duration_ms", 0)),
        )

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-016"])
    async def pause(self, workflow_id: str) -> None:
        """Pause a running workflow."""
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow_id}:state",
            "status",
            WorkflowStatus.PAUSED.value,
        )

    @traces(spec="001", story="US-006", reqs=["REQ-001-US-006-016"])
    async def resume(self, workflow_id: str) -> None:
        """Resume a paused workflow."""
        self.client.hset(
            f"{WORKFLOW_KEY}:{workflow_id}:state",
            "status",
            WorkflowStatus.RUNNING.value,
        )

    @traces(spec="003", story="US-008", reqs=["REQ-003-US-008-023"])
    async def list_workflows(self, status: WorkflowStatus | None = None) -> list[str]:
        """List workflow IDs. Scans Redis keys."""
        pattern = f"{WORKFLOW_KEY}:*:state"
        ids = []
        cursor = 0
        while True:
            cursor, keys = self.client.scan(cursor, match=pattern, count=100)
            for key in keys:
                key_str = key.decode() if isinstance(key, bytes) else key
                wf_id = key_str.split(":")[2]
                if status is not None:
                    raw_status = self.client.hget(key_str, "status")
                    if raw_status:
                        s = (
                            raw_status.decode()
                            if isinstance(raw_status, bytes)
                            else raw_status
                        )
                        if s != status.value:
                            continue
                ids.append(wf_id)
            if cursor == 0:
                break
        return ids