orchestrata_ultimate.orchestrator.intent_lifecycle

Reference › orchestrata_ultimate › orchestrator › intent_lifecycle

The state machine that keeps an intent's status honest, backed by Redis. intent_broker moves every intent through it, intent_broker_factory constructs it, and intent_workflow uses it for multi-step runs. The statuses come from IntentStatus in intent_types.

The VALID_TRANSITIONS table lists, for each status, the statuses that may follow it. IntentLifecycleManager.transition refuses any other move by raising InvalidTransitionError; otherwise it updates the status and timestamp, stores the failure reason as the error message when the new status is FAILED, writes the intent to Redis and records the transition. can_transition answers the same question without acting. store_intent saves a new intent, get_intent, get_history and get_active_intents read back, cleanup_expired removes expired intents and returns how many, and close releases the connection.

The constructor creates a synchronous Redis client through core.redis_factory, using the resolved host and port unless given, and pings it, so a missing Redis fails at construction. Each intent is stored as JSON under a key with a class prefix, with an expiry of twice the intent's ttl_seconds (at least one second).

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

InvalidTransitionError

Bases: Exception · Decorators: traces(spec='003', story='US-005', reqs=['REQ-003-US-005-003', 'REQ-003-US-005-031']) · Line 48

Raised when an invalid state transition is attempted.

IntentLifecycleManager

Decorators: traces(spec='003', story='US-005', reqs=['REQ-003-US-005-015', 'REQ-003-US-005-018', 'REQ-003-US-005-019']) · Line 68

Manages intent state transitions with Redis persistence.

Source

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

# mypy: ignore-errors
"""Intent lifecycle state machine.

Manages intent state transitions with validation and persistence.

Spec: 003-intent / US-005
Author: Mohamed
"""

import json
import logging
from datetime import UTC, datetime
from typing import cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.orchestrator.intent_types import Intent, IntentStatus

logger = logging.getLogger(__name__)

# Valid state transitions
VALID_TRANSITIONS: dict[IntentStatus, set[IntentStatus]] = {
    IntentStatus.PENDING: {
        IntentStatus.VALIDATED,
        IntentStatus.REJECTED,
        IntentStatus.CANCELLED,
    },
    IntentStatus.VALIDATED: {IntentStatus.ROUTING, IntentStatus.CANCELLED},
    IntentStatus.REJECTED: set(),  # terminal
    IntentStatus.ROUTING: {IntentStatus.DISPATCHED, IntentStatus.FAILED},
    IntentStatus.DISPATCHED: {
        IntentStatus.EXECUTING,
        IntentStatus.FAILED,
        IntentStatus.TIMED_OUT,
    },
    IntentStatus.EXECUTING: {
        IntentStatus.COMPLETED,
        IntentStatus.FAILED,
        IntentStatus.TIMED_OUT,
    },
    IntentStatus.COMPLETED: set(),  # terminal
    IntentStatus.FAILED: set(),  # terminal
    IntentStatus.CANCELLED: set(),  # terminal
    IntentStatus.TIMED_OUT: set(),  # terminal
}


@traces(spec="003", story="US-005", reqs=["REQ-003-US-005-003", "REQ-003-US-005-031"])
class InvalidTransitionError(Exception):
    """Raised when an invalid state transition is attempted."""

    def __init__(
        self, intent_id: str, current: IntentStatus, target: IntentStatus
    ) -> None:
        self.intent_id = intent_id
        self.current = current
        self.target = target
        super().__init__(
            f"Invalid transition for intent {intent_id}: "
            f"{current.value} -> {target.value}"
        )


@traces(
    spec="003",
    story="US-005",
    reqs=["REQ-003-US-005-015", "REQ-003-US-005-018", "REQ-003-US-005-019"],
)
class IntentLifecycleManager:
    """Manages intent state transitions with Redis persistence.

    State machine enforces valid transitions and tracks history.
    All state changes are persisted to Redis for durability.
    """

    # Explicitly typed `str` (not left to infer as a narrow Literal) so
    # tests may override these per-instance for Redis-key isolation
    # (test_id-prefixed keys) without a pyright reportAttributeAccessIssue.
    KEY_PREFIX: str = "orchestrata:intent:"
    HISTORY_PREFIX: str = "orchestrata:intent:history:"

    def __init__(
        self,
        redis_host: str | None = None,
        redis_port: int | None = None,
        redis_db: int = 0,
    ) -> None:
        from orchestrata_ultimate.core.redis_factory import (
            create_sync_redis,
            resolve_redis_host,
            resolve_redis_port,
        )

        self._redis = create_sync_redis(
            host=redis_host if redis_host is not None else resolve_redis_host(),
            port=redis_port if redis_port is not None else resolve_redis_port(),
            db=redis_db,
        )
        self._redis.ping()  # fail fast if Redis is down

    @traces(
        spec="003",
        story="US-005",
        reqs=["REQ-003-US-005-003", "REQ-003-US-005-014", "REQ-003-US-005-031"],
    )
    def can_transition(self, current: IntentStatus, target: IntentStatus) -> bool:
        """Check if a state transition is valid."""
        return target in VALID_TRANSITIONS.get(current, set())

    @traces(
        spec="003",
        story="US-005",
        reqs=[
            "REQ-003-US-005-002",
            "REQ-003-US-005-003",
            "REQ-003-US-005-006",
            "REQ-003-US-005-016",
        ],
    )
    def transition(
        self, intent: Intent, new_status: IntentStatus, reason: str = ""
    ) -> Intent:
        """Transition an intent to a new status.

        Validates the transition, updates the intent, persists to Redis,
        and records the transition in history.

        Raises InvalidTransitionError if the transition is not valid.
        """
        if not self.can_transition(intent.status, new_status):
            raise InvalidTransitionError(intent.id, intent.status, new_status)

        old_status = intent.status
        intent.status = new_status
        intent.updated_at = datetime.now(UTC)

        if new_status == IntentStatus.FAILED and reason:
            intent.error_message = reason

        # Persist to Redis
        self._persist_intent(intent)
        self._record_transition(intent.id, old_status, new_status, reason)

        logger.info(
            f"Intent {intent.id}: {old_status.value} -> {new_status.value}"
            + (f" ({reason})" if reason else "")
        )
        return intent

    @traces(spec="003", story="US-005", reqs=["REQ-003-US-005-019"])
    def get_intent(self, intent_id: str) -> Intent | None:
        """Retrieve an intent from Redis."""
        key = f"{self.KEY_PREFIX}{intent_id}"
        # redis-py's command mixins are shared between the sync and async
        # clients, so `get()` is stub-typed `Awaitable[Any] | Any`;
        # `self._redis` is always the sync client here.
        data = cast("str | None", self._redis.get(key))
        if data is None:
            return None
        return Intent.from_dict(json.loads(data))

    def store_intent(self, intent: Intent) -> None:
        """Store an intent in Redis."""
        self._persist_intent(intent)

    @traces(
        spec="003",
        story="US-005",
        reqs=[
            "REQ-003-US-005-005",
            "REQ-003-US-005-010",
            "REQ-003-US-005-019",
            "REQ-003-US-005-025",
        ],
    )
    def get_history(self, intent_id: str) -> list[dict]:
        """Get transition history for an intent."""
        key = f"{self.HISTORY_PREFIX}{intent_id}"
        # Stub deficiency (see get_intent).
        entries = cast("list[str]", self._redis.lrange(key, 0, -1))
        return [json.loads(e) for e in entries]

    @traces(
        spec="003",
        story="US-005",
        reqs=["REQ-003-US-005-008", "REQ-003-US-005-019", "REQ-003-US-005-034"],
    )
    def get_active_intents(self) -> list[Intent]:
        """Get all non-terminal intents."""
        active = []
        pattern = f"{self.KEY_PREFIX}*"
        for key in self._redis.scan_iter(match=pattern):
            if ":history:" in key:
                continue
            # Stub deficiency (see get_intent).
            data = cast("str | None", self._redis.get(key))
            if data:
                intent = Intent.from_dict(json.loads(data))
                if not intent.is_terminal:
                    active.append(intent)
        return active

    def cleanup_expired(self) -> int:
        """Clean up expired intents by transitioning them to TIMED_OUT."""
        count = 0
        for intent in self.get_active_intents():
            if intent.is_expired and self.can_transition(
                intent.status, IntentStatus.TIMED_OUT
            ):
                self.transition(intent, IntentStatus.TIMED_OUT, reason="TTL expired")
                count += 1
        return count

    def close(self) -> None:
        """Close the Redis connection."""
        self._redis.close()

    @traces(spec="003", story="US-005", reqs=["REQ-003-US-005-002"])
    def _persist_intent(self, intent: Intent) -> None:
        """Persist intent state to Redis with TTL."""
        key = f"{self.KEY_PREFIX}{intent.id}"
        # Redis rejects EX 0 / negative; clamp so ttl_seconds=0 still persists.
        self._redis.set(
            key,
            json.dumps(intent.to_dict()),
            ex=max(1, intent.ttl_seconds * 2),
        )

    @traces(
        spec="003",
        story="US-005",
        reqs=["REQ-003-US-005-002", "REQ-003-US-005-018", "REQ-003-US-005-033"],
    )
    def _record_transition(
        self,
        intent_id: str,
        old_status: IntentStatus,
        new_status: IntentStatus,
        reason: str,
    ) -> None:
        """Record a state transition in the history list."""
        key = f"{self.HISTORY_PREFIX}{intent_id}"
        entry = {
            "from": old_status.value,
            "to": new_status.value,
            "reason": reason,
            "timestamp": datetime.now(UTC).isoformat(),
        }
        self._redis.rpush(key, json.dumps(entry))
        # Keep history for 24 hours
        self._redis.expire(key, 86400)