orchestrata_ultimate.kernel.dispatcher

Reference › orchestrata_ultimate › kernel › dispatcher

The delivery step of the kernel's message plane. Where router decides who receives a packet, MessageDispatcher sends the CXPPacket to one recipient VolkDefinition over the protocol that volk declares, counts the attempts and returns a DispatchResult. It serves SPEC-001 stories US-002 and US-003. The harness modules harness.executor, harness.healing and harness.runner import it. The curated kernel page places it in the message plane.

• dispatch first runs two audit-only policy checks, then selects a protocol and loops over attempts. _select_protocol prefers CXP, then gRPC, then WebSocket, and falls back to HTTP. _dispatch_by_protocol raises when the recipient has no endpoint; it sends CXP and gRPC through dispatch_grpc, and the others through dispatch_websocket and dispatch_http.
• _attempt_dispatch runs one attempt under asyncio.wait_for with timeout_seconds (30 by default) and records a DispatchAttempt as success, failed or timeout. After a failure dispatch waits base_retry_delay seconds doubled per attempt and tries again, up to max_retries (3 by default), then reports that it failed after all attempts. Every result is kept and can be read with get_dispatch_history.
• dispatch_grpc parses the endpoint with _parse_grpc_endpoint, builds a CXP client from the shared cxp package and has _send_grpc_packet connect, send and read the acknowledgement. The endpoint parser strips a grpc:// or cxp:// prefix and, when no port is given, takes the port from CXP_GRPC_PORT through cxp_grpc_port(), raising when it is unset; the name is historical, because the gRPC retirement note records that the cxpt direct transport replaced grpcio and only the key name kept its prefix. dispatch_websocket and dispatch_http send the packet serialised by _serialize_packet, and _parse_ws_ack reads the WebSocket acknowledgement.

When policy_audit_mode is on (it is off by default), the constructor builds a PolicyEvaluator and a ConstitutionalEnforcer over a real Prolog bridge, loading the governance rule files through the governance prolog client. _audit_evaluate_routing logs the routing verdict for sender, recipient and message type, and _audit_evaluate_plan_fidelity logs the Article XII verdict. Both only log and never change the delivery outcome; any failure inside them is logged as a warning, and if the builders fail the dispatcher starts without audit.

At a glance

Class diagram

Dependency diagram

API

Classes

DispatchStatus (enum)

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

Status of dispatch operation.

DispatchAttempt (dataclass)

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

Record of a dispatch attempt.

DispatchResult (dataclass)

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

Result of message dispatch.

MessageDispatcher

Decorators: traces(spec='001', story='US-002', reqs=['REQ-001-US-002-011', 'REQ-001-US-003-001', 'REQ-001-US-010-015']) · Line 295

Dispatches CXP messages via various protocols.

Functions

Source

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

# mypy: ignore-errors
"""CXP Message Dispatcher for Orchestration Kernel.

This module implements protocol-specific message dispatch (gRPC, WebSocket, HTTP)
with retry logic and delivery tracking as specified in Stream 2A.

Requirements Implemented:
- REQ-001-US-002-011 to REQ-001-US-002-020: Message dispatch across protocols
- REQ-001-US-003-001: Routing-policy audit logging (staged, non-blocking; see
  ``policy_audit_mode`` on MessageDispatcher and the policy_evaluator.py
  module docstring for the 2026-07-06 wiring note)
"""

import asyncio
import json
import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    # Real module types for the type checker -- see the else branch for the
    # actual optional-import fallback used at runtime. Without this split,
    # every aiohttp/websockets attribute access below is flagged against the
    # `None` branch of the runtime try/except (reportOptionalMemberAccess).
    import aiohttp
    import websockets
    import websockets.exceptions
else:
    try:
        import aiohttp
    except ImportError:
        aiohttp = None  # Optional dependency for HTTP dispatch

    try:
        import websockets
        import websockets.exceptions
    except ImportError:
        websockets = None  # Optional dependency for WebSocket dispatch


from collections.abc import Mapping

from orchestrata_ultimate.agents.volk_registry import VolkDefinition, VolkProtocol
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.core.paths import get_prolog_dir
from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_port
from orchestrata_ultimate.cxp.proto_codec import CXPClientConfig
from orchestrata_ultimate.cxp.transport import CXPClientLike, create_cxp_client
from orchestrata_ultimate.governance._strict import governance_strict
from orchestrata_ultimate.governance.constitutional import ConstitutionalEnforcer
from orchestrata_ultimate.governance.prolog_client import (
    GOVERNANCE_RULE_FILES,
    GovernancePrologClient,
    count_governance_predicates,
    load_governance_rules,
)
from orchestrata_ultimate.infra.prolog_bridge import PrologBridge
from orchestrata_ultimate.kernel.policy_evaluator import PolicyEvaluator
from orchestrata_ultimate.planning.plan_parser import (
    PlanGovernanceFacts,
    load_plan_governance_facts,
)

logger = logging.getLogger(__name__)


@traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
def _build_default_policy_evaluator() -> PolicyEvaluator | None:
    """Construct a PolicyEvaluator backed by a real, process-wide PrologBridge.

    Used by MessageDispatcher when ``policy_audit_mode=True`` and no
    ``policy_evaluator`` was injected by the caller. Mirrors the production
    boot sequence in kernel/cli_main.py (PrologBridge over the prolog/
    rules directory). Loads the COMPLETE governance rule corpus via the one
    blessed loader (``governance.prolog_client.load_governance_rules``) --
    previously this hand-loaded only ``governance.pl``, so ``governance.pl``'s
    ``check_constraint/4`` (ownership) delegation to ``pam:pam_owns/2``
    (``pam_rules.pl``) errored into ``fail`` inside its own ``catch/3`` and
    ``validate_ownership`` silently never discriminated ownership in
    production (W1-P3 finding). Returns None if construction fails for any
    reason so audit mode can never prevent the dispatcher from starting up.
    """
    try:
        rules_dir = get_prolog_dir()
        bridge = PrologBridge(rules_dir=rules_dir)
        loaded = load_governance_rules(bridge, rules_dir)
        predicate_count = count_governance_predicates(bridge, rules_dir)
        strict = governance_strict()
        # W1-P6 boot health line: a rule base that loaded 0 files (or
        # consulted zero predicates despite claiming files loaded) must be
        # unmistakable in the log, not a silent no-op discovered later by a
        # forensic audit (the exact failure class of the three incidents in
        # reports/proposal-wire-prolog-governance.md).
        log_level = (
            logging.WARNING if loaded == 0 or predicate_count == 0 else logging.INFO
        )
        logger.log(
            log_level,
            "governance_boot surface=dispatcher.PolicyEvaluator "
            "files_loaded=%d/%d predicates=%d strict=%s",
            loaded,
            len(GOVERNANCE_RULE_FILES),
            predicate_count,
            strict,
        )
        return PolicyEvaluator(prolog_bridge=bridge)
    except Exception as e:
        logger.warning(
            "policy_audit_mode enabled but the default PolicyEvaluator could "
            "not be constructed; continuing without audit evaluation: %s",
            e,
        )
        return None


def _format_plan_atom(value: str) -> str:
    """Quote ``value`` as a safe Prolog atom (mirrors
    ``GovernancePrologClient._format_atom``)."""
    sanitized = value.replace("\\", "\\\\").replace("'", "\\'")
    return f"'{sanitized}'"


def _assert_plan_governance_facts(
    bridge: PrologBridge, facts: PlanGovernanceFacts
) -> None:
    """Assert Article XII (plan fidelity, RUL-015) ground facts into
    ``bridge`` from a REAL, parsed execution plan -- GROUND FACTS ONLY, per
    prolog/kg_relevance.pl's doctrine and ADR-0005: Python asserts, rules
    only reason over them. Never a variable-bearing rule.

    No-op (no ``active_plan/1`` fact at all) when ``facts.active`` is
    False -- this is not a defect, it is the correct, honest signal:
    ``plans/execution-plan.md`` (the shipped default) is marked
    ``STATUS: DONE``, so Article XII is vacuously satisfied for every
    action today via ``plan_fidelity(Action, faithful) :- \\+ active_plan(_)``.
    """
    if not facts.active:
        return
    plan_atom = _format_plan_atom(facts.plan_id)
    try:
        bridge.query(f"assertz(constitutional:active_plan({plan_atom}))")
        for step in facts.steps:
            action_atom = _format_plan_atom(f"step_{step.step_num}")
            bridge.query(
                f"assertz(constitutional:plan_step({plan_atom}, {action_atom}))"
            )
    except Exception as e:
        logger.warning(
            "Article XII: failed to assert plan governance facts for %s: %s",
            facts.plan_id,
            e,
        )


@traces(spec="007", story="US-002", reqs=["REQ-007-US-002-001"])
def _build_default_constitutional_enforcer(
    env: Mapping[str, str] | None = None,
) -> ConstitutionalEnforcer | None:
    """Construct a ConstitutionalEnforcer backed by a real PrologBridge, and
    assert Article XII's (plan fidelity, RUL-015) ground facts from the
    real, parsed execution plan (``load_plan_governance_facts``).

    Used by MessageDispatcher when ``policy_audit_mode=True`` and no
    ``constitutional_enforcer`` was injected by the caller. Mirrors
    ``_build_default_policy_evaluator`` above: a SEPARATE ``PrologBridge``
    instance, but the same process-wide ``PrologWorker`` singleton
    underneath (``orchestrator.prolog_worker.get_prolog_worker``), so facts
    asserted here ARE visible to that evaluator's queries and vice versa.
    Returns None if construction fails for any reason so audit mode can
    never prevent the dispatcher from starting up.
    """
    try:
        rules_dir = get_prolog_dir()
        bridge = PrologBridge(rules_dir=rules_dir)
        loaded = load_governance_rules(bridge, rules_dir)
        predicate_count = count_governance_predicates(bridge, rules_dir)
        strict = governance_strict()
        log_level = (
            logging.WARNING if loaded == 0 or predicate_count == 0 else logging.INFO
        )
        logger.log(
            log_level,
            "governance_boot surface=dispatcher.ConstitutionalEnforcer "
            "files_loaded=%d/%d predicates=%d strict=%s",
            loaded,
            len(GOVERNANCE_RULE_FILES),
            predicate_count,
            strict,
        )
        facts = load_plan_governance_facts(env=env)
        _assert_plan_governance_facts(bridge, facts)
        logger.info(
            "governance_boot surface=dispatcher.ConstitutionalEnforcer "
            "article=12 plan_id=%s active=%s status=%s steps=%d",
            facts.plan_id,
            facts.active,
            facts.status,
            len(facts.steps),
        )
        return ConstitutionalEnforcer(GovernancePrologClient(bridge))
    except Exception as e:
        logger.warning(
            "policy_audit_mode enabled but the default ConstitutionalEnforcer "
            "could not be constructed; continuing without Article XII "
            "evaluation: %s",
            e,
        )
        return None


@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-016"])
class DispatchStatus(Enum):
    """Status of dispatch operation."""

    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"
    RETRY = "retry"
    QUEUED = "queued"


@traces(spec="001", story="US-002", reqs=["REQ-001-US-002-015", "REQ-001-US-002-016"])
@dataclass
class DispatchAttempt:
    """Record of a dispatch attempt."""

    attempt_number: int
    timestamp: datetime
    protocol: str
    endpoint: str | None
    status: DispatchStatus
    error_message: str | None = None
    latency_ms: float | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "attempt_number": self.attempt_number,
            "timestamp": self.timestamp.isoformat(),
            "protocol": self.protocol,
            "endpoint": self.endpoint,
            "status": self.status.value,
            "error_message": self.error_message,
            "latency_ms": self.latency_ms,
        }


@traces(
    spec="001",
    story="US-002",
    reqs=["REQ-001-US-002-016", "REQ-001-US-002-019"],
)
@dataclass
class DispatchResult:
    """Result of message dispatch."""

    success: bool
    packet_id: str
    recipient: str
    protocol: str
    endpoint: str | None
    attempts: list[DispatchAttempt] = field(default_factory=list)
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    error_message: str | None = None
    retry_after_seconds: int | None = None

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-016"])
    def add_attempt(self, attempt: DispatchAttempt) -> None:
        """Add dispatch attempt record."""
        self.attempts.append(attempt)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary."""
        return {
            "success": self.success,
            "packet_id": self.packet_id,
            "recipient": self.recipient,
            "protocol": self.protocol,
            "endpoint": self.endpoint,
            "attempts": [a.to_dict() for a in self.attempts],
            "timestamp": self.timestamp.isoformat(),
            "error_message": self.error_message,
            "retry_after_seconds": self.retry_after_seconds,
        }


@traces(
    spec="001",
    story="US-002",
    reqs=["REQ-001-US-002-011", "REQ-001-US-003-001", "REQ-001-US-010-015"],
)
class MessageDispatcher:
    """Dispatches CXP messages via various protocols.

    Handles protocol-specific delivery with retry logic and tracking.

    Staged governance wiring (2026-07-06, decision D3/W4.6): when
    ``policy_audit_mode=True``, every ``dispatch()`` call also evaluates the
    routing decision through kernel.policy_evaluator.PolicyEvaluator and logs
    the verdict at INFO (REQ-001-US-003-001) -- AUDIT-ONLY, it never blocks or
    alters the dispatch outcome. Any evaluator error is caught and logged at
    WARNING; dispatch always proceeds as if audit mode were off. Default is
    False (zero behavior change). Promotion to an enforcing mode is a later,
    explicit flip -- not implemented here.
    """

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    def __init__(
        self,
        max_retries: int = 3,
        base_retry_delay: int = 1,
        timeout_seconds: int = 30,
        policy_audit_mode: bool = False,
        policy_evaluator: PolicyEvaluator | None = None,
        constitutional_enforcer: ConstitutionalEnforcer | None = None,
    ) -> None:
        """Initialize dispatcher.

        Args:
            max_retries: Maximum retry attempts (default 3)
            base_retry_delay: Base delay for exponential backoff (default 1 second)
            timeout_seconds: Dispatch timeout (default 30 seconds)
            policy_audit_mode: When True, construct (or use the injected)
                PolicyEvaluator/ConstitutionalEnforcer and evaluate+log the
                routing policy verdict AND the constitutional Article XII
                (plan fidelity, RUL-015) verdict on every dispatch(),
                without ever influencing the dispatch outcome
                (REQ-001-US-003-001). Default False: zero behavior change, and
                neither is constructed at all.
            policy_evaluator: Optional pre-built PolicyEvaluator to use when
                policy_audit_mode is True (e.g. tests injecting a real,
                test-scoped PrologBridge). If None and policy_audit_mode is
                True, a default evaluator is built once here from the
                process prolog/ rules directory.
            constitutional_enforcer: Optional pre-built ConstitutionalEnforcer
                to use when policy_audit_mode is True (e.g. tests injecting a
                real, test-scoped PrologBridge). If None and
                policy_audit_mode is True, a default enforcer is built once
                here, asserting Article XII's ground facts from the real,
                parsed execution plan.
        """
        self.max_retries = max_retries
        self.base_retry_delay = base_retry_delay
        self.timeout_seconds = timeout_seconds
        self.policy_audit_mode = policy_audit_mode
        self._dispatch_history: list[DispatchResult] = (
            []
        )  # Initialize empty list per instance

        self._policy_evaluator: PolicyEvaluator | None = None
        self._constitutional_enforcer: ConstitutionalEnforcer | None = None
        if self.policy_audit_mode:
            self._policy_evaluator = (
                policy_evaluator
                if policy_evaluator is not None
                else _build_default_policy_evaluator()
            )
            self._constitutional_enforcer = (
                constitutional_enforcer
                if constitutional_enforcer is not None
                else _build_default_constitutional_enforcer()
            )

    @traces(
        spec="001",
        story="US-002",
        reqs=[
            "REQ-001-US-002-011",
            "REQ-001-US-002-015",
            "REQ-001-US-002-017",
            "REQ-001-US-003-001",
            "REQ-001-US-010-015",
            "REQ-001-US-002-016",
            "REQ-001-US-003-009",
        ],
    )
    async def dispatch(
        self, packet: CXPPacket, recipient: VolkDefinition
    ) -> DispatchResult:
        """Dispatch message to recipient via appropriate protocol.

        REQ-001-US-002-011: Dispatch via Volk-specified protocol
        REQ-001-US-002-015: Retry with exponential backoff
        REQ-001-US-002-016: Track attempts and failures
        REQ-001-US-002-017: Timeout stale attempts
        REQ-001-US-003-001: (audit-only, when policy_audit_mode is on) evaluate and
            log the routing policy verdict; never blocks.

        Args:
            packet: CXP packet to dispatch
            recipient: Recipient Volk definition

        Returns:
            DispatchResult with delivery outcome
        """
        await self._audit_evaluate_routing(packet, recipient)
        await self._audit_evaluate_plan_fidelity(packet, recipient)

        result = DispatchResult(
            success=False,
            packet_id=packet.header.packet_id,
            recipient=recipient.alias,
            protocol="",
            endpoint=recipient.endpoint,
        )

        protocol = self._select_protocol(recipient)
        result.protocol = protocol.value

        for attempt_num in range(self.max_retries + 1):
            succeeded = await self._attempt_dispatch(
                packet, recipient, protocol, result, attempt_num
            )
            if succeeded:
                self._dispatch_history.append(result)
                return result

            if attempt_num < self.max_retries:
                delay = self.base_retry_delay * (2**attempt_num)
                result.retry_after_seconds = delay
                await asyncio.sleep(delay)

        result.success = False
        result.error_message = f"Failed after {self.max_retries + 1} attempts"
        self._dispatch_history.append(result)
        return result

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    async def _audit_evaluate_routing(
        self, packet: CXPPacket, recipient: VolkDefinition
    ) -> None:
        """Evaluate the routing policy in AUDIT-ONLY mode and log it (REQ-001-US-003-001).

        No-op unless ``policy_audit_mode`` is on and a PolicyEvaluator was
        successfully constructed. Never raises and never influences the
        dispatch outcome -- this is pure observation while the governance
        layer is staged for real traffic (2026-07-06). Any evaluator failure
        is caught and logged at WARNING so audit mode can never break
        dispatch.

        Args:
            packet: CXP packet about to be dispatched.
            recipient: Recipient Volk definition.
        """
        if self._policy_evaluator is None:
            return

        from_volk = packet.header.sender
        to_volk = recipient.alias
        message_type = packet.header.type.value

        try:
            verdict = await self._policy_evaluator.evaluate_routing(
                from_volk=from_volk,
                to_volk=to_volk,
                message_type=message_type,
            )
            logger.info(
                "policy_audit_mode verdict: %s -> %s (%s) decision=%s reason=%s "
                "provenance=%s (audit-only, dispatch unaffected)",
                from_volk,
                to_volk,
                message_type,
                verdict.decision.value,
                verdict.reason,
                verdict.provenance,
            )
        except Exception as e:
            logger.warning(
                "policy_audit_mode: PolicyEvaluator raised evaluating "
                "%s -> %s (%s); dispatch continues unaffected: %s",
                from_volk,
                to_volk,
                message_type,
                e,
            )

    @traces(spec="007", story="US-002", reqs=["REQ-007-US-002-003"])
    async def _audit_evaluate_plan_fidelity(
        self, packet: CXPPacket, recipient: VolkDefinition
    ) -> None:
        """Evaluate constitutional Article XII (plan fidelity, RUL-015) in
        AUDIT-ONLY mode and log it. No-op unless ``policy_audit_mode`` is on
        and a ConstitutionalEnforcer was successfully constructed. Never
        raises and never influences the dispatch outcome -- pure
        observation while Article XII is staged (2026-08-19), mirroring
        ``_audit_evaluate_routing`` above exactly.

        The Action checked is the CXP ``message_type`` -- the SAME
        vocabulary ``_audit_evaluate_routing`` already audits.
        ``plan_step/2`` facts are keyed by the parsed plan's own step ids
        (``step_<N>``), a DIFFERENT vocabulary: a real ``message_type``
        will therefore never equal a plan-step atom, so whenever a plan IS
        active this always observes "divergent" for ordinary dispatch
        traffic. That mismatch is expected and is itself part of what this
        audit measures -- see the builder's report for the observed
        numbers, not a bug in this wiring.

        Args:
            packet: CXP packet about to be dispatched.
            recipient: Recipient Volk definition.
        """
        if self._constitutional_enforcer is None:
            return

        action = packet.header.type.value

        try:
            result = await self._constitutional_enforcer.check_article(
                12, action, {"session_id": packet.header.packet_id}
            )
            logger.info(
                "policy_audit_mode article12 verdict: action=%s decision=%s "
                "articles_passed=%s articles_failed=%s "
                "(audit-only, dispatch unaffected)",
                action,
                "faithful" if result.compliant else "divergent",
                result.articles_passed,
                result.articles_failed,
            )
        except Exception as e:
            logger.warning(
                "policy_audit_mode: ConstitutionalEnforcer raised evaluating "
                "Article XII for action %s; dispatch continues unaffected: %s",
                action,
                e,
            )

    @traces(
        spec="001", story="US-002", reqs=["REQ-001-US-002-016", "REQ-001-US-002-017"]
    )
    async def _attempt_dispatch(
        self,
        packet: CXPPacket,
        recipient: VolkDefinition,
        protocol: VolkProtocol,
        result: DispatchResult,
        attempt_num: int,
    ) -> bool:
        """Execute a single dispatch attempt, recording the outcome on result.

        Args:
            packet: CXP packet to dispatch
            recipient: Recipient Volk definition
            protocol: Selected protocol
            result: DispatchResult to record the attempt on (mutated)
            attempt_num: Zero-based attempt index

        Returns:
            True if dispatch succeeded
        """
        start_time = datetime.now(UTC)
        try:
            success = await asyncio.wait_for(
                self._dispatch_by_protocol(packet, recipient, protocol),
                timeout=self.timeout_seconds,
            )
            latency_ms = (datetime.now(UTC) - start_time).total_seconds() * 1000
            attempt = DispatchAttempt(
                attempt_number=attempt_num + 1,
                timestamp=start_time,
                protocol=protocol.value,
                endpoint=recipient.endpoint,
                status=DispatchStatus.SUCCESS if success else DispatchStatus.FAILED,
                latency_ms=latency_ms,
            )
            result.add_attempt(attempt)
            if success:
                result.success = True
            return success

        except TimeoutError:
            result.add_attempt(
                DispatchAttempt(
                    attempt_number=attempt_num + 1,
                    timestamp=start_time,
                    protocol=protocol.value,
                    endpoint=recipient.endpoint,
                    status=DispatchStatus.TIMEOUT,
                    error_message=f"Dispatch timeout after {self.timeout_seconds}s",
                )
            )
            logger.warning(
                f"Dispatch timeout for {packet.header.packet_id} to {recipient.alias}"
            )
            return False

        except Exception as e:
            result.add_attempt(
                DispatchAttempt(
                    attempt_number=attempt_num + 1,
                    timestamp=start_time,
                    protocol=protocol.value,
                    endpoint=recipient.endpoint,
                    status=DispatchStatus.FAILED,
                    error_message=str(e),
                )
            )
            logger.error(
                f"Dispatch failed for {packet.header.packet_id} to {recipient.alias} "
                f"(attempt {attempt_num + 1}/{self.max_retries + 1}): "
                f"{type(e).__name__}: {e}",
                exc_info=True,
            )
            return False

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-011"])
    def _select_protocol(self, recipient: VolkDefinition) -> VolkProtocol:
        """Select best protocol for recipient.

        Preference order: CXP > gRPC > WebSocket > HTTP
        """
        if VolkProtocol.CXP in recipient.protocols:
            return VolkProtocol.CXP
        elif VolkProtocol.GRPC in recipient.protocols:
            return VolkProtocol.GRPC
        elif VolkProtocol.WS in recipient.protocols:
            return VolkProtocol.WS
        else:
            return VolkProtocol.HTTP

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-011"])
    async def _dispatch_by_protocol(
        self, packet: CXPPacket, recipient: VolkDefinition, protocol: VolkProtocol
    ) -> bool:
        """Dispatch using specific protocol.

        Args:
            packet: CXP packet to send
            recipient: Recipient Volk
            protocol: Protocol to use

        Returns:
            True if dispatch succeeded

        Raises:
            ValueError: If protocol is not supported, or the recipient has no
                endpoint configured (a Volk's ``endpoint`` is optional --
                e.g. in-process/CXP-only Volks -- but every protocol below
                needs a real address to connect to).
        """
        endpoint = recipient.endpoint
        if endpoint is None:
            error_msg = (
                f"Volk {recipient.alias} has no endpoint configured; cannot "
                f"dispatch packet {packet.header.packet_id} via {protocol.value}"
            )
            logger.error(error_msg)
            raise ValueError(error_msg)

        if protocol == VolkProtocol.GRPC:
            return await self.dispatch_grpc(packet, endpoint)
        elif protocol == VolkProtocol.WS:
            return await self.dispatch_websocket(packet, endpoint)
        elif protocol == VolkProtocol.HTTP:
            return await self.dispatch_http(packet, endpoint)
        elif protocol == VolkProtocol.CXP:
            # CXP is abstraction layer, use gRPC as transport
            return await self.dispatch_grpc(packet, endpoint)
        else:
            # Explicit error for unknown protocols
            error_msg = (
                f"Unsupported protocol: {protocol} for packet {packet.header.packet_id}"
            )
            logger.error(error_msg)
            raise ValueError(error_msg)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-012"])
    async def dispatch_grpc(self, packet: CXPPacket, endpoint: str) -> bool:
        """Dispatch message via gRPC.

        REQ-001-US-002-012: Support CXP over gRPC

        Args:
            packet: CXP packet to send
            endpoint: gRPC endpoint (e.g., grpc://localhost:50051)

        Returns:
            True if dispatch succeeded
        """
        host, port = self._parse_grpc_endpoint(endpoint)
        config = CXPClientConfig(
            host=host,
            port=port,
            timeout_seconds=self.timeout_seconds,
            max_retries=1,
        )
        client = create_cxp_client(config=config)
        try:
            return await self._send_grpc_packet(client, packet)
        finally:
            await client.disconnect()

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-012"])
    def _parse_grpc_endpoint(self, endpoint: str) -> tuple[str, int]:
        """Parse gRPC endpoint into host and port.

        Strips the ``grpc://`` / ``cxp://`` scheme. When no ``:port`` is present
        the port falls back to ``CXP_GRPC_PORT`` from the environment (.env) —
        there is no hardcoded default; a missing value raises.

        Args:
            endpoint: Endpoint string (e.g., grpc://host:port or host:port)

        Returns:
            Tuple of (host, port)

        Raises:
            ValueError: If *endpoint* is empty/None, the host is empty, the port
                is non-numeric, or the port falls outside the 1..65535 range.
        """
        if not endpoint:
            raise ValueError("gRPC endpoint must be a non-empty string")

        cleaned = endpoint.replace("grpc://", "").replace("cxp://", "")
        if ":" in cleaned:
            host, port_str = cleaned.rsplit(":", 1)
            try:
                port = int(port_str)
            except ValueError as exc:
                raise ValueError(
                    f"Invalid gRPC endpoint '{endpoint}': port '{port_str}' "
                    "is not an integer"
                ) from exc
        else:
            host, port = cleaned, cxp_grpc_port()

        if not host:
            raise ValueError(
                f"Invalid gRPC endpoint '{endpoint}': host must not be empty"
            )
        if not (1 <= port <= 65535):
            raise ValueError(
                f"Invalid gRPC endpoint '{endpoint}': port {port} is outside "
                "the valid range 1..65535"
            )

        return host, port

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-012"])
    async def _send_grpc_packet(self, client: CXPClientLike, packet: CXPPacket) -> bool:
        """Connect and send a packet via gRPC client.

        Args:
            client: Configured CXP client instance
            packet: CXP packet to send

        Returns:
            True if the server acknowledged success
        """
        await client.connect()
        result = await client.send_packet(packet)
        success = result.get("success", False)
        if success:
            logger.info(
                "gRPC dispatch succeeded for packet %s",
                packet.header.packet_id,
            )
        else:
            logger.error(
                "gRPC dispatch failed for packet %s: %s",
                packet.header.packet_id,
                result.get("error", "unknown"),
            )
        return success

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-013"])
    async def dispatch_websocket(self, packet: CXPPacket, endpoint: str) -> bool:
        """Dispatch message via WebSocket.

        REQ-001-US-002-013: Support CXP over WebSocket

        Args:
            packet: CXP packet to send
            endpoint: WebSocket endpoint (e.g., ws://localhost:8080)

        Returns:
            True if dispatch succeeded

        Raises:
            RuntimeError: If websockets library is not installed
        """
        if websockets is None:
            error_msg = (
                f"websockets not installed. Cannot dispatch to {endpoint} "
                f"for packet {packet.header.packet_id}. "
                "Install with: uv add websockets"
            )
            logger.error(error_msg)
            raise RuntimeError(error_msg)

        payload = json.dumps(packet.to_dict(), default=str)
        return await self._send_websocket(packet.header.packet_id, endpoint, payload)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-013"])
    async def _send_websocket(
        self, packet_id: str, endpoint: str, payload: str
    ) -> bool:
        """Connect, send payload, and await ack over WebSocket.

        Args:
            packet_id: Packet identifier for logging
            endpoint: WebSocket URI
            payload: JSON-serialized packet data

        Returns:
            True if server acknowledged successfully
        """
        try:
            async with websockets.connect(
                endpoint,
                close_timeout=self.timeout_seconds,
            ) as ws:
                return await self._exchange_websocket(ws, packet_id, endpoint, payload)
        except websockets.exceptions.InvalidHandshake as exc:
            logger.error(
                "WebSocket handshake failed for %s to %s: %s",
                packet_id,
                endpoint,
                exc,
            )
            return False
        except (OSError, ConnectionRefusedError) as exc:
            logger.error(
                "WebSocket connection refused for %s to %s: %s",
                packet_id,
                endpoint,
                exc,
            )
            return False

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-013"])
    async def _exchange_websocket(
        self,
        ws: Any,
        packet_id: str,
        endpoint: str,
        payload: str,
    ) -> bool:
        """Send payload and parse ack on an open WebSocket.

        Args:
            ws: Open WebSocket connection
            packet_id: Packet identifier for logging
            endpoint: Endpoint URI for logging
            payload: Serialized packet JSON

        Returns:
            True if ack indicates success
        """
        try:
            await ws.send(payload)
            raw_ack = await asyncio.wait_for(ws.recv(), timeout=self.timeout_seconds)
            return self._parse_ws_ack(raw_ack, packet_id, endpoint)
        except TimeoutError:
            logger.error(
                "WebSocket ack timeout for %s to %s after %ss",
                packet_id,
                endpoint,
                self.timeout_seconds,
            )
            return False
        except websockets.exceptions.ConnectionClosed as exc:
            logger.error(
                "WebSocket closed during exchange for %s to %s: %s",
                packet_id,
                endpoint,
                exc,
            )
            return False

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-019"])
    def _parse_ws_ack(self, raw_ack: str, packet_id: str, endpoint: str) -> bool:
        """Parse WebSocket acknowledgment JSON.

        Args:
            raw_ack: Raw ack string from server
            packet_id: Packet identifier for logging
            endpoint: Endpoint URI for logging

        Returns:
            True if ack status indicates success
        """
        try:
            ack = json.loads(raw_ack)
        except (json.JSONDecodeError, TypeError) as exc:
            logger.error(
                "Invalid ack JSON for %s from %s: %s",
                packet_id,
                endpoint,
                exc,
            )
            return False

        success = ack.get("status") == "ok"
        if success:
            logger.info(
                "WebSocket dispatch succeeded for %s to %s",
                packet_id,
                endpoint,
            )
        else:
            logger.error(
                "WebSocket NACK for %s from %s: %s",
                packet_id,
                endpoint,
                ack,
            )
        return success

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-014"])
    async def dispatch_http(self, packet: CXPPacket, endpoint: str) -> bool:
        """Dispatch message via HTTP POST.

        REQ-001-US-002-014: Support CXP over HTTP

        Args:
            packet: CXP packet to send
            endpoint: HTTP endpoint (e.g., http://localhost:8080/cxp)

        Returns:
            True if dispatch succeeded

        Raises:
            RuntimeError: If aiohttp is not installed
            aiohttp.ClientError: On HTTP client errors
        """
        if aiohttp is None:
            error_msg = (
                f"aiohttp not installed. Cannot dispatch to {endpoint} for packet {packet.header.packet_id}. "
                "Install with: uv add aiohttp"
            )
            logger.error(error_msg)
            raise RuntimeError(error_msg)

        payload = self._serialize_packet(packet)
        return await self._post_http(packet, endpoint, payload)

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-014"])
    def _serialize_packet(self, packet: CXPPacket) -> dict[str, Any]:
        """Serialize CXPPacket to a JSON-safe dict.

        Args:
            packet: Packet to serialize

        Returns:
            Dictionary representation

        Raises:
            ValueError: If serialization fails
        """
        try:
            return packet.to_dict()
        except Exception as e:
            error_msg = f"Failed to serialize packet {packet.header.packet_id}: {e}"
            logger.error(error_msg)
            raise ValueError(error_msg) from e

    @traces(spec="001", story="US-002", reqs=["REQ-001-US-002-014"])
    async def _post_http(
        self, packet: CXPPacket, endpoint: str, payload: dict[str, Any]
    ) -> bool:
        """POST serialized payload to HTTP endpoint.

        Args:
            packet: Original packet (used for logging)
            endpoint: HTTP endpoint URL
            payload: JSON-serializable payload

        Returns:
            True if HTTP 2xx response received

        Raises:
            aiohttp.ClientError: On transport errors
            asyncio.TimeoutError: On timeout
        """
        timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
        try:
            async with aiohttp.ClientSession(timeout=timeout) as session:
                async with session.post(
                    endpoint, json=payload, headers={"Content-Type": "application/json"}
                ) as response:
                    if 200 <= response.status < 300:
                        logger.info(
                            f"HTTP dispatch succeeded to {endpoint} for packet {packet.header.packet_id}, "
                            f"status {response.status}"
                        )
                        return True
                    error_msg = (
                        f"HTTP dispatch failed to {endpoint} for packet {packet.header.packet_id}, "
                        f"status {response.status}: {await response.text()}"
                    )
                    logger.error(error_msg)
                    return False
        except aiohttp.ClientConnectionError as e:
            logger.error(f"Connection error during HTTP dispatch to {endpoint}: {e}")
            raise
        except aiohttp.ClientResponseError as e:
            logger.error(f"Response error during HTTP dispatch to {endpoint}: {e}")
            raise
        except TimeoutError:
            logger.error(
                f"Timeout during HTTP dispatch to {endpoint} after {self.timeout_seconds}s"
            )
            raise
        except Exception as e:
            logger.error(
                f"Unexpected error during HTTP dispatch to {endpoint}: {type(e).__name__}: {e}"
            )
            raise

    @traces(
        spec="001", story="US-002", reqs=["REQ-001-US-002-016", "REQ-001-US-002-015"]
    )
    def get_dispatch_history(
        self, packet_id: str | None = None
    ) -> list[DispatchResult]:
        """Get dispatch history.

        Args:
            packet_id: Optional packet ID to filter by

        Returns:
            Copy of dispatch results list (prevents external mutation)
        """
        if packet_id:
            return [r for r in self._dispatch_history if r.packet_id == packet_id]
        return list(self._dispatch_history)