orchestrata_ultimate.harness.models

Reference › orchestrata_ultimate › harness › models

The shared data model of the Volk Test Harness (SPEC-102). Nearly every other harness module imports it. There are two Pydantic models and a set of dataclasses.

• EvaluatorConfig names an evaluator in a scenario, with params, a weight (default 1.0) and a threshold.
• Scenario describes one test: id, target volk, description, a mode (cxp_inject or direct_call), an input dictionary, an optional setup hook path, the evaluator list, an optional model override and a timeout (default 300 seconds).
• EvaluatorResult, TrialResult and ScenarioResult are frozen dataclasses for the three levels of outcome: one evaluator, one trial (with evaluator results, snapshot diff, healing hops used, time and error) and all trials of a scenario (aggregate score, pass flag, git commit, timestamp).
• HarnessContext is the per-trial runtime context: scenario id, volk, trial number, a test_id, the worktree path, the Redis database number, the graph resource prefix, the config and a logger.
• HarnessConfig holds the defaults: models, snapshot, history and diagnostics directories, the command whitelist, max_healing_hops (3), timeouts, trials_per_scenario (3), Redis and graph endpoints, and isolation_mode (worktree or incus). Redis and graph endpoints resolve through the core factories, which raise when a key is unset rather than falling back to a literal.
• HopDiagnostic and HealingResult record the self-healing loop.
• Evaluator is the base class with a name class variable and an async evaluate method; concrete evaluators are in the evaluators package.

At a glance

Class diagram

Dependency diagram

API

Classes

EvaluatorConfig (pydantic)

Bases: BaseModel · Decorators: traces(spec='102', story='US-005', reqs=['REQ-102-076', 'REQ-102-077']) · Line 26

Configuration for a single evaluator within a scenario.

Scenario (pydantic)

Bases: BaseModel · Decorators: traces(spec='102', story='US-001', reqs=['REQ-102-002', 'REQ-102-003', 'REQ-102-014']) · Line 42

A single test scenario for evaluating a volk.

EvaluatorResult (dataclass)

Decorators: dataclass(frozen=True) · Line 62

Result of a single evaluator run.

TrialResult (dataclass)

Decorators: dataclass(frozen=True) · Line 72

Result of a single scenario trial.

ScenarioResult (dataclass)

Decorators: dataclass(frozen=True) · Line 86

Aggregated result across all trials of a scenario.

HarnessContext (dataclass)

Decorators: dataclass · Line 100

Runtime context passed to setup hooks, evaluators, and executors.

HarnessConfig (dataclass)

Decorators: dataclass, traces(spec='102', story='US-001', reqs=['REQ-102-107', 'REQ-102-120']) · Line 116

Harness configuration loaded from harness/config.yaml.

HopDiagnostic (dataclass)

Decorators: dataclass(frozen=True) · Line 169

Diagnostic record for a single healing hop.

HealingResult (dataclass)

Decorators: dataclass(frozen=True) · Line 181

Result of a self-healing loop attempt.

Evaluator

Decorators: traces(spec='102', story='US-005', reqs=['REQ-102-064', 'REQ-102-065', 'REQ-102-066', 'REQ-102-067', 'REQ-102-070']) · Line 200

Base class for deterministic and LLM evaluators.

Source

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

"""Core dataclasses and Pydantic models for the Volk Test Harness."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, ClassVar, Literal

from pydantic import BaseModel, Field

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_uri
from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)

# ---------------------------------------------------------------------------
# Pydantic models (validation-heavy, loaded from YAML)
# ---------------------------------------------------------------------------


@traces(spec="102", story="US-005", reqs=["REQ-102-076", "REQ-102-077"])
class EvaluatorConfig(BaseModel):
    """Configuration for a single evaluator within a scenario."""

    name: str
    params: dict[str, Any] = Field(default_factory=dict)
    weight: float = 1.0
    threshold: float = 0.0


# The scenario dispatch modes, named once. Callers and tests that need to
# annotate a mode value referenced this literal set by re-spelling it, which
# is how it drifted out of one place into several.
ScenarioMode = Literal["cxp_inject", "direct_call"]


@traces(spec="102", story="US-001", reqs=["REQ-102-002", "REQ-102-003", "REQ-102-014"])
class Scenario(BaseModel):
    """A single test scenario for evaluating a volk."""

    id: str
    volk: str
    description: str
    mode: ScenarioMode
    input: dict[str, Any] = Field(default_factory=dict)
    setup_hook: str | None = None
    evaluation: list[EvaluatorConfig] = Field(default_factory=list)
    llm_model: str | None = None
    timeout_seconds: int = 300


# ---------------------------------------------------------------------------
# Dataclasses (lightweight, used at runtime)
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class EvaluatorResult:
    """Result of a single evaluator run."""

    name: str
    passed: bool
    score: float  # 0.0–1.0 for deterministic, 0–100 for judge
    details: str


@dataclass(frozen=True)
class TrialResult:
    """Result of a single scenario trial."""

    trial_number: int
    passed: bool
    evaluator_results: list[EvaluatorResult] = field(default_factory=list)
    llm_judge_score: float | None = None
    snapshot_diff: str | None = None
    healing_hops_used: int = 0
    execution_time_ms: int = 0
    error: str | None = None


@dataclass(frozen=True)
class ScenarioResult:
    """Aggregated result across all trials of a scenario."""

    scenario_id: str
    volk_name: str
    trials: list[TrialResult] = field(default_factory=list)
    aggregate_score: float = 0.0
    passed: bool = False
    execution_time_ms: int = 0
    git_commit: str = ""
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))


@dataclass
class HarnessContext:
    """Runtime context passed to setup hooks, evaluators, and executors."""

    scenario_id: str
    volk_name: str
    trial_number: int
    test_id: str  # UUID for resource isolation
    worktree_path: Path
    redis_db: int  # Worker-isolated Redis DB number
    graph_prefix: str  # test_id prefix for Neo4j resources
    config: HarnessConfig
    logger: logging.Logger


@dataclass
@traces(spec="102", story="US-001", reqs=["REQ-102-107", "REQ-102-120"])
class HarnessConfig:
    """Harness configuration loaded from harness/config.yaml."""

    default_llm: str = "deepseek-v4-flash"
    reasoning_llm: str = "deepseek-v4-pro"
    snapshot_dir: Path = field(
        default_factory=lambda: Path("tests/volk_harness/snapshots")
    )
    history_dir: Path = field(default_factory=lambda: Path("harness/history/runs"))
    diagnostics_dir: Path = field(
        default_factory=lambda: Path("harness/history/diagnostics")
    )
    command_whitelist: list[str] = field(
        default_factory=lambda: [
            "ruff",
            "python",
            "git",
            "orchestrata-tools",
            "black",
            "isort",
            "uv",
            "pytest",
        ]
    )
    max_healing_hops: int = 3
    default_timeout: int = 300
    # Budget for a scenario's setup hook. Configurable so the timeout-path
    # test can wait out a REAL TimeoutError in ~1s instead of 30s.
    setup_hook_timeout_seconds: int = 30
    trials_per_scenario: int = 3
    # No literal endpoint default (RUL-003/RUL-011): a bare "localhost:6379"
    # fails silently by connecting to nothing when REDIS_HOST/PORT aren't the
    # target — the resolver raises naming the unset key instead.
    redis_host: str = field(default_factory=resolve_redis_host)
    redis_port: int = field(default_factory=resolve_redis_port)
    graph_uri: str = field(default_factory=resolve_uri)
    # Blank by contract, like graph_password beside it. This was "neo4j" — a
    # credential retired with the Neo4j containers on 2026-07-27 (migration
    # plan Step 5.1); every remaining graph endpoint negotiates no auth, and
    # tests/config.py deliberately aliases GRAPH_USERNAME/GRAPH_PASSWORD to
    # env keys nothing sets so a stale shell export cannot resurrect them.
    # A hardcoded default here was that resurrection, in production.
    graph_username: str = ""
    graph_password: str = ""
    isolation_mode: Literal["worktree", "incus"] = "worktree"


# ---------------------------------------------------------------------------
# Self-healing loop models
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class HopDiagnostic:
    """Diagnostic record for a single healing hop."""

    hop_number: int
    action: str
    # packet_sent and response are stored as dicts to avoid CXP import here
    packet_sent: dict[str, Any] = field(default_factory=dict)
    response: dict[str, Any] | None = None
    error: str | None = None


@dataclass(frozen=True)
class HealingResult:
    """Result of a self-healing loop attempt."""

    success: bool
    hops_used: int
    diagnostics: list[HopDiagnostic] = field(default_factory=list)
    final_trial_result: TrialResult | None = None


# ---------------------------------------------------------------------------
# Evaluator ABC (defined here so evaluators/ can import from models)
# ---------------------------------------------------------------------------


@traces(
    spec="102",
    story="US-005",
    reqs=["REQ-102-064", "REQ-102-065", "REQ-102-066", "REQ-102-067", "REQ-102-070"],
)
class Evaluator:
    """Base class for deterministic and LLM evaluators."""

    name: ClassVar[str] = ""

    @traces(spec="102", story="US-005", reqs=["REQ-102-064"])
    async def evaluate(
        self,
        output: Any,
        context: HarnessContext,
        config: EvaluatorConfig,
    ) -> EvaluatorResult:
        """Evaluate scenario output and return a structured result."""
        raise NotImplementedError