orchestrata_ultimate.agents.volk_registry

Reference › orchestrata_ultimate › agents › volk_registry

volk_registry is the kernel's registry of volks: who exists, what they can do, how to reach them and whether they are alive (specification 001, user story US-001). VolkRegistry is an IComponent. runtime uses it to resolve definitions, and so do the kernel dispatcher, router, CLI and status bridge, the orchestrator pipeline runner and the harness executor and healing modules. AgentRegistry is an alias subclass. The curated agents page gives the wider picture.

The data model is a set of dataclasses and enums: VolkDefinition (alias, name, role, capabilities, protocols, resource requirements, endpoint, metadata), VolkCapability, ResourceRequirements, VolkRegistration (definition plus token, status, registration time, last heartbeat and in-flight message count), VolkHealth, VolkMetadata, VolkQueryFilter, and the enums VolkProtocol (cxp, grpc, ws, http) and VolkStatus (registered, active, inactive, suspended, terminated).

register_volk validates the alias format, uniqueness and capability schema (raising InvalidAliasError, DuplicateAliasError or InvalidCapabilityError, all under VolkRegistrationError), creates the registration with a generated token, then runs the post-registration hooks: notification handlers, persistence to Redis and the optional registration callback. deregister_volk waits up to a graceful timeout for in-flight messages before removing the volk. Lookups are get_volk, is_registered, list_volks, list_volks_by_capability, list_volks_by_protocol and query_volks with filtering and pagination. Health is tracked with record_heartbeat, get_volk_health and the in-flight counters. A governance fact sink attached with set_governance mirrors registration and active status into the governance rules as facts, and lifecycle events are written to the audit trail on a best-effort basis that never blocks the caller.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkProtocol (enum)

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

Supported communication protocols for Volks.

GovernanceVolkFacts (protocol)

Bases: Protocol · Line 55

The slice of the governance client this registry needs.

VolkStatus (enum)

Bases: Enum · Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-018']) · Line 72

Status of a registered Volk.

ResourceRequirements (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-004']), dataclass · Line 84

Resource requirements for a Volk.

VolkCapability (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-002']), dataclass · Line 117

A capability that a Volk provides.

VolkDefinition (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-001', 'REQ-001-002', 'REQ-001-003', 'REQ-001-004']), dataclass · Line 158

Definition for registering a Volk with the kernel.

VolkHealth (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-018']), dataclass · Line 210

Health status of a Volk.

VolkMetadata (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-015']), dataclass · Line 236

Metadata about a registered Volk (REQ-001-015).

VolkQueryFilter (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-019']), dataclass · Line 267

Filter criteria for querying Volks (REQ-001-019).

VolkRegistration (dataclass)

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-007']), dataclass · Line 280

A registered Volk instance in the kernel.

VolkRegistrationError

Bases: Exception · Line 324

Error during Volk registration.

DuplicateAliasError

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

Alias already registered.

InvalidCapabilityError

Bases: VolkRegistrationError · Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-006']) · Line 338

Invalid capability schema.

InvalidAliasError

Bases: VolkRegistrationError · Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-001']) · Line 346

Invalid alias format.

VolkRegistry

Bases: IComponent · Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-001', 'REQ-001-011', 'REQ-001-021']) · Line 364

Registry for managing Volk registrations.

AgentRegistry

Bases: VolkRegistry · Line 1350

Registry alias for VolkRegistry.

Source

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

"""Volk Registration and Discovery for Orchestration Kernel.

This module implements Volk (agent) registration, validation, and discovery
as specified in US-001 of the Orchestration Kernel spec.

Requirements Implemented:
- REQ-001-001: Register with unique alias (U:Ox, U:PS, etc.)
- REQ-001-002: Include capabilities declaration
- REQ-001-003: Include supported protocols (CXP, gRPC, WS)
- REQ-001-004: Include resource requirements
- REQ-001-005: Validate alias uniqueness
- REQ-001-006: Validate capability schema
- REQ-001-007: Return registration token
- REQ-001-008: Update Volk registry
- REQ-001-009: Notify other Volks of new agent
- REQ-001-010: Persist to Redis state
"""

import hashlib
import logging
import secrets
import threading
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    ComponentMetadata,
    ComponentType,
    HealthStatus,
    IComponent,
)

logger = logging.getLogger(__name__)

# Volk (de)registration is an administrative/boot-time event, not scoped to
# any one conversation — there is no natural per-request session_id here, so
# audit entries use this fixed value (AuditEntry.session_id is required).
_AUDIT_SESSION_ID = "kernel-volk-registry"


@traces(spec="001", story="US-001", reqs=["REQ-001-003"])
class VolkProtocol(Enum):
    """Supported communication protocols for Volks."""

    CXP = "cxp"  # Cross-agent eXchange Protocol
    GRPC = "grpc"  # gRPC over HTTP/2
    WS = "ws"  # WebSocket
    HTTP = "http"  # HTTP REST (fallback)


class GovernanceVolkFacts(Protocol):
    """The slice of the governance client this registry needs.

    A Protocol rather than an import of ``GovernancePrologClient`` so the
    kernel registry does not take a hard dependency on the governance package
    (and its Prolog engine) just to be constructible. Every method is
    best-effort by contract and returns False rather than raising.
    """

    def assert_volk_operational(self, alias: str) -> bool: ...

    def retract_volk_operational(self, alias: str) -> bool: ...

    def set_volk_active(self, alias: str, active: bool) -> bool: ...


@traces(spec="001", story="US-001", reqs=["REQ-001-018"])
class VolkStatus(Enum):
    """Status of a registered Volk."""

    REGISTERED = "registered"  # Successfully registered
    ACTIVE = "active"  # Online and accepting messages
    INACTIVE = "inactive"  # Temporarily unavailable
    SUSPENDED = "suspended"  # Suspended by governance
    TERMINATED = "terminated"  # Deregistered


@traces(spec="001", story="US-001", reqs=["REQ-001-004"])
@dataclass
class ResourceRequirements:
    """Resource requirements for a Volk."""

    cpu_cores: float = 1.0  # CPU cores (can be fractional)
    memory_mb: int = 256  # Memory in megabytes
    storage_mb: int = 100  # Storage in megabytes
    network_bandwidth_mbps: int = 10  # Network bandwidth in Mbps
    max_concurrent_tasks: int = 10  # Max parallel task handling

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "cpu_cores": self.cpu_cores,
            "memory_mb": self.memory_mb,
            "storage_mb": self.storage_mb,
            "network_bandwidth_mbps": self.network_bandwidth_mbps,
            "max_concurrent_tasks": self.max_concurrent_tasks,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ResourceRequirements:
        """Create from dictionary."""
        return cls(
            cpu_cores=data.get("cpu_cores", 1.0),
            memory_mb=data.get("memory_mb", 256),
            storage_mb=data.get("storage_mb", 100),
            network_bandwidth_mbps=data.get("network_bandwidth_mbps", 10),
            max_concurrent_tasks=data.get("max_concurrent_tasks", 10),
        )


@traces(spec="001", story="US-001", reqs=["REQ-001-002"])
@dataclass
class VolkCapability:
    """A capability that a Volk provides."""

    name: str  # Capability name (e.g., "code_generation")
    version: str = "1.0.0"  # Capability version
    description: str = ""  # Human-readable description
    required_permissions: list[str] = field(
        default_factory=list
    )  # Required permissions

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "name": self.name,
            "version": self.version,
            "description": self.description,
            "required_permissions": self.required_permissions,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> VolkCapability:
        """Create from dictionary."""
        return cls(
            name=data["name"],
            version=data.get("version", "1.0.0"),
            description=data.get("description", ""),
            required_permissions=data.get("required_permissions", []),
        )


@traces(
    spec="001",
    story="US-001",
    reqs=[
        "REQ-001-001",
        "REQ-001-002",
        "REQ-001-003",
        "REQ-001-004",
    ],
)
@dataclass
class VolkDefinition:
    """Definition for registering a Volk with the kernel.

    This represents the complete registration request including
    alias, capabilities, protocols, and resource requirements.
    """

    alias: str  # Unique alias (e.g., U:Ox, U:PS)
    name: str  # Human-readable name
    role: str  # Role description (e.g., "orchestrator")
    capabilities: list[VolkCapability]  # Capabilities provided
    protocols: list[VolkProtocol]  # Supported protocols
    resource_requirements: ResourceRequirements = field(
        default_factory=ResourceRequirements
    )
    endpoint: str | None = None  # Primary endpoint (e.g., "grpc://localhost:50051")
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "alias": self.alias,
            "name": self.name,
            "role": self.role,
            "capabilities": [c.to_dict() for c in self.capabilities],
            "protocols": [p.value for p in self.protocols],
            "resource_requirements": self.resource_requirements.to_dict(),
            "endpoint": self.endpoint,
            "metadata": self.metadata,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> VolkDefinition:
        """Create from dictionary."""
        return cls(
            alias=data["alias"],
            name=data["name"],
            role=data["role"],
            capabilities=[
                VolkCapability.from_dict(c) for c in data.get("capabilities", [])
            ],
            protocols=[VolkProtocol(p) for p in data.get("protocols", ["cxp"])],
            resource_requirements=ResourceRequirements.from_dict(
                data.get("resource_requirements", {})
            ),
            endpoint=data.get("endpoint"),
            metadata=data.get("metadata", {}),
        )


@traces(spec="001", story="US-001", reqs=["REQ-001-018"])
@dataclass
class VolkHealth:
    """Health status of a Volk."""

    alias: str
    is_healthy: bool
    last_heartbeat: datetime | None = None
    heartbeat_age_seconds: float | None = None
    status: VolkStatus = VolkStatus.REGISTERED
    error_message: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "alias": self.alias,
            "is_healthy": self.is_healthy,
            "last_heartbeat": (
                self.last_heartbeat.isoformat() if self.last_heartbeat else None
            ),
            "heartbeat_age_seconds": self.heartbeat_age_seconds,
            "status": self.status.value,
            "error_message": self.error_message,
        }


@traces(spec="001", story="US-001", reqs=["REQ-001-015"])
@dataclass
class VolkMetadata:
    """Metadata about a registered Volk (REQ-001-015)."""

    alias: str
    name: str
    role: str
    capabilities: list[str]  # Capability names
    protocols: list[str]  # Protocol values
    status: str
    endpoint: str | None = None
    registered_at: str | None = None
    last_heartbeat: str | None = None

    @traces(spec="001", story="US-001", reqs=["REQ-001-015"])
    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "alias": self.alias,
            "name": self.name,
            "role": self.role,
            "capabilities": self.capabilities,
            "protocols": self.protocols,
            "status": self.status,
            "endpoint": self.endpoint,
            "registered_at": self.registered_at,
            "last_heartbeat": self.last_heartbeat,
        }


@traces(spec="001", story="US-001", reqs=["REQ-001-019"])
@dataclass
class VolkQueryFilter:
    """Filter criteria for querying Volks (REQ-001-019)."""

    capability: str | None = None
    protocol: str | None = None
    status: str | None = None
    role: str | None = None
    limit: int | None = None
    offset: int = 0


@traces(spec="001", story="US-001", reqs=["REQ-001-007"])
@dataclass
class VolkRegistration:
    """A registered Volk instance in the kernel.

    This extends VolkDefinition with registration-specific data
    like token, status, and timestamps.
    """

    definition: VolkDefinition
    token: str  # Registration token for authentication
    status: VolkStatus  # Current status
    registered_at: datetime  # Registration timestamp
    last_heartbeat: datetime | None = None
    in_flight_messages: int = 0  # REQ-001-025: Track in-flight messages

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization."""
        return {
            "definition": self.definition.to_dict(),
            "token": self.token,
            "status": self.status.value,
            "registered_at": self.registered_at.isoformat(),
            "last_heartbeat": (
                self.last_heartbeat.isoformat() if self.last_heartbeat else None
            ),
            "in_flight_messages": self.in_flight_messages,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> VolkRegistration:
        """Create from dictionary."""
        return cls(
            definition=VolkDefinition.from_dict(data["definition"]),
            token=data["token"],
            status=VolkStatus(data["status"]),
            registered_at=datetime.fromisoformat(data["registered_at"]),
            last_heartbeat=(
                datetime.fromisoformat(data["last_heartbeat"])
                if data.get("last_heartbeat")
                else None
            ),
            in_flight_messages=data.get("in_flight_messages", 0),
        )


class VolkRegistrationError(Exception):
    """Error during Volk registration."""


@traces(spec="001", story="US-001", reqs=["REQ-001-005"])
class DuplicateAliasError(VolkRegistrationError):
    """Alias already registered."""

    def __init__(self, alias: str) -> None:
        self.alias = alias
        super().__init__(f"Volk with alias '{alias}' is already registered")


@traces(spec="001", story="US-001", reqs=["REQ-001-006"])
class InvalidCapabilityError(VolkRegistrationError):
    """Invalid capability schema."""

    def __init__(self, message: str) -> None:
        super().__init__(f"Invalid capability: {message}")


@traces(spec="001", story="US-001", reqs=["REQ-001-001"])
class InvalidAliasError(VolkRegistrationError):
    """Invalid alias format."""

    def __init__(self, alias: str, reason: str) -> None:
        self.alias = alias
        self.reason = reason
        super().__init__(f"Invalid alias '{alias}': {reason}")


@traces(
    spec="001",
    story="US-001",
    reqs=[
        "REQ-001-001",
        "REQ-001-011",
        "REQ-001-021",
    ],
)
class VolkRegistry(IComponent):
    """Registry for managing Volk registrations.

    This class implements the core Volk registration and discovery
    functionality for the Orchestration Kernel.

    Thread-safe: Uses internal locking for concurrent access.
    """

    # Standard Volk alias prefixes
    VALID_PREFIXES = ("U:", "S:", "T:")  # User, System, Test

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-016", "REQ-001-018", "REQ-001-014"],
    )
    def __init__(
        self,
        redis_state: Any | None = None,  # RedisStateManager
        on_registration: Callable[[VolkRegistration], None] | None = None,
        cache_ttl_seconds: int = 300,  # REQ-001-016: Cache TTL
        heartbeat_timeout_seconds: int = 60,  # REQ-001-018: Heartbeat timeout
        # AuditLogger (kernel.audit.logger, L4) — Any because a TYPE_CHECKING
        # import NameErrors under PEP 649/749 lazy annotation evaluation, and
        # a runtime import would be an upward edge (agents L3 -> kernel L4).
        audit: Any | None = None,
        *,
        on_status_change: Callable[[str, VolkStatus, VolkStatus], None] | None = None,
        governance: GovernanceVolkFacts | None = None,
    ) -> None:
        """Initialize the Volk registry.

        Args:
            redis_state: Optional RedisStateManager for persistence.
            on_registration: Callback invoked when a Volk is registered.
            cache_ttl_seconds: Cache TTL for performance (REQ-001-016).
            heartbeat_timeout_seconds: Heartbeat timeout for health checks (REQ-001-018).
            audit: Optional SPEC-001 US-008 :class:`AuditLogger` for
                register/deregister audit trail (REQ-001-US-008-007). ``None``
                (default) constructs the process's default audit logger via
                :func:`~orchestrata_ultimate.kernel.audit_bootstrap.build_default_audit_logger`
                — disabled unless ``ORCHESTRATA_KERNEL_AUDIT_DISABLED`` is
                unset/false, so every caller (including
                ``orchestrator.pipeline_runner.PipelineRunner``'s own
                ``VolkRegistry()``, the gateway's live path) gets the audit
                trail without threading a shared instance through. Pass an
                explicit instance to share one, or ``AuditLogger`` configured
                for a test.
            on_status_change: Optional REQ-001-014 callback invoked as
                ``(alias, old_status, new_status)`` whenever a Volk's status
                genuinely transitions (heartbeat ACTIVE flip, lazy timeout
                INACTIVE flip). Keyword-only; mirrors ``on_registration``.
                Best-effort — see :meth:`_fire_status_change`.
            governance: Optional sink for the ``governance:volk_*`` Prolog
                facts that ``may_execute/2`` reads. Registration is what makes
                a volk KNOWN to the rule base, and since 2026-08-10 an unknown
                agent is denied rather than waved through, so this is the wire
                that lets governance answer at all. ``None`` (the default,
                and what every test that does not care passes) simply skips
                the sync: the registry is not the governance layer and must
                not require one to work.
        """
        self._volks: dict[str, VolkRegistration] = {}
        # Guards all mutations of _volks and the per-registration in-flight
        # message counters. Reentrant so a locked method may call another.
        self._lock = threading.RLock()
        self._redis_state = redis_state
        self._on_registration = on_registration
        self._on_status_change = on_status_change
        self._governance = governance
        self._notification_handlers: list[Callable[[str, VolkRegistration], None]] = []
        self._deregistration_handlers: list[Callable[[str], None]] = []  # REQ-001-022
        self._cache_ttl = cache_ttl_seconds
        self._heartbeat_timeout = heartbeat_timeout_seconds
        self._running = False
        if audit is not None:
            self.audit = audit
        else:
            # Lazy (RULED D7): kernel.audit_bootstrap is Layer 4 while this
            # registry lives in agents/ (Layer 3) — a module-level import
            # would be an upward layer violation.
            from orchestrata_ultimate.kernel.audit_bootstrap import (  # noqa: PLC0415
                build_default_audit_logger,
            )

            self.audit = build_default_audit_logger()
        self._metadata = ComponentMetadata(
            id="orchestrata-volk-registry",
            name="VolkRegistry",
            version="0.1.0",
            component_type=ComponentType.VOLK_RUNTIME,
            description="Volk registration and discovery",
            capabilities=["registration", "discovery", "heartbeat", "health"],
        )

        logger.info("VolkRegistry initialized")

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-007"])
    def _fire_volk_audit(
        self,
        alias: str,
        event: str,
        definition: VolkDefinition | None,
    ) -> None:
        """Fire a REQ-001-US-008-007 volk-lifecycle audit entry, never blocking or
        raising into register_volk/deregister_volk (see audit_bootstrap's
        fire_and_forget_audit — audit off, or no running event loop at the
        call site, means zero behavior change)."""
        if self.audit is None:
            return
        # Lazy (RULED D7): see __init__ — kernel.audit_bootstrap is Layer 4.
        from orchestrata_ultimate.kernel.audit_bootstrap import (  # noqa: PLC0415
            fire_and_forget_audit,
        )

        capabilities = (
            [c.name for c in definition.capabilities] if definition is not None else []
        )
        protocols = (
            [p.value for p in definition.protocols] if definition is not None else []
        )
        fire_and_forget_audit(
            self.audit.log_volk_lifecycle(
                session_id=_AUDIT_SESSION_ID,
                volk_id=alias,
                event=event,
                volk_alias=alias,
                capabilities=capabilities,
                protocols=protocols,
            )
        )

    @traces(spec="001", story="US-001", reqs=["REQ-001-014", "REQ-001-017"])
    def _fire_status_change(self, alias: str, old: VolkStatus, new: VolkStatus) -> None:
        """Fire a REQ-001-014 status-change notification, best-effort.

        Mirrors :meth:`_fire_volk_audit`: errors are logged, never
        propagated into ``record_heartbeat``/``get_volk_health``
        (REQ-001-017). Callers must invoke this AFTER releasing
        ``self._lock`` — the callback may emit kernel events and must
        never run while the registry lock is held.
        """
        # Liveness is not registration: a volk that stops answering its
        # heartbeat goes INACTIVE here, and governance has to hear about it or
        # the gate keeps dispatching to something that is not there.
        self._sync_governance_active(alias, new)

        if self._on_status_change is None:
            return
        try:
            self._on_status_change(alias, old, new)
        except Exception as e:
            logger.warning(
                f"on_status_change callback failed for {alias} "
                f"({old.value} -> {new.value}): {e}"
            )

    def set_governance(self, governance: GovernanceVolkFacts | None) -> None:
        """Attach (or clear) the governance fact sink AFTER construction.

        Exists because the two live runners build their registry before their
        Prolog gate — ``orchestrator.pipeline_runner`` constructs
        ``VolkRegistry()`` near the top of ``__init__`` and ``PrologGate``
        forty lines later — and reordering a constructor to satisfy a wiring
        concern is the tail wagging the dog.

        Attaching BACKFILLS every volk already registered. Without that, a
        caller who injects a populated registry (the documented test seam, and
        how ``PipelineRunner(volk_registry=...)`` is used) would leave those
        volks unknown to governance, and unknown means denied — the pipeline
        would refuse stages for volks that are demonstrably present.
        """
        self._governance = governance
        if governance is None:
            return
        with self._lock:
            aliases = list(self._volks)
        for alias in aliases:
            self._sync_governance_facts(alias, registered=True)

    @traces(spec="001", story="US-001", reqs=["REQ-001-008"])
    def _sync_governance_facts(self, alias: str, *, registered: bool) -> None:
        """Assert or retract *alias*'s ``governance:volk_*`` facts.

        Best-effort in the same sense as :meth:`_fire_volk_audit`: a rule base
        that is absent or broken must not fail a registration. The failure is
        logged and the volk stays unknown to governance — which, under the
        deny-unknown posture, is the safe direction.
        """
        if self._governance is None:
            return
        try:
            if registered:
                self._governance.assert_volk_operational(alias)
            else:
                self._governance.retract_volk_operational(alias)
        except Exception as e:
            logger.warning(f"governance volk-fact sync failed for {alias}: {e}")

    @traces(spec="001", story="US-001", reqs=["REQ-001-014"])
    def _sync_governance_active(self, alias: str, status: VolkStatus) -> None:
        """Mirror a status transition onto ``governance:volk_active/1``.

        REGISTERED and ACTIVE both count as active: a volk that has registered
        but not yet beaten is available for its first dispatch, which is the
        state every volk passes through on the way to doing any work at all.
        """
        if self._governance is None:
            return
        try:
            self._governance.set_volk_active(
                alias, status in (VolkStatus.ACTIVE, VolkStatus.REGISTERED)
            )
        except Exception as e:
            logger.warning(f"governance volk_active sync failed for {alias}: {e}")

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

    async def initialize(self, config: dict[str, Any]) -> None:
        """Initialize registry from config."""
        self._running = True

    async def start(self) -> None:
        """Start the registry."""
        self._running = True

    async def stop(self) -> None:
        """Stop the registry and clean up."""
        self._running = False

    async def get_health(self) -> HealthStatus:
        """Return registry health status."""
        return HealthStatus.HEALTHY if self._running else HealthStatus.STOPPED

    def get_metadata(self) -> ComponentMetadata:
        """Return registry component metadata."""
        return self._metadata

    @traces(spec="001", story="US-001", reqs=["REQ-001-007"])
    def _generate_token(self, alias: str) -> str:
        """Generate a secure registration token.

        Args:
            alias: Volk alias for token generation.

        Returns:
            Secure token string.
        """
        # Combine random bytes with alias hash for uniqueness
        random_part = secrets.token_hex(16)
        hash_part = hashlib.sha256(
            f"{alias}:{datetime.now(UTC).isoformat()}".encode()
        ).hexdigest()[:16]
        return f"volk_{random_part}_{hash_part}"

    @traces(spec="001", story="US-001", reqs=["REQ-001-001"])
    def _validate_alias(self, alias: str) -> None:
        """Validate alias format.

        Args:
            alias: Alias to validate.

        Raises:
            InvalidAliasError: If alias format is invalid.
        """
        if not alias:
            raise InvalidAliasError(alias, "Alias cannot be empty")

        if len(alias) < 2:
            raise InvalidAliasError(alias, "Alias must be at least 2 characters")

        if len(alias) > 32:
            raise InvalidAliasError(alias, "Alias must be at most 32 characters")

        # Check for valid prefix
        has_valid_prefix = any(
            alias.startswith(prefix) for prefix in self.VALID_PREFIXES
        )
        if not has_valid_prefix:
            raise InvalidAliasError(
                alias, f"Alias must start with one of: {', '.join(self.VALID_PREFIXES)}"
            )

    @traces(spec="001", story="US-001", reqs=["REQ-001-006"])
    def _validate_capabilities(self, capabilities: list[VolkCapability]) -> None:
        """Validate capability schema (REQ-001-006).

        Args:
            capabilities: List of capabilities to validate.

        Raises:
            InvalidCapabilityError: If any capability is invalid.
        """
        for cap in capabilities:
            if not cap.name:
                raise InvalidCapabilityError("Capability name cannot be empty")

            if not cap.name.replace("_", "").replace("-", "").isalnum():
                raise InvalidCapabilityError(
                    f"Capability name '{cap.name}' contains invalid characters"
                )

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-005", "REQ-001-006"],
    )
    def _validate_definition(self, definition: VolkDefinition) -> None:
        """Validate alias format, uniqueness, and capability schema.

        Args:
            definition: Volk definition to validate.

        Raises:
            InvalidAliasError: If alias format is invalid.
            DuplicateAliasError: If alias already registered (REQ-001-005).
            InvalidCapabilityError: If capability schema is invalid (REQ-001-006).
        """
        self._validate_alias(definition.alias)

        # REQ-001-005: Validate alias uniqueness
        if definition.alias in self._volks:
            logger.warning(
                f"Duplicate registration attempt for alias: {definition.alias}"
            )
            raise DuplicateAliasError(definition.alias)

        # REQ-001-006: Validate capability schema
        self._validate_capabilities(definition.capabilities)

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-007", "REQ-001-008"],
    )
    def _create_registration(self, definition: VolkDefinition) -> VolkRegistration:
        """Generate token, build VolkRegistration, and insert into the registry.

        Args:
            definition: Validated Volk definition.

        Returns:
            Newly created VolkRegistration stored in self._volks.
        """
        token = self._generate_token(definition.alias)
        registration = VolkRegistration(
            definition=definition,
            token=token,
            status=VolkStatus.REGISTERED,
            registered_at=datetime.now(UTC),
        )
        self._volks[definition.alias] = registration
        logger.info(
            f"Registered Volk: alias={definition.alias}, "
            f"capabilities={[c.name for c in definition.capabilities]}, "
            f"protocols={[p.value for p in definition.protocols]}"
        )
        return registration

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-009", "REQ-001-010"],
    )
    def _post_register_hooks(self, registration: VolkRegistration) -> None:
        """Run post-registration side effects: notifications, Redis persistence, callback.

        Args:
            registration: The newly created registration.
        """
        # REQ-001-009: Notify other Volks
        self._notify_registration(registration.definition.alias, registration)

        # REQ-001-010: Persist to Redis
        if self._redis_state is not None:
            self._persist_registration(registration)

        # Invoke registration callback
        if self._on_registration is not None:
            self._on_registration(registration)

        # Make the volk KNOWN to governance. `governance:may_execute/2` denies
        # any agent it has no volk_exists/1 fact for, so without this a
        # freshly-registered volk would be refused its own first dispatch.
        self._sync_governance_facts(registration.definition.alias, registered=True)

        # REQ-001-US-008-007: audit trail (best-effort, never blocks/raises)
        self._fire_volk_audit(
            registration.definition.alias, "register", registration.definition
        )

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-001", "REQ-001-008", "REQ-001-005", "REQ-001-018"],
    )
    def register_volk(self, definition: VolkDefinition) -> VolkRegistration:
        """Register a new Volk with the kernel.

        Implements:
        - REQ-001-001: Register with unique alias
        - REQ-001-002: Include capabilities declaration
        - REQ-001-003: Include supported protocols
        - REQ-001-004: Include resource requirements
        - REQ-001-005: Validate alias uniqueness
        - REQ-001-006: Validate capability schema
        - REQ-001-007: Return registration token
        - REQ-001-008: Update Volk registry
        - REQ-001-009: Notify other Volks
        - REQ-001-010: Persist to Redis

        Args:
            definition: Volk definition to register.

        Returns:
            VolkRegistration with token and status.

        Raises:
            DuplicateAliasError: If alias already registered (REQ-001-005).
            InvalidAliasError: If alias format is invalid.
            InvalidCapabilityError: If capability schema is invalid (REQ-001-006).
        """
        # Hold the lock across validation (which reads _volks for the
        # uniqueness check) and insertion so the TOCTOU window between the
        # duplicate-alias check and the dict write cannot be interleaved.
        with self._lock:
            self._validate_definition(definition)
            registration = self._create_registration(definition)
        self._post_register_hooks(registration)
        return registration

    @traces(spec="001", story="US-001", reqs=["REQ-001-009"])
    def _notify_registration(self, alias: str, registration: VolkRegistration) -> None:
        """Notify registered handlers of new Volk registration.

        Args:
            alias: Alias of newly registered Volk.
            registration: Registration details.
        """
        for handler in self._notification_handlers:
            try:
                handler(alias, registration)
            except Exception as e:
                logger.error(f"Error in notification handler: {e}")

    @traces(spec="001", story="US-001", reqs=["REQ-001-010"])
    def _persist_registration(self, registration: VolkRegistration) -> None:
        """Persist registration to Redis state.

        Args:
            registration: Registration to persist.
        """
        if self._redis_state is None:
            return

        try:
            key = f"volk:{registration.definition.alias}"
            self._redis_state.cache_set(
                key,
                registration.to_dict(),
                ttl_seconds=3600 * 24,  # 24 hours
            )
            logger.debug(
                f"Persisted registration to Redis: {registration.definition.alias}"
            )
        except Exception as e:
            logger.error(f"Failed to persist registration to Redis: {e}")

    def get_volk(self, alias: str) -> VolkRegistration | None:
        """Get a registered Volk by alias.

        Args:
            alias: Volk alias.

        Returns:
            VolkRegistration if found, None otherwise.
        """
        return self._volks.get(alias)

    def is_registered(self, alias: str) -> bool:
        """Check if a Volk is registered.

        Args:
            alias: Volk alias.

        Returns:
            True if registered, False otherwise.
        """
        return alias in self._volks

    @traces(spec="001", story="US-001", reqs=["REQ-001-011"])
    def list_volks(self) -> list[VolkRegistration]:
        """List all registered Volks.

        Returns:
            List of all registrations.
        """
        return list(self._volks.values())

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-012"],
    )
    def list_volks_by_capability(self, capability_name: str) -> list[VolkRegistration]:
        """List Volks that have a specific capability (REQ-001-012).

        Args:
            capability_name: Name of capability to search for.

        Returns:
            List of registrations with the capability.
        """
        return [
            reg
            for reg in self._volks.values()
            if any(cap.name == capability_name for cap in reg.definition.capabilities)
        ]

    @traces(spec="001", story="US-001", reqs=["REQ-001-013"])
    def list_volks_by_protocol(self, protocol: VolkProtocol) -> list[VolkRegistration]:
        """List Volks that support a specific protocol (REQ-001-013).

        Args:
            protocol: Protocol to search for.

        Returns:
            List of registrations supporting the protocol.
        """
        return [
            reg for reg in self._volks.values() if protocol in reg.definition.protocols
        ]

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-014"],
    )
    def add_notification_handler(
        self, handler: Callable[[str, VolkRegistration], None]
    ) -> None:
        """Add a handler for Volk registration notifications (REQ-001-014).

        Args:
            handler: Callback receiving (alias, registration) on new registrations.
        """
        self._notification_handlers.append(handler)

    @traces(spec="001", story="US-001", reqs=["REQ-001-025"])
    def _drain_in_flight_messages(
        self, alias: str, registration: VolkRegistration, graceful_timeout: float
    ) -> None:
        """Wait for in-flight messages to complete before deregistration.

        REQ-001-025: Deregistration handles in-flight messages

        Args:
            alias: Volk alias being deregistered.
            registration: Registration object tracking in-flight count.
            graceful_timeout: Maximum seconds to wait.
        """
        if registration.in_flight_messages <= 0:
            return

        logger.info(
            f"Waiting up to {graceful_timeout}s for {registration.in_flight_messages} "
            f"in-flight messages for {alias}"
        )
        import time

        start_time = datetime.now(UTC)
        while registration.in_flight_messages > 0:
            elapsed = (datetime.now(UTC) - start_time).total_seconds()
            if elapsed >= graceful_timeout:
                logger.warning(
                    f"Timeout waiting for in-flight messages. "
                    f"Forcing deregistration of {alias} with "
                    f"{registration.in_flight_messages} messages remaining"
                )
                break
            time.sleep(0.1)

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-022", "REQ-001-024"],
    )
    def _post_deregister_cleanup(self, alias: str) -> None:
        """Run post-deregistration side effects: notifications and Redis cleanup.

        Args:
            alias: Alias of the deregistered Volk.
        """
        # REQ-001-022: Notify dependent Volks
        self._notify_deregistration(alias)

        # REQ-001-024: Clean up resources from Redis
        if self._redis_state is not None:
            try:
                self._redis_state.cache_delete(f"volk:{alias}")
                logger.debug(f"Cleaned up Redis state for {alias}")
            except Exception as e:
                logger.error(f"Failed to delete registration from Redis: {e}")
                # Continue with deregistration despite Redis failure (REQ-001-017)

        # A terminated alias must stop satisfying volk_operational/1 at once;
        # leaving the facts behind would keep the gate open for something that
        # is no longer there.
        self._sync_governance_facts(alias, registered=False)

        # REQ-001-US-008-007: audit trail (best-effort, never blocks/raises)
        self._fire_volk_audit(alias, "deregister", None)

    @traces(
        spec="001",
        story="US-001",
        reqs=[
            "REQ-001-021",
            "REQ-001-023",
            "REQ-001-024",
        ],
    )
    def deregister_volk(self, alias: str, graceful_timeout: float = 5.0) -> bool:
        """Deregister a Volk from the kernel (REQ-001-021).

        Implements:
        - REQ-001-021: Volk can deregister cleanly
        - REQ-001-022: Deregistration notifies dependent Volks
        - REQ-001-023: Deregistration updates registry
        - REQ-001-024: Deregistration cleans up resources
        - REQ-001-025: Deregistration handles in-flight messages

        Args:
            alias: Volk alias to deregister.
            graceful_timeout: Seconds to wait for in-flight messages (REQ-001-025).

        Returns:
            True if deregistered, False if not found.
        """
        with self._lock:
            registration = self._volks.get(alias)
            if registration is None:
                return False

        # REQ-001-025: Wait for in-flight messages to complete. Done outside
        # the lock so other threads can still decrement the in-flight counter
        # (and so concurrent operations on other aliases are not blocked).
        self._drain_in_flight_messages(alias, registration, graceful_timeout)

        # REQ-001-023: Update registry. Re-acquire the lock and atomically
        # remove the alias; a concurrent deregistration may have won the race.
        with self._lock:
            registration = self._volks.pop(alias, None)
            if registration is None:
                return False
        registration.status = VolkStatus.TERMINATED
        logger.info(f"Deregistered Volk: {alias}")

        self._post_deregister_cleanup(alias)

        return True

    @traces(spec="001", story="US-001", reqs=["REQ-001-022"])
    def _notify_deregistration(self, alias: str) -> None:
        """Notify handlers of Volk deregistration (REQ-001-022).

        Args:
            alias: Alias of deregistered Volk.
        """
        for handler in self._deregistration_handlers:
            try:
                handler(alias)
            except Exception as e:
                # REQ-001-017: Handle partial failures gracefully
                logger.error(f"Error in deregistration handler for {alias}: {e}")

    @traces(spec="001", story="US-001", reqs=["REQ-001-022"])
    def add_deregistration_handler(self, handler: Callable[[str], None]) -> None:
        """Add a handler for Volk deregistration notifications (REQ-001-022).

        Args:
            handler: Callback receiving alias on deregistration.
        """
        self._deregistration_handlers.append(handler)

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-015", "REQ-001-016", "REQ-001-017"],
    )
    def get_volk_metadata(self, alias: str) -> VolkMetadata | None:
        """Get metadata for a Volk (REQ-001-015).

        Returns Volk metadata including alias, capabilities, and protocols.
        Uses caching for performance (REQ-001-016).

        Args:
            alias: Volk alias.

        Returns:
            VolkMetadata if found, None otherwise.
        """
        # REQ-001-016: Try cache first
        if self._redis_state is not None:
            try:
                cached = self._redis_state.cache_get(f"volk:metadata:{alias}")
                if cached is not None:
                    logger.debug(f"Cache hit for metadata: {alias}")
                    return VolkMetadata(**cached)
            except Exception as e:
                # REQ-001-017: Handle cache failure gracefully
                logger.warning(f"Cache lookup failed for {alias}: {e}")

        # Get from in-memory registry
        registration = self._volks.get(alias)
        if registration is None:
            return None

        metadata = VolkMetadata(
            alias=registration.definition.alias,
            name=registration.definition.name,
            role=registration.definition.role,
            capabilities=[c.name for c in registration.definition.capabilities],
            protocols=[p.value for p in registration.definition.protocols],
            status=registration.status.value,
            endpoint=registration.definition.endpoint,
            registered_at=registration.registered_at.isoformat(),
            last_heartbeat=(
                registration.last_heartbeat.isoformat()
                if registration.last_heartbeat
                else None
            ),
        )

        # REQ-001-016: Cache the result
        if self._redis_state is not None:
            try:
                self._redis_state.cache_set(
                    f"volk:metadata:{alias}",
                    metadata.to_dict(),
                    ttl_seconds=self._cache_ttl,
                )
                logger.debug(f"Cached metadata for {alias}")
            except Exception as e:
                # REQ-001-017: Handle cache failure gracefully
                logger.warning(f"Failed to cache metadata for {alias}: {e}")

        return metadata

    @traces(spec="001", story="US-001", reqs=["REQ-001-018"])
    def get_volk_health(self, alias: str) -> VolkHealth | None:
        """Get health status for a Volk (REQ-001-018).

        Health is determined by heartbeat age and Volk status. A Volk whose
        heartbeat has exceeded ``heartbeat_timeout_seconds`` is lazily marked
        INACTIVE, so the registry reflects stale peers in real time.

        Args:
            alias: Volk alias.

        Returns:
            VolkHealth if found, None otherwise.
        """
        # Transition is captured WHILE holding the lock, but
        # _fire_status_change is invoked AFTER the lock is released — the
        # callback may emit kernel events and must never run under the
        # registry lock (REQ-001-014, decision 3 of the volk-comms-heartbeat
        # design).
        transition: tuple[str, VolkStatus, VolkStatus] | None = None
        with self._lock:
            registration = self._volks.get(alias)
            if registration is None:
                return None

            now = datetime.now(UTC)
            is_healthy = True
            error_message = None
            heartbeat_age_seconds = None

            # Check heartbeat
            if registration.last_heartbeat is not None:
                heartbeat_age = now - registration.last_heartbeat
                heartbeat_age_seconds = heartbeat_age.total_seconds()

                if heartbeat_age_seconds > self._heartbeat_timeout:
                    is_healthy = False
                    if registration.status == VolkStatus.ACTIVE:
                        old_status = registration.status
                        registration.status = VolkStatus.INACTIVE
                        transition = (alias, old_status, VolkStatus.INACTIVE)
                        logger.warning(
                            f"Volk {alias} transitioned to INACTIVE due to heartbeat timeout"
                        )
                    error_message = f"Heartbeat timeout: {heartbeat_age_seconds:.1f}s > {self._heartbeat_timeout}s"

            # Check status
            if registration.status in (VolkStatus.SUSPENDED, VolkStatus.TERMINATED):
                is_healthy = False
                if error_message is None:
                    error_message = f"Status: {registration.status.value}"

            health = VolkHealth(
                alias=alias,
                is_healthy=is_healthy,
                last_heartbeat=registration.last_heartbeat,
                heartbeat_age_seconds=heartbeat_age_seconds,
                status=registration.status,
                error_message=error_message,
            )

        if transition is not None:
            self._fire_status_change(*transition)

        return health

    @traces(spec="001", story="US-001", reqs=["REQ-001-018", "REQ-001-017"])
    def record_heartbeat(self, alias: str) -> bool:
        """Record a heartbeat for a Volk (REQ-001-018).

        A fresh heartbeat transitions the Volk to ACTIVE status, making it
        eligible for health-aware routing.

        Args:
            alias: Volk alias.

        Returns:
            True if heartbeat recorded, False if Volk not found.
        """
        registration = self._volks.get(alias)
        if registration is None:
            return False

        # Transition is captured WHILE holding the lock, but
        # _fire_status_change is invoked AFTER the lock is released — see
        # get_volk_health for the same discipline.
        old_status: VolkStatus | None = None
        with self._lock:
            registration.last_heartbeat = datetime.now(UTC)
            if registration.status != VolkStatus.ACTIVE:
                old_status = registration.status
                registration.status = VolkStatus.ACTIVE
                logger.info(f"Volk {alias} transitioned to ACTIVE via heartbeat")
        logger.debug(f"Heartbeat recorded for {alias}")

        if old_status is not None:
            self._fire_status_change(alias, old_status, VolkStatus.ACTIVE)

        # Update in Redis if available
        if self._redis_state is not None:
            try:
                self._persist_registration(registration)
            except Exception as e:
                # REQ-001-017: Handle partial failures gracefully
                logger.warning(f"Failed to persist heartbeat for {alias}: {e}")

        return True

    @traces(spec="001", story="US-001", reqs=["REQ-001-025"])
    def increment_in_flight_messages(self, alias: str) -> bool:
        """Increment in-flight message count (REQ-001-025).

        Args:
            alias: Volk alias.

        Returns:
            True if incremented, False if Volk not found.
        """
        with self._lock:
            registration = self._volks.get(alias)
            if registration is None:
                return False

            registration.in_flight_messages += 1
        return True

    @traces(spec="001", story="US-001", reqs=["REQ-001-025"])
    def decrement_in_flight_messages(self, alias: str) -> bool:
        """Decrement in-flight message count (REQ-001-025).

        Args:
            alias: Volk alias.

        Returns:
            True if decremented, False if Volk not found.
        """
        with self._lock:
            registration = self._volks.get(alias)
            if registration is None:
                return False

            if registration.in_flight_messages > 0:
                registration.in_flight_messages -= 1
        return True

    @traces(
        spec="001",
        story="US-001",
        reqs=["REQ-001-019", "REQ-001-020"],
    )
    def query_volks(self, query_filter: VolkQueryFilter) -> list[VolkRegistration]:
        """Query Volks with filtering and pagination (REQ-001-019, REQ-001-020).

        Implements:
        - REQ-001-019: Support filtering and pagination
        - REQ-001-020: Validate query parameters

        Args:
            query_filter: Filter criteria for querying.

        Returns:
            List of matching registrations with pagination applied.

        Raises:
            ValueError: If query parameters are invalid (REQ-001-020).
        """
        # REQ-001-020: Validate query parameters
        if query_filter.offset < 0:
            raise ValueError("Offset must be non-negative")

        if query_filter.limit is not None and query_filter.limit <= 0:
            raise ValueError("Limit must be positive")

        results = self._apply_query_filters(list(self._volks.values()), query_filter)

        # REQ-001-019: Apply pagination
        if query_filter.offset > 0:
            results = results[query_filter.offset :]

        if query_filter.limit is not None:
            results = results[: query_filter.limit]

        return results

    @staticmethod
    @traces(spec="001", story="US-001", reqs=["REQ-001-019", "REQ-001-020"])
    def _apply_query_filters(
        results: list[VolkRegistration], query_filter: VolkQueryFilter
    ) -> list[VolkRegistration]:
        """Apply capability/protocol/status/role filters (REQ-001-019/020).

        Raises ValueError for invalid protocol/status enum values.
        """
        if query_filter.capability is not None:
            results = [
                reg
                for reg in results
                if any(
                    cap.name == query_filter.capability
                    for cap in reg.definition.capabilities
                )
            ]

        if query_filter.protocol is not None:
            try:
                protocol = VolkProtocol(query_filter.protocol)
            except ValueError:
                # REQ-001-020: Invalid protocol value
                raise ValueError(f"Invalid protocol: {query_filter.protocol}")
            results = [reg for reg in results if protocol in reg.definition.protocols]

        if query_filter.status is not None:
            try:
                status = VolkStatus(query_filter.status)
            except ValueError:
                # REQ-001-020: Invalid status value
                raise ValueError(f"Invalid status: {query_filter.status}")
            results = [reg for reg in results if reg.status == status]

        if query_filter.role is not None:
            results = [
                reg for reg in results if reg.definition.role == query_filter.role
            ]

        return results

    @property
    @traces(spec="001", story="US-001", reqs=["REQ-001-011"])
    def count(self) -> int:
        """Get number of registered Volks."""
        return len(self._volks)


# Class alias for orchestrata-ultimate naming
class AgentRegistry(VolkRegistry):
    """Registry alias for VolkRegistry."""

    pass