orchestrata_ultimate.agents.lifecycle

Reference › orchestrata_ultimate › agents › lifecycle

lifecycle gives agents a formal spawn, terminate and failure lifecycle with constitutional checks (specification 005, user story US-003). AgentLifecycle composes an optional VolkRuntime (runtime), ConstitutionalConstraints (constraints) and an AgentHierarchy (hierarchy).

spawn_with_validation validates an agent definition against the constraints, creates the instance through the runtime (or a minimal stub instance when no runtime is given, in _create_instance), adds it to the hierarchy under an optional parent, rolling back when that fails, and records a LifecycleState (_track_lifecycle_state). It returns None when validation fails. terminate_with_cleanup terminates and removes the agent, check_lifespan tests whether it is still valid, and handle_failure records the error while preserving context. LifespanType is one of task-bound, session-bound or persistent; LifecycleError reports failed operations. The module is not imported by other reference modules.

At a glance

Class diagram

Dependency diagram

API

Classes

LifespanType (enum)

Bases: StrEnum · Decorators: traces(spec='005', story='US-003', reqs=['REQ-005-US-003-001']) · Line 31

Agent lifespan types.

LifecycleState (dataclass)

Decorators: dataclass, traces(spec='005', story='US-003', reqs=['REQ-005-FR3-001']) · Line 41

Current lifecycle state of an agent.

LifecycleError

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

Lifecycle operation error.

AgentLifecycle

Decorators: traces(spec='005', story='US-003', reqs=['REQ-005-FR3-001', 'REQ-005-FR4-001', 'REQ-005-AC3-001', 'REQ-005-AC5-001', 'R… · Line 89

Manages formal lifecycle of agents with constitutional checks.

Source

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

# mypy: ignore-errors
"""Formal agent lifecycle with constitutional checks.

Manages spawn, terminate, and failure handling with
constitutional validation at each stage.
"""

# Spec: 005 - Multi-Agent System

from __future__ import annotations

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

from orchestrata_ultimate.agents.constraints import ConstitutionalConstraints
from orchestrata_ultimate.agents.hierarchy import AgentHierarchy
from orchestrata_ultimate.agents.types import InstanceStatus, VolkInstance
from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.agents.runtime import VolkRuntime

logger = logging.getLogger(__name__)


@traces(spec="005", story="US-003", reqs=["REQ-005-US-003-001"])
class LifespanType(StrEnum):
    """Agent lifespan types."""

    TASK_BOUND = "task_bound"
    SESSION_BOUND = "session_bound"
    PERSISTENT = "persistent"


@dataclass
@traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
class LifecycleState:
    """Current lifecycle state of an agent."""

    agent_id: str
    status: InstanceStatus
    spawned_at: datetime
    lifespan_type: LifespanType = LifespanType.SESSION_BOUND
    remaining_ttl: float | None = None
    error: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        return {
            "agent_id": self.agent_id,
            "status": self.status.value,
            "spawned_at": self.spawned_at.isoformat(),
            "lifespan_type": self.lifespan_type.value,
            "remaining_ttl": self.remaining_ttl,
            "error": self.error,
        }


@traces(
    spec="005",
    story="US-003",
    reqs=["REQ-005-FR3-001", "REQ-005-US-003-001"],
)
class LifecycleError(Exception):
    """Lifecycle operation error."""

    def __init__(self, message: str, agent_id: str = "") -> None:
        self.agent_id = agent_id
        super().__init__(message)


@traces(
    spec="005",
    story="US-003",
    reqs=[
        "REQ-005-FR3-001",
        "REQ-005-FR4-001",
        "REQ-005-AC3-001",
        "REQ-005-AC5-001",
        "REQ-005-US-003-001",
        "REQ-005-US-004-001",
        "REQ-005-AC6-001",
    ],
)
class AgentLifecycle:
    """Manages formal lifecycle of agents with constitutional checks.

    Lifecycle stages:
        1. Spawn: validate constraints → create instance → add to hierarchy
        2. Run: monitor lifespan, handle actions
        3. Terminate: cleanup → notify parent → remove from hierarchy
        4. Failure: preserve context → notify parent → cleanup
    """

    def __init__(
        self,
        runtime: VolkRuntime | None = None,
        constraints: ConstitutionalConstraints | None = None,
        hierarchy: AgentHierarchy | None = None,
    ) -> None:
        self._runtime = runtime
        self._constraints = constraints or ConstitutionalConstraints()
        self._hierarchy = hierarchy or AgentHierarchy()
        self._states: dict[str, LifecycleState] = {}
        self._spawn_count = 0
        self._terminate_count = 0

    @traces(spec="005", story="US-001", reqs=["REQ-005-FR1-001"])
    async def _create_instance(
        self, alias: str, definition: dict[str, Any]
    ) -> VolkInstance:
        """Spawn via runtime or create a minimal stub instance."""
        if self._runtime is not None:
            # Avoid duplicate alias keyword argument
            kwargs = {k: v for k, v in definition.items() if k != "alias"}
            return await self._runtime.spawn(alias, **kwargs)
        now = datetime.now(UTC)
        return VolkInstance(
            alias=alias,
            status=InstanceStatus.READY,
            spawned_at=now,
            last_activity=now,
            metadata=definition.get("metadata", {}),
        )

    @traces(spec="005", story="US-003", reqs=["REQ-005-FR5-001"])
    async def _add_to_hierarchy(
        self, instance: VolkInstance, alias: str, parent_id: str | None
    ) -> None:
        """Add the instance to the agent hierarchy, rolling back on failure."""
        if parent_id:
            try:
                self._hierarchy.add_child(parent_id, instance.instance_id, role=alias)
            except Exception as exc:
                if self._runtime is not None:
                    await self._runtime.terminate(instance.instance_id)
                raise LifecycleError(str(exc), agent_id=instance.instance_id)
        else:
            self._hierarchy.add_root(instance.instance_id, role=alias)

    @traces(spec="005", story="US-003", reqs=["REQ-005-US-003-001"])
    def _track_lifecycle_state(
        self, instance: VolkInstance, definition: dict[str, Any]
    ) -> None:
        """Record the initial LifecycleState for a newly spawned instance."""
        lifespan = LifespanType(definition.get("lifespan", "session_bound"))
        self._states[instance.instance_id] = LifecycleState(
            agent_id=instance.instance_id,
            status=instance.status,
            spawned_at=instance.spawned_at,
            lifespan_type=lifespan,
            remaining_ttl=definition.get("ttl_seconds"),
        )

    @traces(
        spec="005",
        story="US-001",
        reqs=["REQ-005-FR4-001", "REQ-005-FR1-001", "REQ-005-AC1-001"],
    )
    async def spawn_with_validation(
        self,
        definition: dict[str, Any],
        parent_id: str | None = None,
    ) -> VolkInstance | None:
        """Spawn an agent after constitutional validation."""
        start = time.perf_counter()

        # Validate constraints
        result = self._constraints.validate_at_spawn(definition)
        if not result.valid:
            raise LifecycleError(
                f"Spawn rejected: {', '.join(result.violations)}",
                agent_id=definition.get("alias", ""),
            )

        alias = definition.get("alias", "")
        instance = await self._create_instance(alias, definition)
        await self._add_to_hierarchy(instance, alias, parent_id)
        self._track_lifecycle_state(instance, definition)

        elapsed = (time.perf_counter() - start) * 1000
        logger.info(f"Agent {alias} spawned in {elapsed:.1f}ms")
        self._spawn_count += 1
        return instance

    @traces(
        spec="005",
        story="US-003",
        reqs=[
            "REQ-005-AC5-001",
            "REQ-005-FR3-001",
            "REQ-005-NFR5-001",
            "REQ-005-US-003-001",
        ],
    )
    async def terminate_with_cleanup(self, agent_id: str) -> bool:
        """Terminate an agent with full cleanup."""
        start = time.perf_counter()

        # Notify parent before termination
        self._hierarchy.get_parent(agent_id)

        # Terminate children first (bottom-up)
        children = self._hierarchy.get_children(agent_id)
        for child_id in children:
            await self.terminate_with_cleanup(child_id)

        # Terminate via runtime
        if self._runtime is not None:
            try:
                await self._runtime.terminate(agent_id)
            except Exception as exc:
                logger.warning(f"Runtime terminate failed for {agent_id}: {exc}")

        # Remove from hierarchy
        self._hierarchy.remove_agent(agent_id)

        # Clean up state
        if agent_id in self._states:
            del self._states[agent_id]

        elapsed = (time.perf_counter() - start) * 1000
        logger.info(f"Agent {agent_id} terminated in {elapsed:.1f}ms")
        self._terminate_count += 1
        return True

    @traces(spec="005", story="US-003", reqs=["REQ-005-US-003-001"])
    def check_lifespan(self, agent_id: str) -> bool:
        """Check if agent's lifespan is still valid."""
        state = self._states.get(agent_id)
        if state is None:
            return False
        if state.remaining_ttl is not None:
            elapsed = (datetime.now(UTC) - state.spawned_at).total_seconds()
            if elapsed > state.remaining_ttl:
                return False
        return True

    @traces(spec="005", story="US-002", reqs=["REQ-005-FR7-001"])
    async def handle_failure(self, agent_id: str, error: str) -> None:
        """Handle agent failure with context preservation."""
        state = self._states.get(agent_id)
        if state is not None:
            state.status = InstanceStatus.FAILED
            state.error = error

        # Preserve error context (Article VII)
        logger.error(f"Agent {agent_id} failed: {error}")

        # Notify parent
        parent_id = self._hierarchy.get_parent(agent_id)
        if parent_id:
            logger.info(f"Notifying parent {parent_id} of failure for {agent_id}")

    def get_lifecycle_state(self, agent_id: str) -> LifecycleState | None:
        """Get current lifecycle state."""
        return self._states.get(agent_id)

    def get_all_states(self) -> dict[str, LifecycleState]:
        """Get all lifecycle states."""
        return dict(self._states)

    @property
    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001"])
    def active_count(self) -> int:
        return sum(
            1
            for s in self._states.values()
            if s.status in (InstanceStatus.READY, InstanceStatus.BUSY)
        )

    @property
    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001", "REQ-005-FR8-001"])
    def spawn_count(self) -> int:
        return self._spawn_count

    @property
    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001"])
    def terminate_count(self) -> int:
        return self._terminate_count