orchestrata_ultimate.harness.executor

Reference › orchestrata_ultimate › harness › executor

The execution engines of the harness (SPEC-102 US-006) and its command whitelist. Executor is the abstract base with one async method, execute(scenario, context), returning the raw output that the evaluators then judge. It is used by runner and the cli.

CXPacketExecutor builds a CXPPacket from the scenario: the subject is the input's intent (or the scenario id), the payload is the input plus the trial's test_id, the sender is harness and the recipient is the scenario's volk. It wraps the volk in a minimal VolkDefinition addressed to the CXP gRPC endpoint from cxp.endpoint_config, dispatches through the kernel's MessageDispatcher under the scenario timeout and returns the dispatch result as a dictionary. DirectCallExecutor imports the volk class, instantiates it and calls the method named by input["method"] with input["args"]; coroutine methods are awaited and synchronous ones run in a thread, both under the scenario timeout. A missing method raises ValueError.

enforce_whitelist(command, whitelist) takes the base name of the first command element and raises SecurityError (a ValueError) if it is not in the whitelist or the command is empty. The default whitelist is in HarnessConfig.

At a glance

Class diagram

Dependency diagram

API

Classes

SecurityError

Bases: ValueError · Decorators: traces(spec='102', story='US-011', reqs=['REQ-102-120', 'REQ-102-121']) · Line 25

Raised when a subprocess command violates the harness whitelist.

Executor (abstract)

Bases: ABC · Line 29

Abstract base for scenario executors.

CXPacketExecutor

Bases: Executor · Decorators: traces(spec='102', story='US-006', reqs=['REQ-102-095', 'REQ-102-096', 'REQ-102-097']) · Line 39

Execute a scenario by injecting a CXP packet into the kernel dispatcher.

DirectCallExecutor

Bases: Executor · Decorators: traces(spec='102', story='US-006', reqs=['REQ-102-099', 'REQ-102-100', 'REQ-102-102']) · Line 88

Execute a scenario by instantiating the target volk and calling a method.

Functions

Source

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

"""Scenario execution engines — CXP packet injection and direct method calls."""

from __future__ import annotations

import asyncio
import importlib
import inspect
import logging
from abc import ABC, abstractmethod
from collections.abc import Mapping
from pathlib import Path
from typing import Any

from orchestrata_ultimate.agents.volk_registry import VolkDefinition, VolkProtocol
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, PacketType
from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_endpoint
from orchestrata_ultimate.harness.models import HarnessContext, Scenario
from orchestrata_ultimate.kernel.dispatcher import MessageDispatcher

_logger = logging.getLogger(__name__)


@traces(spec="102", story="US-011", reqs=["REQ-102-120", "REQ-102-121"])
class SecurityError(ValueError):
    """Raised when a subprocess command violates the harness whitelist."""


class Executor(ABC):
    """Abstract base for scenario executors."""

    @abstractmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-095", "REQ-102-099"])
    async def execute(self, scenario: Scenario, context: HarnessContext) -> Any:
        """Run *scenario* and return its raw output."""


@traces(spec="102", story="US-006", reqs=["REQ-102-095", "REQ-102-096", "REQ-102-097"])
class CXPacketExecutor(Executor):
    """Execute a scenario by injecting a CXP packet into the kernel dispatcher."""

    def __init__(
        self,
        dispatcher: MessageDispatcher | None = None,
        environ: Mapping[str, str] | None = None,
    ) -> None:
        self.dispatcher = dispatcher or MessageDispatcher()
        self.environ = environ

    @traces(spec="102", story="US-006", reqs=["REQ-102-098"])
    async def execute(self, scenario: Scenario, context: HarnessContext) -> Any:
        """Build a :class:`CXPPacket` from *scenario* and dispatch it.

        The packet subject is the ``intent`` field from ``scenario.input`` (falling
        back to ``scenario.id``). The payload carries the full input dict plus the
        ``test_id`` for resource isolation. The method waits for the dispatcher
        result and returns it as a dictionary.
        """
        intent = scenario.input.get("intent", scenario.id)
        payload: dict[str, Any] = {**scenario.input, "test_id": context.test_id}

        packet = CXPPacket(
            from_agent="harness",
            to_agent=scenario.volk,
            subject=intent,
            payload=payload,
            packet_type=PacketType.COMMAND,
        )

        # Minimal VolkDefinition so the dispatcher can route the packet.
        recipient = VolkDefinition(
            alias=scenario.volk,
            name=scenario.volk,
            role=scenario.volk,
            capabilities=[],
            protocols=[VolkProtocol.CXP],
            endpoint=cxp_grpc_endpoint(self.environ),
        )

        dispatch_result = await asyncio.wait_for(
            self.dispatcher.dispatch(packet, recipient),
            timeout=scenario.timeout_seconds,
        )
        return dispatch_result.to_dict()


@traces(spec="102", story="US-006", reqs=["REQ-102-099", "REQ-102-100", "REQ-102-102"])
class DirectCallExecutor(Executor):
    """Execute a scenario by instantiating the target volk and calling a method."""

    @traces(spec="102", story="US-006", reqs=["REQ-102-101", "REQ-102-103"])
    async def execute(self, scenario: Scenario, context: HarnessContext) -> Any:
        """Dynamically import the target volk, instantiate it, and invoke the
        requested method.

        Required in ``scenario.input``:
        - ``method``: Name of the volk method to call.

        Optional in ``scenario.input``:
        - ``args``: Dictionary of keyword arguments passed to the method.
        """
        volk_class = self._import_volk_class(scenario.volk)
        instance = volk_class()

        method_name = scenario.input.get("method")
        if not method_name:
            raise ValueError("direct_call mode requires 'method' in scenario.input")

        args = scenario.input.get("args", {})
        method = getattr(instance, method_name)

        if inspect.iscoroutinefunction(method):
            return await asyncio.wait_for(
                method(**args),
                timeout=scenario.timeout_seconds,
            )

        # Sync method — offload to a thread so we don't block the event loop.
        loop = asyncio.get_running_loop()
        return await asyncio.wait_for(
            loop.run_in_executor(None, lambda: method(**args)),
            timeout=scenario.timeout_seconds,
        )

    @staticmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-099", "REQ-102-103"])
    def _import_volk_class(volk_name: str) -> type[Any]:
        """Import and return the primary class for *volk_name*.

        Tries well-known naming patterns (e.g. ``QaVolk``, ``QualityAssuranceAgent``)
        before falling back to the first class in the module whose name ends with
        ``Volk`` or ``Agent``.
        """
        module_path = f"orchestrata_ultimate.agents.volks.{volk_name}"
        try:
            module = importlib.import_module(module_path)
        except ImportError as exc:
            raise ImportError(
                f"Failed to import volk module '{module_path}': {exc}"
            ) from exc

        # Special-case mappings for volks whose class names don't follow a simple
        # title-casing rule.
        special_cases: dict[str, list[str]] = {
            "analyst": ["AnalystAgent"],
            "architect": ["ArchitectVolk"],
            "codedirector": ["CodeDirectorVolk", "CodeDirectorAgent"],
            "compliance": ["ComplianceVolk", "ComplianceAgent"],
            "concierge": ["ConciergeVolk", "ConciergeAgent"],
            "junior": ["JuniorVolk", "JuniorAgent"],
            "pm": ["PMVolk", "ProjectManagerAgent"],
            "qa": ["QaVolk", "QualityAssuranceAgent"],
            "senior_dev": ["SeniorDevAgent"],
            "ux": ["UXVolk", "UserExperienceAgent"],
        }

        guesses = special_cases.get(volk_name, [])
        # Heuristic fallback: title-case the volk name and append Volk/Agent.
        normalized = volk_name.title().replace("_", "")
        guesses.extend([f"{normalized}Volk", f"{normalized}Agent"])

        for class_name in guesses:
            if hasattr(module, class_name):
                cls = getattr(module, class_name)
                if isinstance(cls, type):
                    return cls

        # Last resort: scan the module for any class ending in Volk or Agent.
        for name, obj in inspect.getmembers(module, inspect.isclass):
            if name.endswith(("Volk", "Agent")):
                return obj

        raise ImportError(
            f"No volk class found in module '{module_path}'. " f"Tried: {guesses}"
        )


@traces(spec="102", story="US-011", reqs=["REQ-102-120", "REQ-102-121"])
def enforce_whitelist(command: list[str], whitelist: list[str]) -> None:
    """Validate that the base command in *command* appears in *whitelist*.

    Args:
        command: The command line as a list of strings (e.g. ``["git", "status"]``).
        whitelist: Allowed base command names.

    Raises:
        SecurityError: If the base command is not present in the whitelist.
    """
    if not command:
        raise SecurityError("Empty command is not allowed")

    base = Path(command[0]).name
    if base not in whitelist:
        raise SecurityError(
            f"Command '{base}' is not in the command whitelist. "
            f"Allowed commands: {whitelist}"
        )