orchestrata_ultimate.harness.scenario

Reference › orchestrata_ultimate › harness › scenario

Loading and validating scenarios, and running their setup hooks (SPEC-102 US-001). load_scenario(path) reads a YAML file (_read_scenario_mapping requires a non-empty mapping), checks it with _validate_scenario_fields and builds a Scenario. Validation requires the fields id, volk, mode and evaluation, a mode of cxp_inject or direct_call, and a non-empty evaluation list whose items each have a name; problems raise ValueError naming the scenario. load_scenarios_for_volk and list_all_scenarios discover scenario files under harness/scenarios by default. run_setup_hook(hook_path, context) imports a hook given as module.path:function_name, and requires an async function that accepts the HarnessContext and returns None; every failure (import, missing name, not async, no parameter, an exception inside it, a non-None return) is reported as a ValueError. Used by the runner and the cli; the models are in models.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Scenario loading, validation, and setup hook execution."""

from __future__ import annotations

import importlib
import inspect
import logging
from pathlib import Path
from typing import Any

import yaml
from pydantic import ValidationError

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.harness.models import HarnessContext, Scenario

_logger = logging.getLogger(__name__)


@traces(spec="102", story="US-001", reqs=["REQ-102-004"])
def _read_scenario_mapping(path: Path) -> dict:
    """Load *path* as YAML and ensure it is a non-empty mapping.

    Raises:
        ValueError: If the file is missing, invalid YAML, empty, or not a mapping.
    """
    if not path.exists():
        raise ValueError(f"Scenario file not found: {path}")

    try:
        raw: Any = yaml.safe_load(path.read_text(encoding="utf-8"))
    except Exception as exc:
        raise ValueError(
            f"Scenario file contains invalid YAML: {path} ({exc})"
        ) from exc
    if raw is None:
        raise ValueError(f"Scenario file is empty: {path}")
    if not isinstance(raw, dict):
        raise ValueError(f"Scenario file must contain a YAML mapping: {path}")
    return raw


@traces(spec="102", story="US-001", reqs=["REQ-102-002"])
def _validate_scenario_fields(raw: dict, name: str) -> None:
    """Validate required fields, mode, and the evaluation list of a scenario dict.

    Raises:
        ValueError: If a required field is missing, the mode is invalid, or the
            evaluation list is empty/malformed.
    """
    required = {"id", "volk", "mode", "evaluation"}
    missing = required - raw.keys()
    if missing:
        raise ValueError(f"Scenario {name} missing required fields: {sorted(missing)}")

    mode = raw.get("mode")
    if mode not in {"cxp_inject", "direct_call"}:
        raise ValueError(
            f"Scenario {name} has invalid mode '{mode}'; "
            "must be 'cxp_inject' or 'direct_call'"
        )

    evaluation = raw.get("evaluation")
    if not isinstance(evaluation, list) or len(evaluation) == 0:
        raise ValueError(f"Scenario {name} 'evaluation' must be a non-empty list")

    for idx, ev in enumerate(evaluation):
        if not isinstance(ev, dict) or "name" not in ev:
            raise ValueError(
                f"Scenario {name} evaluator at index {idx} "
                "is missing required field 'name'"
            )


@traces(spec="102", story="US-001", reqs=["REQ-102-004"])
def load_scenario(path: Path) -> Scenario:
    """Parse a scenario YAML file and validate it with the Pydantic model.

    Args:
        path: Path to the YAML scenario file.

    Returns:
        A validated :class:`Scenario` instance.

    Raises:
        ValueError: If the YAML is invalid or fails validation rules.
    """
    raw = _read_scenario_mapping(path)
    _validate_scenario_fields(raw, path.name)

    try:
        return Scenario.model_validate(raw)
    except ValidationError as exc:
        raise ValueError(f"Scenario {path.name} validation failed: {exc}") from exc


@traces(spec="102", story="US-001", reqs=["REQ-102-001"])
def load_scenarios_for_volk(
    volk_name: str,
    base_dir: Path = Path("harness/scenarios"),
) -> list[Scenario]:
    """Discover and load all scenario YAML files for a given volk.

    Args:
        volk_name: Name of the volk directory to scan.
        base_dir: Root directory containing volk scenario folders.

    Returns:
        List of validated :class:`Scenario` instances.

    Raises:
        ValueError: If duplicate scenario ids are found within the volk.
    """
    volk_dir = base_dir / volk_name
    if not volk_dir.exists():
        _logger.warning(
            "Scenario directory not found for volk '%s': %s",
            volk_name,
            volk_dir,
        )
        return []

    scenarios: list[Scenario] = []
    seen_ids: set[str] = set()

    for yaml_path in sorted(volk_dir.glob("*.yaml")):
        scenario = load_scenario(yaml_path)
        if scenario.id in seen_ids:
            raise ValueError(
                f"Duplicate scenario id '{scenario.id}' in volk '{volk_name}'"
            )
        seen_ids.add(scenario.id)
        scenarios.append(scenario)

    return scenarios


@traces(spec="102", story="US-001", reqs=["REQ-102-006"])
def list_all_scenarios(
    base_dir: Path = Path("harness/scenarios"),
) -> list[Scenario]:
    """Discover and load all scenarios across all volks.

    Args:
        base_dir: Root directory containing volk scenario folders.

    Returns:
        List of all validated :class:`Scenario` instances.

    Raises:
        ValueError: If duplicate scenario ids are found across volks.
    """
    if not base_dir.exists():
        _logger.warning(
            "Scenario base directory not found: %s",
            base_dir,
        )
        return []

    all_scenarios: list[Scenario] = []
    global_ids: set[str] = set()

    for volk_dir in sorted(base_dir.iterdir()):
        if not volk_dir.is_dir():
            continue

        for yaml_path in sorted(volk_dir.glob("*.yaml")):
            scenario = load_scenario(yaml_path)
            if scenario.id in global_ids:
                raise ValueError(
                    f"Duplicate scenario id '{scenario.id}' across volks "
                    f"(file: {yaml_path})"
                )
            global_ids.add(scenario.id)
            all_scenarios.append(scenario)

    return all_scenarios


@traces(spec="102", story="US-006", reqs=["REQ-102-087", "REQ-102-088"])
async def run_setup_hook(hook_path: str, context: HarnessContext) -> None:
    """Dynamically import and execute an async setup hook function.

    Args:
        hook_path: Module path and function name in the format
            ``"module.path:function_name"``.
        context: Runtime harness context to pass to the hook.

    Raises:
        ValueError: If the module cannot be imported, the function does not
            exist, is not an async function, does not accept
            :class:`HarnessContext`, or returns a non-None value.
    """
    if ":" not in hook_path:
        raise ValueError(
            f"Invalid hook path '{hook_path}'; "
            "expected format 'module.path:function_name'"
        )

    module_path, func_name = hook_path.split(":", 1)

    try:
        module = importlib.import_module(module_path)
    except Exception as exc:
        raise ValueError(
            f"Failed to import setup hook module '{module_path}': {exc}"
        ) from exc

    func = getattr(module, func_name, None)
    if func is None:
        raise ValueError(
            f"Setup hook function '{func_name}' not found in module '{module_path}'"
        )

    if not callable(func):
        raise ValueError(f"Setup hook '{hook_path}' is not callable")

    if not inspect.iscoroutinefunction(func):
        raise ValueError(f"Setup hook '{hook_path}' must be an async function")

    # Verify the function accepts at least one positional argument
    sig = inspect.signature(func)
    params = list(sig.parameters.values())
    if len(params) == 0:
        raise ValueError(
            f"Setup hook '{hook_path}' must accept a HarnessContext argument"
        )

    try:
        result = await func(context)
    except Exception as exc:
        raise ValueError(
            f"Setup hook '{hook_path}' raised an exception: {exc}"
        ) from exc

    if result is not None:
        raise ValueError(
            f"Setup hook '{hook_path}' must return None, "
            f"got {type(result).__name__}"
        )