orchestrata_ultimate.harness.cli

Reference › orchestrata_ultimate › harness › cli

The command-line interface of the Volk Test Harness, built with argparse and started by main (python -m orchestrata_ultimate.harness). main(argv) loads the dotenv file, sets up logging, and defines six subcommands, each handled by an async function:

• run runs the scenarios selected by a volk name, --scenario (an id or a file path, resolved by _resolve_scenario_arg) or --all, with --trials (default 3), through the HarnessRunner, and prints an ASCII results table.
• list prints the discovered scenarios, optionally filtered by volk.
• snapshot update refreshes stored snapshots.
• report writes the HTML report (default harness_report.html) through reporting.
• trends shows trend data, optionally for one volk.
• config prints the effective configuration from config.

With no subcommand it prints help and exits with the configuration-error code. The command functions return exit codes, which main passes to sys.exit.

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/cli.py

"""CLI interface for the Volk Test Harness — argparse-based, minimal dependencies."""

from __future__ import annotations

import argparse
import logging
import sys
from collections.abc import Sequence
from pathlib import Path

import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.harness.config import load_config
from orchestrata_ultimate.harness.executor import Executor, SecurityError
from orchestrata_ultimate.harness.models import Scenario
from orchestrata_ultimate.harness.reporting import TrendQueries, generate_report
from orchestrata_ultimate.harness.runner import HarnessRunner
from orchestrata_ultimate.harness.scenario import list_all_scenarios, load_scenario

_logger = logging.getLogger(__name__)

EXIT_OK = 0
EXIT_FAILURE = 1
EXIT_CONFIG_ERROR = 2
EXIT_SECURITY_VIOLATION = 3


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase env contract (thin delegation).

    Exactly ONE canonical loader may exist (``scripts/check_env_contract.py``
    enforces it); this name is kept because it is the documented entry hook
    (see ``gateway.cli._load_dotenv``).
    """
    load_env_contract()


def _setup_logging() -> None:
    """Configure basic logging for CLI runs."""
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )


@traces(spec="102", story="US-001", reqs=["REQ-102-018"])
def _print_results_table(results: list) -> None:
    """Print a simple ASCII table of scenario results."""
    if not results:
        print("No results to display.")
        return

    # Determine column widths
    id_width = max(len(r.scenario_id) for r in results)
    volk_width = max(len(r.volk_name) for r in results)
    id_width = max(id_width, len("Scenario"))
    volk_width = max(volk_width, len("Volk"))

    header = (
        f"{'Scenario':<{id_width}}  "
        f"{'Volk':<{volk_width}}  "
        f"{'Trials':>6}  "
        f"{'Passed':>6}  "
        f"{'Score':>8}  "
        f"{'Time(ms)':>10}"
    )
    print(header)
    print("-" * len(header))
    for r in results:
        passed_str = "PASS" if r.passed else "FAIL"
        print(
            f"{r.scenario_id:<{id_width}}  "
            f"{r.volk_name:<{volk_width}}  "
            f"{len(r.trials):>6}  "
            f"{passed_str:>6}  "
            f"{r.aggregate_score:>8.3f}  "
            f"{r.execution_time_ms:>10}"
        )


@traces(spec="102", story="US-001", reqs=["REQ-102-006"])
def _print_scenarios_table(scenarios: list) -> None:
    """Print a formatted table of scenarios."""
    if not scenarios:
        print("No scenarios found.")
        return

    id_width = max(len(s.id) for s in scenarios)
    volk_width = max(len(s.volk) for s in scenarios)
    mode_width = max(len(s.mode) for s in scenarios)
    id_width = max(id_width, len("ID"))
    volk_width = max(volk_width, len("Volk"))
    mode_width = max(mode_width, len("Mode"))

    header = (
        f"{'ID':<{id_width}}  "
        f"{'Volk':<{volk_width}}  "
        f"{'Mode':<{mode_width}}  "
        f"Description"
    )
    print(header)
    print("-" * len(header))
    for s in scenarios:
        desc = s.description
        if len(desc) > 60:
            desc = desc[:57] + "..."
        print(
            f"{s.id:<{id_width}}  {s.volk:<{volk_width}}  {s.mode:<{mode_width}}  {desc}"
        )


@traces(spec="102", story="US-001", reqs=["REQ-102-008"])
def _resolve_scenario_arg(scenario_arg: str) -> Scenario | None:
    """Resolve a ``--scenario`` value (file path or scenario id) to a Scenario.

    Returns the matching :class:`Scenario`, or ``None`` if no scenario matches.
    """
    try:
        return load_scenario(Path(scenario_arg))
    except ValueError:
        # scenario may be an id rather than a path — search across all
        matched = [s for s in list_all_scenarios() if s.id == scenario_arg]
        return matched[0] if matched else None


@traces(spec="102", story="US-001", reqs=["REQ-102-008", "REQ-102-009"])
async def _run_selected_scenarios(
    runner: HarnessRunner, args: argparse.Namespace
) -> list | int:
    """Run the scenarios selected by *args*.

    Returns the list of results, or an exit code (int) on a selection error.
    """
    if args.all:
        scenarios = list_all_scenarios()
        if not scenarios:
            print("No scenarios found.", file=sys.stderr)
            return EXIT_CONFIG_ERROR
        return [
            await runner.run_scenario(scenario, trials=args.trials)
            for scenario in scenarios
        ]
    if args.scenario:
        scenario = _resolve_scenario_arg(args.scenario)
        if scenario is None:
            print(f"Scenario not found: {args.scenario}", file=sys.stderr)
            return EXIT_CONFIG_ERROR
        return [await runner.run_scenario(scenario, trials=args.trials)]
    if args.volk:
        return await runner.run_volk(args.volk, trials=args.trials)
    print(
        "Usage: harness run <volk> | harness run --scenario <id> | harness run --all",
        file=sys.stderr,
    )
    return EXIT_CONFIG_ERROR


@traces(spec="102", story="US-001", reqs=["REQ-102-017"])
async def _cmd_run(args: argparse.Namespace) -> int:
    """Execute the ``run`` subcommand."""
    try:
        config = load_config()
    except Exception as exc:
        print(f"Configuration error: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    runner = HarnessRunner(config)

    try:
        outcome = await _run_selected_scenarios(runner, args)
    except SecurityError as exc:
        print(f"Security violation: {exc}", file=sys.stderr)
        return EXIT_SECURITY_VIOLATION
    except Exception as exc:
        print(f"Run failed: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    if isinstance(outcome, int):
        return outcome

    results = outcome
    _print_results_table(results)
    return EXIT_OK if all(r.passed for r in results) else EXIT_FAILURE


@traces(spec="102", story="US-001", reqs=["REQ-102-006"])
async def _cmd_list(args: argparse.Namespace) -> int:
    """Execute the ``list`` subcommand."""
    try:
        config = load_config()
    except Exception as exc:
        print(f"Configuration error: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    runner = HarnessRunner(config)
    scenarios = runner.list_scenarios(volk_name=args.volk)
    _print_scenarios_table(scenarios)
    return EXIT_OK


@traces(spec="102", story="US-004", reqs=["REQ-102-055"])
async def _cmd_snapshot(args: argparse.Namespace) -> int:
    """Execute the ``snapshot`` subcommand."""
    try:
        config = load_config()
    except Exception as exc:
        print(f"Configuration error: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    runner = HarnessRunner(config)
    scenarios: list = []

    if args.scenario:
        all_scenarios = list_all_scenarios()
        matched = [s for s in all_scenarios if s.id == args.scenario]
        if not matched:
            print(f"Scenario not found: {args.scenario}", file=sys.stderr)
            return EXIT_CONFIG_ERROR
        scenarios = matched
    else:
        scenarios = list_all_scenarios()

    if args.action == "update":
        for scenario in scenarios:
            try:
                result = await runner.run_scenario(scenario, trials=1)
                if result.trials:
                    # Re-execute a single trial to get fresh output for snapshot
                    # We need to run again because TrialResult doesn't store raw output.
                    import uuid

                    from orchestrata_ultimate.harness.executor import (
                        CXPacketExecutor,
                        DirectCallExecutor,
                    )
                    from orchestrata_ultimate.harness.isolation import (
                        WorktreeIsolation,
                        get_graph_prefix,
                        get_redis_db_for_trial,
                    )
                    from orchestrata_ultimate.harness.models import HarnessContext
                    from orchestrata_ultimate.harness.scenario import run_setup_hook

                    test_id = uuid.uuid4().hex
                    worktree_path = await WorktreeIsolation().create(test_id)
                    try:
                        ctx = HarnessContext(
                            scenario_id=scenario.id,
                            volk_name=scenario.volk,
                            trial_number=1,
                            test_id=test_id,
                            worktree_path=worktree_path,
                            redis_db=get_redis_db_for_trial(0),
                            graph_prefix=get_graph_prefix(test_id),
                            config=config,
                            logger=_logger,
                        )
                        if scenario.setup_hook:
                            await run_setup_hook(scenario.setup_hook, ctx)
                        executor: Executor
                        if scenario.mode == "cxp_inject":
                            executor = CXPacketExecutor(dispatcher=runner.dispatcher)
                        else:
                            executor = DirectCallExecutor()
                        output = await executor.execute(scenario, ctx)
                        runner.snapshot_manager.update(
                            scenario.id, output, runner._current_git_commit()
                        )
                        print(f"Updated snapshot for {scenario.id}")
                    finally:
                        await WorktreeIsolation().cleanup(test_id)
            except Exception as exc:
                print(
                    f"Failed to update snapshot for {scenario.id}: {exc}",
                    file=sys.stderr,
                )
                return EXIT_FAILURE
    return EXIT_OK


@traces(spec="102", story="US-002", reqs=["REQ-102-029"])
async def _cmd_report(args: argparse.Namespace) -> int:
    """Execute the ``report`` subcommand."""
    try:
        config = load_config()
    except Exception as exc:
        print(f"Configuration error: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    output_path = Path(args.output)
    try:
        generate_report(config.history_dir, output_path)
        print(f"Report written to {output_path}")
    except Exception as exc:
        print(f"Report generation failed: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR
    return EXIT_OK


@traces(spec="102", story="US-002", reqs=["REQ-102-028"])
async def _cmd_trends(args: argparse.Namespace) -> int:
    """Execute the ``trends`` subcommand."""
    print("Trend Queries (Cypher)")
    print("=" * 40)
    if args.volk:
        print("\nPass rate last 30 days:")
        print(TrendQueries.pass_rate_last_30_days(args.volk))
    print("\nRegression count (example scenario):")
    print(TrendQueries.regression_count("qa-001"))
    print("\nAverage score trend (example scenario):")
    print(TrendQueries.avg_score_trend("qa-001"))
    return EXIT_OK


@traces(spec="102", reqs=["REQ-102-104"])
async def _cmd_config(args: argparse.Namespace) -> int:  # noqa: ARG001
    """Execute the ``config`` subcommand."""
    try:
        config = load_config()
    except Exception as exc:
        print(f"Configuration error: {exc}", file=sys.stderr)
        return EXIT_CONFIG_ERROR

    # Convert dataclass to dict for YAML serialization
    config_dict = {}
    for key in dir(config):
        if key.startswith("_"):
            continue
        value = getattr(config, key)
        if callable(value):
            continue
        if isinstance(value, Path):
            value = str(value)
        config_dict[key] = value

    print(yaml.safe_dump(config_dict, sort_keys=False, default_flow_style=False))
    return EXIT_OK


@traces(spec="102", story="US-001", reqs=["REQ-102-006", "REQ-102-007"])
def main(argv: Sequence[str] | None = None) -> None:
    """Entry point for the harness CLI."""
    _load_dotenv()
    _setup_logging()

    parser = argparse.ArgumentParser(
        prog="harness",
        description="Volk Test Harness — scenario-driven evaluation for autonomous agents.",
    )
    subparsers = parser.add_subparsers(dest="command")

    # run
    run_parser = subparsers.add_parser("run", help="Run scenarios")
    run_parser.add_argument("volk", nargs="?", help="Volk name to run")
    run_parser.add_argument("--scenario", help="Specific scenario ID or file path")
    run_parser.add_argument("--all", action="store_true", help="Run all scenarios")
    run_parser.add_argument(
        "--trials", type=int, default=3, help="Number of trials (default: 3)"
    )

    # list
    list_parser = subparsers.add_parser("list", help="List scenarios")
    list_parser.add_argument("--volk", help="Filter by volk name")

    # snapshot
    snap_parser = subparsers.add_parser("snapshot", help="Snapshot management")
    snap_parser.add_argument(
        "action", choices=["update"], help="Update stored snapshots"
    )
    snap_parser.add_argument("--scenario", help="Specific scenario ID")

    # report
    report_parser = subparsers.add_parser("report", help="Generate HTML report")
    report_parser.add_argument(
        "--output",
        default="harness_report.html",
        help="Output path (default: harness_report.html)",
    )

    # trends
    trends_parser = subparsers.add_parser("trends", help="Show trend data")
    trends_parser.add_argument("--volk", help="Filter by volk name")

    # config
    subparsers.add_parser("config", help="Show effective configuration")

    args = parser.parse_args(argv)

    if args.command is None:
        parser.print_help()
        sys.exit(EXIT_CONFIG_ERROR)

    import asyncio

    try:
        if args.command == "run":
            code = asyncio.run(_cmd_run(args))
        elif args.command == "list":
            code = asyncio.run(_cmd_list(args))
        elif args.command == "snapshot":
            code = asyncio.run(_cmd_snapshot(args))
        elif args.command == "report":
            code = asyncio.run(_cmd_report(args))
        elif args.command == "trends":
            code = asyncio.run(_cmd_trends(args))
        elif args.command == "config":
            code = asyncio.run(_cmd_config(args))
        else:
            parser.print_help()
            code = EXIT_CONFIG_ERROR
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        code = EXIT_CONFIG_ERROR

    sys.exit(code)


if __name__ == "__main__":
    main()