orchestrata_ultimate.agents.runtime

Reference › orchestrata_ultimate › agents › runtime

runtime is the instance lifecycle manager of the Multi-Agent Runtime (specification 005, user stories US-001 and US-004). VolkRuntime is an IComponent that spawns, tracks and terminates volk instances. It is built from a VolkRegistry (volk_registry), a MultiagentPrologBridge (prolog_bridge) and an optional RedisStateManager. pool, lifecycle and the task_executor depend on it. AgentRuntime is an alias subclass.

spawn runs the registration governance check (_check_registration_governance, which raises VolkSpawnError on denial), asserts the Prolog facts for the new volk (_assert_spawn_facts, _assert_running_facts), records a VolkInstance locally (_register_instance) and persists it to Redis when a state manager is present (_persist_instance). terminate checks governance, drains the instance (_drain_gracefully) and then cleans up the facts and local tracking (_finalize_termination); force_terminate takes an emergency reason. mark_busy, mark_ready and mark_failed change instance state for the executor. Query methods are get_instance, list_instances and get_instance_health, and instance_count and active_count are properties. As an IComponent it exposes initialize, start, stop, get_health and get_metadata.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkRuntime

Bases: IComponent · Decorators: traces(spec='005', story='US-001', reqs=['REQ-005-US-001-001', 'REQ-005-FR3-001', 'REQ-005-US-004-001']) · Line 54

Manages Volk instance lifecycle.

AgentRuntime

Bases: VolkRuntime · Line 441

Runtime alias for VolkRuntime.

Functions

Source

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

"""Volk Runtime — instance lifecycle management.

Implements IComponent with ComponentType.VOLK_RUNTIME. Manages spawning,
terminating, and tracking Volk instances with Prolog governance checks
and Redis state persistence.

Author: Mohamed
Spec: 005-multi-agent (amended)
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
import uuid
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from orchestrata_ultimate.agents.volk_registry import VolkRegistry
    from orchestrata_ultimate.infra.redis_state import RedisStateManager

from orchestrata_ultimate.agents.constants import REDIS_VOLK_INSTANCE_PREFIX
from orchestrata_ultimate.agents.exceptions import VolkSpawnError, VolkTerminateError
from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge
from orchestrata_ultimate.agents.types import InstanceStatus, VolkInstance
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    ComponentMetadata,
    ComponentType,
    HealthStatus,
    IComponent,
)

_logger = logging.getLogger(__name__)


@traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
def _utc_now() -> datetime:
    return datetime.now(UTC)


@traces(
    spec="005",
    story="US-001",
    reqs=[
        "REQ-005-US-001-001",
        "REQ-005-FR3-001",
        "REQ-005-US-004-001",
    ],
)
class VolkRuntime(IComponent):
    """Manages Volk instance lifecycle.

    Coordinates with VolkRegistry for agent definitions, Prolog governance
    for permission checks, and Redis for state persistence.

    Parameters
    ----------
    registry : VolkRegistry
        Registry for looking up Volk definitions.
    prolog : MultiagentPrologBridge
        Bridge to multiagent.pl governance rules.
    redis_state : RedisStateManager, optional
        Redis state manager for persistence.
    """

    def __init__(
        self,
        registry: VolkRegistry,
        prolog: MultiagentPrologBridge,
        redis_state: RedisStateManager | None = None,
    ) -> None:
        self._registry = registry
        self._prolog = prolog
        self._redis = redis_state
        self._instances: dict[str, VolkInstance] = {}
        self._alias_instances: dict[str, list[str]] = {}  # alias -> [instance_ids]
        self._initialized = False
        self._running = False

    # ================================================================
    # IComponent Implementation
    # ================================================================

    async def initialize(self, config: dict[str, Any]) -> None:
        self._initialized = True

    async def start(self) -> None:
        self._running = True

    @traces(spec="005", story="US-003", reqs=["REQ-005-NFR5-001"])
    async def stop(self) -> None:
        # Terminate all running instances
        for iid in list(self._instances.keys()):
            instance = self._instances[iid]
            if instance.is_active:
                try:
                    await self.terminate(iid)
                except Exception as exc:
                    _logger.warning(
                        "Failed to terminate instance '%s' during stop; forcing TERMINATED: %s",
                        iid,
                        exc,
                        exc_info=True,
                    )
                    instance.status = InstanceStatus.TERMINATED
        self._running = False

    @traces(
        spec="005",
        story="US-004",
        reqs=["REQ-005-US-004-001"],
    )
    async def get_health(self) -> HealthStatus:
        if not self._initialized:
            return HealthStatus.UNKNOWN
        if not self._running:
            return HealthStatus.STOPPED
        active = [i for i in self._instances.values() if i.is_active]
        failed = [
            i for i in self._instances.values() if i.status == InstanceStatus.FAILED
        ]
        if failed and not active:
            return HealthStatus.UNHEALTHY
        if failed:
            return HealthStatus.DEGRADED
        return HealthStatus.HEALTHY

    def get_metadata(self) -> ComponentMetadata:
        return ComponentMetadata(
            id="volk-runtime",
            name="VolkRuntime",
            version="1.0.0",
            component_type=ComponentType.VOLK_RUNTIME,
            description="Volk instance lifecycle management",
        )

    # ================================================================
    # Spawn
    # ================================================================

    @traces(
        spec="005",
        story="US-001",
        reqs=["REQ-005-FR4-001"],
    )
    def _check_registration_governance(
        self, instance_id: str, alias: str, definition: Any
    ) -> None:
        """Run Prolog registration check; raise VolkSpawnError on denial."""
        volk_type = definition.role or "worker"
        cpu = int(definition.resource_requirements.cpu_cores)
        memory = definition.resource_requirements.memory_mb
        if not self._prolog.can_register_volk(instance_id, volk_type, cpu, memory):
            raise VolkSpawnError(
                f"Prolog governance denied registration for '{alias}' "
                f"(type={volk_type}, cpu={cpu}, memory={memory})",
                alias=alias,
            )

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001"])
    def _assert_spawn_facts(self, instance_id: str, definition: Any) -> None:
        """Assert Prolog facts for a newly registered Volk."""
        self._prolog.assert_volk_registered(instance_id)
        self._prolog.assert_volk_resources_allocated(instance_id)
        for cap in definition.capabilities:
            self._prolog.assert_volk_capable(instance_id, cap.name)

    @traces(spec="005", story="US-001", reqs=["REQ-005-FR4-001"])
    def _assert_running_facts(self, instance_id: str, alias: str) -> None:
        """Assert Prolog running/available facts; raise VolkSpawnError if denied."""
        if not self._prolog.can_start_volk(instance_id):
            raise VolkSpawnError(
                f"Prolog governance denied start for '{alias}'",
                alias=alias,
            )
        self._prolog.assert_volk_running(instance_id)
        self._prolog.assert_volk_available(instance_id)
        self._prolog.assert_volk_no_pending_tasks(instance_id)

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001"])
    def _register_instance(
        self, instance_id: str, alias: str, metadata: dict
    ) -> VolkInstance:
        """Create and register the VolkInstance in local tracking."""
        instance = VolkInstance(
            instance_id=instance_id,
            alias=alias,
            status=InstanceStatus.READY,
            metadata=metadata,
        )
        self._instances[instance_id] = instance
        if alias not in self._alias_instances:
            self._alias_instances[alias] = []
        self._alias_instances[alias].append(instance_id)
        self._persist_instance(instance)
        return instance

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001", "REQ-005-FR1-001"])
    async def spawn(self, alias: str, **kwargs: Any) -> VolkInstance:
        """Spawn a new Volk instance.

        Flow:
        1. VolkRegistry.get_volk(alias) -- get registration
        2. prolog.can_register_volk() -- governance check
        3. Assert Prolog facts: registered, resources_allocated, capabilities
        4. prolog.can_start_volk() -- lifecycle check
        5. Assert volk_running
        6. Create VolkInstance, persist to Redis
        7. Return VolkInstance

        Raises
        ------
        VolkSpawnError
            If the alias is not registered, governance denies spawn,
            or any step in the spawn flow fails.
        """
        start_time = time.monotonic()

        registration = self._registry.get_volk(alias)
        if registration is None:
            raise VolkSpawnError(
                f"Alias '{alias}' not found in VolkRegistry", alias=alias
            )

        definition = registration.definition
        instance_id = kwargs.get("instance_id", str(uuid.uuid4()))

        self._check_registration_governance(instance_id, alias, definition)
        self._assert_spawn_facts(instance_id, definition)
        self._assert_running_facts(instance_id, alias)

        instance = self._register_instance(
            instance_id, alias, kwargs.get("metadata", {})
        )

        elapsed_ms = (time.monotonic() - start_time) * 1000
        _logger.info(
            f"Spawned instance {instance_id} for '{alias}' in {elapsed_ms:.1f}ms"
        )

        return instance

    # ================================================================
    # Terminate
    # ================================================================

    @traces(
        spec="005",
        story="US-003",
        reqs=["REQ-005-FR3-001", "REQ-005-NFR5-001"],
    )
    async def _drain_gracefully(self, instance_id: str, instance: VolkInstance) -> None:
        """Check governance, then drain the instance before termination."""
        self._prolog.assert_volk_no_pending_tasks(instance_id)
        if not self._prolog.can_stop_volk(instance_id):
            raise VolkTerminateError(
                f"Prolog governance denied stop for instance '{instance_id}'",
                instance_id=instance_id,
            )
        instance.status = InstanceStatus.DRAINING
        if instance.current_task:
            await asyncio.sleep(0.01)

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def _finalize_termination(self, instance_id: str, instance: VolkInstance) -> None:
        """Clean up Prolog facts, update state, and remove from alias tracking."""
        self._prolog.retract_volk_running(instance_id)
        self._prolog.retract_volk_available(instance_id)
        instance.status = InstanceStatus.TERMINATED
        instance.current_task = None
        alias_list = self._alias_instances.get(instance.alias, [])
        if instance_id in alias_list:
            alias_list.remove(instance_id)
        self._persist_instance(instance)

    @traces(
        spec="005",
        story="US-003",
        reqs=["REQ-005-AC5-001", "REQ-005-FR3-001"],
    )
    async def terminate(self, instance_id: str, force: bool = False) -> bool:
        """Terminate a Volk instance.

        Graceful: checks governance, drains tasks, then terminates.
        Force: bypasses task drain.
        """
        instance = self._instances.get(instance_id)
        if instance is None:
            raise VolkTerminateError(
                f"Instance '{instance_id}' not found",
                instance_id=instance_id,
            )

        if instance.status == InstanceStatus.TERMINATED:
            return True

        if not force:
            await self._drain_gracefully(instance_id, instance)

        self._finalize_termination(instance_id, instance)

        _logger.info(
            f"Terminated instance {instance_id} "
            f"(force={force}, tasks_completed={instance.tasks_completed})"
        )

        return True

    @traces(spec="005", story="US-003", reqs=["REQ-005-NFR5-001"])
    async def force_terminate(self, instance_id: str, reason: str) -> bool:
        """Force-terminate with an emergency reason."""
        instance = self._instances.get(instance_id)
        if instance is None:
            raise VolkTerminateError(
                f"Instance '{instance_id}' not found",
                instance_id=instance_id,
            )

        if not self._prolog.can_force_stop_volk(instance_id, reason):
            raise VolkTerminateError(
                f"Prolog governance denied force stop for '{instance_id}' "
                f"(reason={reason})",
                instance_id=instance_id,
            )

        return await self.terminate(instance_id, force=True)

    # ================================================================
    # Query
    # ================================================================

    async def get_instance(self, instance_id: str) -> VolkInstance | None:
        return self._instances.get(instance_id)

    async def list_instances(
        self,
        alias: str | None = None,
        status: InstanceStatus | None = None,
    ) -> list[VolkInstance]:
        instances = list(self._instances.values())
        if alias is not None:
            instances = [i for i in instances if i.alias == alias]
        if status is not None:
            instances = [i for i in instances if i.status == status]
        return instances

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    async def get_instance_health(self, instance_id: str) -> HealthStatus:
        instance = self._instances.get(instance_id)
        if instance is None:
            return HealthStatus.UNKNOWN

        if instance.status == InstanceStatus.TERMINATED:
            return HealthStatus.STOPPED
        if instance.status == InstanceStatus.FAILED:
            return HealthStatus.UNHEALTHY
        if instance.status == InstanceStatus.DRAINING:
            return HealthStatus.DEGRADED

        # Check Prolog health
        if self._prolog.volk_healthy(instance_id):
            return HealthStatus.HEALTHY

        return HealthStatus.DEGRADED

    # ================================================================
    # Instance State Management
    # ================================================================

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def mark_busy(self, instance_id: str, task_id: str) -> None:
        """Mark an instance as busy with a task."""
        instance = self._instances.get(instance_id)
        if instance is None:
            return
        instance.status = InstanceStatus.BUSY
        instance.current_task = task_id
        instance.last_activity = _utc_now()
        self._prolog.retract_volk_available(instance_id)
        self._prolog.assert_volk_busy(instance_id)
        self._prolog.retract_volk_no_pending_tasks(instance_id)

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def mark_ready(self, instance_id: str) -> None:
        """Mark an instance as ready (task completed)."""
        instance = self._instances.get(instance_id)
        if instance is None:
            return
        instance.status = InstanceStatus.READY
        instance.current_task = None
        instance.last_activity = _utc_now()
        self._prolog.retract_volk_busy(instance_id)
        self._prolog.assert_volk_available(instance_id)
        self._prolog.assert_volk_no_pending_tasks(instance_id)

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def mark_failed(self, instance_id: str) -> None:
        """Mark an instance as failed."""
        instance = self._instances.get(instance_id)
        if instance is None:
            return
        instance.status = InstanceStatus.FAILED
        instance.current_task = None
        self._prolog.retract_volk_running(instance_id)
        self._prolog.retract_volk_available(instance_id)
        self._prolog.assert_volk_failed(instance_id)

    # ================================================================
    # Persistence
    # ================================================================

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
    def _persist_instance(self, instance: VolkInstance) -> None:
        """Persist instance state to Redis if available."""
        if self._redis is None:
            return
        key = f"{REDIS_VOLK_INSTANCE_PREFIX}:{instance.instance_id}"
        try:
            self._redis.client.set(key, json.dumps(instance.to_dict()), ex=3600)
        except Exception as e:
            _logger.warning(f"Failed to persist instance {instance.instance_id}: {e}")

    @property
    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001"])
    def instance_count(self) -> int:
        """Total number of tracked instances (all statuses)."""
        return len(self._instances)

    @property
    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001"])
    def active_count(self) -> int:
        """Number of active instances (READY or BUSY)."""
        return sum(1 for i in self._instances.values() if i.is_active)


# Class alias for orchestrata-ultimate naming
class AgentRuntime(VolkRuntime):
    """Runtime alias for VolkRuntime."""

    pass