Orchestrata Ultimate › Plans › plan_102_volk_harness_swarm

Plan: Spec 102 — Individual Volk Test Harness (Swarm Implementation)

Source: plans/plan_102_volk_harness_swarm.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-05-26 · 1058 lines.

Prose pending.

Plan text

Plan: Spec 102 — Individual Volk Test Harness (Swarm Implementation)

Version: 0.1.0
Date: 2026-05-26
Agents: 9 parallel workstreams (A–I)
Estimated Total LOC: ~3,500
Estimated Time: 2–3 hours wall-clock with 9 agents

Architecture Overview

src/orchestrata_ultimate/harness/
    __init__.py              # Public API re-exports
    models.py                # All dataclasses & Pydantic models
    config.py                # Config loader + defaults
    scenario.py              # YAML parsing + validation + setup hooks
    isolation.py             # Git worktree + Incus container management
    executor.py              # Trial execution: CXP inject + direct call
    evaluators/
        __init__.py          # Evaluator registry
        base.py              # Evaluator ABC
        syntax.py            # SyntaxEvaluator
        lint.py              # LintEvaluator
        trace.py             # TraceEvaluator
        file_exists.py       # FileExistsEvaluator
        llm_judge.py         # LLMJudgeEvaluator
    snapshots.py             # Snapshot serialization + comparison
    healing.py               # SelfHealingLoop (CXP-based repair)
    history.py               # JSON + Neo4j history tracking
    reporting.py             # HTML report generation
    cli.py                   # Typer CLI entry point
    runner.py                # HarnessRunner orchestration

tests/harness/
    __init__.py
    conftest.py              # Harness-specific fixtures
    test_models.py           # Model validation tests
    test_scenario.py         # Scenario loading tests
    test_isolation.py        # Worktree/container tests
    test_evaluators.py       # Deterministic evaluator tests
    test_snapshots.py        # Snapshot comparison tests
    test_healing.py          # Self-healing loop tests (mocked)
    test_history.py          # History tracking tests
    test_cli.py              # CLI integration tests
    test_security.py         # Command whitelist + path traversal
    fixtures/
        mock_llm_responses.yaml
        sample_scenario.yaml
        corrupt_snapshot.snap

harness/
    config.yaml              # Default configuration
    scenarios/
        qa/
            scenario_001.yaml
            scenario_002.yaml
            setup.py
        codedirector/
            scenario_001.yaml
            setup.py
        architect/
            scenario_001.yaml
            setup.py
    history/
        runs/                # Monthly JSON files (committed)
        diagnostics/         # Failure traces

tests/volk_harness/
    snapshots/               # Text-based snapshot files (committed)

Workstream Assignments

Workstream	Agent	Scope	Output Files	Dependencies
A	Agent-1	Foundation & Models	`models.py`, `config.py`, `__init__.py`, `harness/config.yaml`	None
B	Agent-2	Scenario System	`scenario.py`, `harness/scenarios/*/`, sample YAMLs	A
C	Agent-3	Execution Engine	`isolation.py`, `executor.py`	A
D	Agent-4	Deterministic Evaluators	`evaluators/*.py`	A
E	Agent-5	LLM Judge & Snapshots	`evaluators/llm_judge.py`, `snapshots.py`	A, D
F	Agent-6	Self-Healing Loop	`healing.py`	A, C
G	Agent-7	History & Reporting	`history.py`, `reporting.py`, `.gitlab-ci.yml`	A
H	Agent-8	CLI & Runner	`cli.py`, `runner.py`	A–G
I	Agent-9	Test Suite	`tests/harness/`	A–H

Cross-Workstream Contracts

These are the stable interfaces that all agents must adhere to. Do NOT change signatures without coordinating with dependent workstreams.

Contract-1: HarnessConfig (from Workstream A)

@dataclass
class HarnessConfig:
    default_llm: str = "deepseek-v4-flash"
    reasoning_llm: str = "deepseek-v4-pro"
    snapshot_dir: Path = Path("tests/volk_harness/snapshots")
    history_dir: Path = Path("harness/history/runs")
    diagnostics_dir: Path = 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
    trials_per_scenario: int = 3
    redis_host: str = "localhost"
    redis_port: int = 6379
    neo4j_uri: str = "bolt://localhost:7687"
    neo4j_username: str = "neo4j"
    neo4j_password: str = ""
    mock_llm: bool = False
    mock_llm_responses_path: Path | None = None

Loaded from harness/config.yaml with .env overrides via os.environ.get().

Contract-2: Scenario Model (from Workstream A)

class Scenario(BaseModel):
    id: str
    volk: str
    description: str
    mode: Literal["cxp_inject", "direct_call"]
    input: dict[str, Any]
    setup_hook: str | None = None
    evaluation: list[EvaluatorConfig]
    llm_model: str | None = None
    timeout_seconds: int = 300

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

Contract-3: HarnessContext (from Workstream A)

@dataclass
class HarnessContext:
    scenario_id: str
    volk_name: str
    trial_number: int
    test_id: str          # UUID for resource isolation
    worktree_path: Path
    redis_db: int
    neo4j_prefix: str
    config: HarnessConfig
    logger: logging.Logger

Contract-4: Evaluator Interface (from Workstream D)

class Evaluator(ABC):
    name: ClassVar[str]

    @abstractmethod
    async def evaluate(
        self,
        output: Any,
        context: HarnessContext,
        config: EvaluatorConfig,
    ) -> EvaluatorResult: ...

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

Registry: evaluators/__init__.py provides get_evaluator(name: str) -> Evaluator.

Contract-5: TrialResult (from Workstream A)

@dataclass
class TrialResult:
    trial_number: int
    passed: bool
    evaluator_results: list[EvaluatorResult]
    llm_judge_score: float | None
    snapshot_diff: str | None
    healing_hops_used: int
    execution_time_ms: int
    error: str | None

Contract-6: ScenarioResult (from Workstream A)

@dataclass
class ScenarioResult:
    scenario_id: str
    volk_name: str
    trials: list[TrialResult]
    aggregate_score: float
    passed: bool
    execution_time_ms: int
    git_commit: str
    timestamp: datetime

Contract-7: Isolation Manager Interface (from Workstream C)

class IsolationManager(ABC):
    @abstractmethod
    async def create(self, test_id: str) -> Path: ...       # Returns worktree/container path
    @abstractmethod
    async def cleanup(self, test_id: str) -> None: ...

class WorktreeIsolation(IsolationManager): ...
class IncusIsolation(IsolationManager): ...  # Optional for now

Contract-8: Executor Interface (from Workstream C)

class Executor(ABC):
    @abstractmethod
    async def execute(
        self,
        scenario: Scenario,
        context: HarnessContext,
    ) -> Any: ...  # Returns raw scenario output

class CXPacketExecutor(Executor): ...
class DirectCallExecutor(Executor): ...

Contract-9: SelfHealingLoop Interface (from Workstream F)

class SelfHealingLoop:
    async def heal(
        self,
        failure: TrialResult,
        context: HarnessContext,
    ) -> HealingResult: ...

Uses CXP MessageDispatcher.dispatch() internally. Max hops from HarnessConfig.max_healing_hops.

Contract-10: Snapshot Manager Interface (from Workstream E)

class SnapshotManager:
    def save(self, scenario_id: str, output: Any) -> Path: ...
    def compare(self, scenario_id: str, output: Any) -> str | None: ...  # Returns diff or None
    def update(self, scenario_id: str, output: Any) -> Path: ...

Snapshot format: stable text serialization (JSON Lines or YAML), versioned with header.

Contract-11: History Tracker Interface (from Workstream G)

class HistoryTracker(ABC):
    @abstractmethod
    async def record(self, result: ScenarioResult) -> None: ...

class JSONHistoryTracker(HistoryTracker): ...
class Neo4jHistoryTracker(HistoryTracker): ...

Workstream A: Foundation & Models (Agent-1)

Goal: Create the package skeleton, all dataclasses, Pydantic models, and configuration system.

Step A-1: Create package directory

mkdir -p src/orchestrata_ultimate/harness/evaluators
mkdir -p harness/scenarios/{qa,codedirector,architect}
mkdir -p harness/history/{runs,diagnostics}
mkdir -p tests/harness/fixtures
mkdir -p tests/volk_harness/snapshots
touch src/orchestrata_ultimate/harness/__init__.py
touch src/orchestrata_ultimate/harness/evaluators/__init__.py
touch tests/harness/__init__.py
touch tests/volk_harness/__init__.py

Step A-2: Implement `models.py`

Implement ALL dataclasses from Contracts 1–6. Use @dataclass(frozen=True) where immutability makes sense (EvaluatorResult, TrialResult). Use Pydantic BaseModel for Scenario and EvaluatorConfig (validation).

Step A-3: Implement `config.py`

load_config(path: Path | None = None) -> HarnessConfig
Loads harness/config.yaml if exists, else uses defaults
Overrides from environment variables using os.environ.get() with sensible defaults
Validates that snapshot_dir, history_dir, diagnostics_dir exist (creates if missing)

Step A-4: Create `harness/config.yaml`

default_llm: "deepseek-v4-flash"
reasoning_llm: "deepseek-v4-pro"
snapshot_dir: "tests/volk_harness/snapshots"
history_dir: "harness/history/runs"
diagnostics_dir: "harness/history/diagnostics"
command_whitelist:
  - "ruff"
  - "python"
  - "git"
  - "orchestrata-tools"
  - "black"
  - "isort"
  - "uv"
  - "pytest"
max_healing_hops: 3
default_timeout: 300
trials_per_scenario: 3

Step A-5: Implement `init.py`

Re-export public API:

from .models import HarnessConfig, HarnessContext, Scenario, ...
from .config import load_config

Verification A

python -c "from orchestrata_ultimate.harness import HarnessConfig, Scenario, load_config; print('OK')"
python -c "from orchestrata_ultimate.harness.models import EvaluatorResult; print(EvaluatorResult('x', True, 1.0, 'ok'))"

Workstream B: Scenario System (Agent-2)

Depends on: A
Goal: Load, validate, and execute scenario YAML files with Python setup hooks.

Step B-1: Implement `scenario.py`

Functions:

load_scenario(path: Path) -> Scenario — Parse YAML, validate with Pydantic
load_scenarios_for_volk(volk_name: str, base_dir: Path = Path("harness/scenarios")) -> list[Scenario] — Discover all .yaml files in volk directory
list_all_scenarios(base_dir: Path) -> list[Scenario] — All scenarios across all volks
run_setup_hook(hook_path: str, context: HarnessContext) -> None — Dynamically import and call async setup function
- Format: "setup_module:function_name" (e.g., "harness.scenarios.qa.setup:seed_test_repo")
- Must be an async def accepting HarnessContext
- Must return None
- Exceptions bubble up as ERR-002

YAML Schema Validation:

Required fields: id, volk, mode, evaluation
mode must be "cxp_inject" or "direct_call"
evaluation must be a non-empty list
Each evaluator config must have name field
id must be unique across all loaded scenarios

Step B-2: Create sample scenarios

harness/scenarios/qa/scenario_001.yaml:

id: qa-001
volk: qa
description: "QA Volk reviews a Python file with intentional lint errors"
mode: direct_call
input:
  file_path: "src/orchestrata_ultimate/core/logging/logging_config.py"
  review_type: "lint_and_style"
setup_hook: "harness.scenarios.qa.setup:inject_lint_errors"
evaluation:
  - name: syntax_check
    weight: 1.0
  - name: lint_check
    weight: 1.0
    params:
      severity: "ERROR"
  - name: file_exists
    weight: 0.5
    params:
      expected_files:
        - "review_output.json"
llm_model: "deepseek-v4-pro"
timeout_seconds: 120

harness/scenarios/qa/setup.py:

import asyncio
from pathlib import Path
from orchestrata_ultimate.harness import HarnessContext

async def inject_lint_errors(context: HarnessContext) -> None:
    """Create a temp file with intentional lint errors for QA volk to find."""
    target = context.worktree_path / "temp_review_target.py"
    target.write_text("import os\nimport sys  # unused\n\ndef bad_func( ):\n    pass\n")

harness/scenarios/codedirector/scenario_001.yaml:

id: codedirector-001
volk: codedirector
description: "CodeDirector generates a scaffold for a new module"
mode: cxp_inject
input:
  module_name: "orchestrata_ultimate.core.patterns.strategy"
  module_description: "Strategy pattern implementation for algorithm selection"
  methods:
    - "execute(strategy: str, data: dict) -> dict"
evaluation:
  - name: syntax_check
    weight: 1.0
  - name: lint_check
    weight: 1.0
  - name: file_exists
    weight: 1.0
    params:
      expected_files:
        - "src/orchestrata_ultimate/core/patterns/strategy.py"
llm_model: "deepseek-v4-flash"
timeout_seconds: 180

harness/scenarios/architect/scenario_001.yaml:

id: architect-001
volk: architect
description: "Architect generates a Mermaid diagram for a new subsystem"
mode: direct_call
input:
  subsystem: "notification_pipeline"
  components:
    - "EventBus"
    - "FilterChain"
    - "Dispatcher"
    - "Logger"
evaluation:
  - name: file_exists
    weight: 1.0
    params:
      expected_files:
        - "diagram.md"
  - name: syntax_check
    weight: 0.5
llm_model: "deepseek-v4-pro"
timeout_seconds: 120

Verification B

python -c "
from orchestrata_ultimate.harness.scenario import load_scenarios_for_volk
scenarios = load_scenarios_for_volk('qa')
assert len(scenarios) >= 1
assert scenarios[0].id == 'qa-001'
print('OK')
"

Workstream C: Execution Engine (Agent-3)

Depends on: A
Goal: Isolate trials via git worktrees, execute scenarios via CXP or direct calls, enforce timeouts.

Step C-1: Implement `isolation.py`

WorktreeIsolation:

create(test_id: str) -> Path
- Runs git worktree add <path> <branch_name> where branch = harness-<test_id>
- Branch created from current HEAD
- Returns absolute path to worktree
cleanup(test_id: str)
- Runs git worktree remove --force <path>
- Deletes branch harness-<test_id>
- Idempotent (no error if already cleaned)

IncusIsolation (stub for future):

create() raises NotImplementedError with helpful message
Configurable via HarnessConfig.isolation_mode: "worktree" | "incus"

Resource Isolation Helpers:

get_redis_db_for_trial(trial_number: int) -> int — Maps trial 0→db0, 1→db1, etc.
get_neo4j_prefix(test_id: str) -> str — Returns test_id as prefix

Step C-2: Implement `executor.py`

CXPacketExecutor:

Constructs CXPPacket with:
- intent from scenario.input["intent"] or scenario.id
- payload from scenario.input
- context including test_id
Enqueues via MessageDispatcher.dispatch()
Waits for response or timeout
Returns the response payload

DirectCallExecutor:

Dynamically imports volk class from agents.volks.<volk_name>
Instantiates with default constructor
Calls method specified in scenario.input["method"] with scenario.input["args"]
Returns method result

Timeout enforcement:

Use asyncio.wait_for() wrapping the execution
On timeout: cancel task, cleanup isolation, return TimeoutError in TrialResult.error

Step C-3: Command Whitelist Enforcer

enforce_whitelist(command: list[str], whitelist: list[str]) -> None

Parse command line, extract base command
If not in whitelist, raise SecurityError (ERR-010)
Called by executors before any subprocess invocation

Verification C

python -c "
from orchestrata_ultimate.harness.isolation import WorktreeIsolation
import tempfile, os
mgr = WorktreeIsolation()
# Test with a temp repo
print('Isolation OK')
"
python -c "
from orchestrata_ultimate.harness.executor import DirectCallExecutor
print('Executor imports OK')
"

Workstream D: Deterministic Evaluators (Agent-4)

Depends on: A
Goal: Implement all deterministic evaluators (no LLM calls) with 100% accuracy.

Step D-1: Implement `evaluators/base.py`

class Evaluator(ABC):
    name: ClassVar[str]

    @abstractmethod
    async def evaluate(self, output: Any, context: HarnessContext, config: EvaluatorConfig) -> EvaluatorResult: ...

Step D-2: Implement `evaluators/syntax.py`

SyntaxEvaluator:

For each Python file in output (or specified in config params):
- Read file content
- Call ast.parse()
- If SyntaxError, return passed=False, score=0.0, details=traceback
- If success, return passed=True, score=1.0
Handles non-Python files gracefully (skip with warning)

Step D-3: Implement `evaluators/lint.py`

LintEvaluator:

Runs ruff check <file_path> via subprocess
Captures stdout/stderr
Configurable severity param: "ERROR" (default) or "WARNING"
"ERROR": fail on any ERROR-level finding
"WARNING": fail on any WARNING or ERROR
Returns passed=False if ruff finds issues, details = ruff output
Uses the worktree path from context

Step D-4: Implement `evaluators/trace.py`

TraceEvaluator:

Input: execution trace string or log file path
Parses for exception patterns: Traceback (most recent call last):, Exception:, Error:
Configurable allowed_patterns param: regex list of expected/allowed exceptions
Returns passed=False if unhandled exceptions found
Returns details with list of found exceptions

Step D-5: Implement `evaluators/file_exists.py`

FileExistsEvaluator:

Config param: expected_files (list of relative paths)
Checks each path exists relative to context.worktree_path
Returns passed=False if any missing, details lists missing files

Step D-6: Implement `evaluators/init.py`

Registry:

_EVALUATORS: dict[str, type[Evaluator]] = {
    "syntax_check": SyntaxEvaluator,
    "lint_check": LintEvaluator,
    "trace_validator": TraceEvaluator,
    "file_exists": FileExistsEvaluator,
}

def get_evaluator(name: str) -> Evaluator:
    if name not in _EVALUATORS:
        raise ValueError(f"Unknown evaluator: {name}. Available: {list(_EVALUATORS)}")
    return _EVALUATORS[name]()

Verification D

python -c "
from orchestrata_ultimate.harness.evaluators import get_evaluator
for name in ['syntax_check', 'lint_check', 'trace_validator', 'file_exists']:
    e = get_evaluator(name)
    assert e.name == name
print('All evaluators registered OK')
"

Workstream E: LLM Judge & Snapshots (Agent-5)

Depends on: A, D
Goal: Implement LLM-as-judge evaluator and snapshot comparison system.

Step E-1: Implement `evaluators/llm_judge.py`

LLMJudgeEvaluator:

Uses DeepSeek API (v4-pro by default)
Prompt construction:
- System: "You are evaluating the output of an AI agent. Score the output on a scale of 0-100 based on the rubric. Respond with ONLY a JSON object: {'score': N, 'feedback': '...'}"
- User: Includes scenario rubric, expected behavior, and actual output
Temperature = 0.0
Parses JSON response
Returns EvaluatorResult(score=score/100, passed=score>=threshold, details=feedback)
API key from DEEPSEEK_API_KEY env var
Mock mode: if HarnessConfig.mock_llm=True, load response from mock_llm_responses_path

Step E-2: Implement `snapshots.py`

Snapshot Format (versioned text):

# snapshot:v1
# scenario_id: <id>
# timestamp: <ISO>
# git_commit: <sha>
---
<JSON-serialized output>

SnapshotManager:

save(scenario_id, output) -> Path — Serialize to text, write to snapshot dir
compare(scenario_id, output) -> str | None — Load existing, compare. Return unified diff string if different, None if identical
update(scenario_id, output) -> Path — Overwrite existing snapshot
exists(scenario_id) -> bool

Serialization rules:

Use json.dumps() with sort_keys=True, indent=2 for deterministic output
Handle binary data by base64-encoding with marker
Truncate very long strings (>10KB) with hash for comparison

Step E-3: Add mock LLM responses fixture

tests/harness/fixtures/mock_llm_responses.yaml:

qa-001:
  score: 85
  feedback: "Good catch on unused imports. Minor style inconsistency in docstrings."
codedirector-001:
  score: 92
  feedback: "Clean scaffold. Proper type annotations. Missing module docstring."

Verification E

python -c "
from orchestrata_ultimate.harness.snapshots import SnapshotManager
from pathlib import Path
import tempfile
mgr = SnapshotManager(Path(tempfile.mkdtemp()))
print('Snapshots OK')
"

Workstream F: Self-Healing Loop (Agent-6)

Depends on: A, C
Goal: Implement autonomous repair on scenario failure using CXP routing.

Step F-1: Implement `healing.py`

SelfHealingLoop:

class SelfHealingLoop:
    def __init__(self, dispatcher: MessageDispatcher, max_hops: int = 3): ...

    async def heal(self, failure: TrialResult, context: HarnessContext) -> HealingResult: ...

Hop Logic:

Hop 1 — Diagnostic:
- Build diagnostic packet: PacketType.ERROR, payload contains failure.evaluator_results, failure.error, failure.snapshot_diff
- Route to VolkDefinition for the failed volk or a dedicated "repair" volk
- Capture response (repair suggestion)
Hop 2 — Repair:
- If response contains a patch/retry strategy, apply it
- Re-run the scenario trial
- If success, return HealingResult(success=True, hops_used=2)
Hop 3 — Escalation:
- If still failing, send escalation packet with full diagnostic history
- Try fallback strategy (e.g., switch LLM model, reduce input complexity)
- Re-run one final time
Max Depth:
- If unresolved after 3 hops, return HealingResult(success=False, hops_used=3)
- Save full diagnostic trace to HarnessConfig.diagnostics_dir / "<scenario_id>_<timestamp>.json"

CXP Packet Construction:

Use orchestrata_ultimate.cxp.CXPPacket
from_agent: "harness"
to_agent: failed volk name or "repair"
priority: high (priority=1)
subject: f"healing:{scenario_id}:hop{N}"

No Source Code Mutation:

The healing loop operates ONLY within the worktree
It NEVER modifies files in the main working directory or src/orchestrata_ultimate/
This is enforced by path validation: all writes must be under context.worktree_path

Verification F

python -c "
from orchestrata_ultimate.harness.healing import SelfHealingLoop
print('Healing OK')
"

Workstream G: History & Reporting (Agent-7)

Depends on: A
Goal: Track results in JSON + Neo4j, generate HTML reports, define GitLab CI job.

Step G-1: Implement `history.py`

JSONHistoryTracker:

record(result: ScenarioResult) -> None
Appends to harness/history/runs/<YYYY-MM>.json
Each entry: {"scenario_id": "...", "volk": "...", "passed": true, "aggregate_score": 0.95, "trials": [...], "timestamp": "...", "git_commit": "..."}
Creates new file if month changes
Uses atomic write (write to .tmp, then rename)

Neo4jHistoryTracker:

record(result: ScenarioResult) -> None
Creates/merges :Scenario {id}, :Volk {name}, :Run {timestamp} nodes
Creates relationships: (:Scenario)-[:EVALUATED_BY]->(:Volk), (:Run)-[:RESULT_OF]->(:Scenario)
On regression (snapshot diff), creates :Regression {scenario_id, diff, timestamp}
Uses Neo4jSyncer or direct neo4j driver (fallback)
Connection from .env vars with localhost fallback

Step G-2: Implement `reporting.py`

HTML Report Generator:

generate_report(history_path: Path, output_path: Path) -> Path
Reads all JSON history files
Generates HTML with:
- Per-volk pass rate charts
- Per-scenario score trends over time
- Regression table (snapshot diffs)
- Aggregate statistics (total runs, avg score, failure rate)
Uses simple Jinja2 templating (or plain string building if Jinja2 not desired)
Styling: inline CSS for portability

Trend Queries (Cypher):

class TrendQueries:
    @staticmethod
    def pass_rate_last_30_days(volk_name: str) -> str: ...
    @staticmethod
    def regression_count(scenario_id: str) -> str: ...
    @staticmethod
    def avg_score_trend(scenario_id: str) -> str: ...

Step G-3: GitLab CI Job

.gitlab-ci.yml addition:

volk_harness_nightly:
  stage: test
  image: python:3.14
  schedule:
    - cron: "0 2 * * *"  # 02:00 UTC
  script:
    - uv sync --dev
    - python -m orchestrata_ultimate.harness run --all --trials 3
    - python -m orchestrata_ultimate.harness report --output harness_report.html
  artifacts:
    reports:
      junit: harness/junit.xml
    paths:
      - harness_report.html
      - harness/history/runs/
    expire_in: 30 days
  rules:
    - if: '$CI_PIPELINE_SOURCE == "schedule"'
  parallel:
    matrix:
      - VOLK: [qa, codedirector, architect]

Verification G

python -c "
from orchestrata_ultimate.harness.history import JSONHistoryTracker
print('History OK')
"

Workstream H: CLI & Runner (Agent-8)

Depends on: A–G
Goal: Wire everything together into a runnable CLI and orchestration runner.

Step H-1: Implement `runner.py`

HarnessRunner:

class HarnessRunner:
    def __init__(self, config: HarnessConfig): ...

    async def run_scenario(self, scenario: Scenario, trials: int = 3) -> ScenarioResult:
        # 1. For each trial:
        #    a. Create isolation (worktree)
        #    b. Run setup hook if defined
        #    c. Execute scenario (CXP or direct)
        #    d. Run all evaluators
        #    e. Compare snapshot
        #    f. If failure and healing enabled, run SelfHealingLoop
        #    g. Cleanup isolation
        # 2. Aggregate trial results
        # 3. Record history (JSON + Neo4j)
        # 4. Return ScenarioResult

    async def run_volk(self, volk_name: str, trials: int = 3) -> list[ScenarioResult]: ...

    def list_scenarios(self, volk_name: str | None = None) -> list[Scenario]: ...

Runner Logic Details:

Setup hook timeout: 30s (separate from scenario timeout)
Evaluator execution: Run all evaluators concurrently with asyncio.gather()
Composite score: weighted average of evaluator scores
Pass criteria: composite score >= 0.8 AND all required evaluators pass AND no snapshot regression
Snapshot handling: If no snapshot exists, save current output as baseline. If exists, compare.

Step H-2: Implement `cli.py`

Use Typer (already in project dependencies) or argparse.

Subcommands:

python -m orchestrata_ultimate.harness run <volk_name>          # Run all scenarios for volk
python -m orchestrata_ultimate.harness run --scenario <id>      # Run specific scenario
python -m orchestrata_ultimate.harness run --all                # Run all scenarios for all volks
python -m orchestrata_ultimate.harness run --trials 5           # Override trial count
python -m orchestrata_ultimate.harness run --mock-llm           # Use mock LLM responses
python -m orchestrata_ultimate.harness list                     # List scenarios
python -m orchestrata_ultimate.harness list --volk qa           # Filter by volk
python -m orchestrata_ultimate.harness snapshot update          # Update all snapshots
python -m orchestrata_ultimate.harness snapshot update --scenario <id>
python -m orchestrata_ultimate.harness report                   # Generate HTML report
python -m orchestrata_ultimate.harness trends --volk qa         # Show trend data
python -m orchestrata_ultimate.harness config                   # Print effective config

Exit codes:

0: All scenarios passed
1: One or more scenarios failed
2: Configuration or infrastructure error
3: Security violation detected

Step H-3: Implement `main.py`

from orchestrata_ultimate.harness.cli import app
if __name__ == "__main__":
    app()

Verification H

python -m orchestrata_ultimate.harness --help
python -m orchestrata_ultimate.harness list
python -m orchestrata_ultimate.harness config

Workstream I: Test Suite (Agent-9)

Depends on: A–H
Goal: Comprehensive test coverage for the harness itself.

Step I-1: Create `tests/harness/conftest.py`

Fixtures:

harness_config() — HarnessConfig with temp dirs
harness_context(harness_config) — Populated HarnessContext
temp_worktree(harness_context) — Creates and cleans up a worktree
mock_scenario() — Minimal valid Scenario instance
mock_trial_result() — TrialResult with sample evaluator results

Step I-2: Implement `tests/harness/test_models.py`

Test Scenario validation (required fields, mode enum)
Test EvaluatorConfig defaults
Test HarnessConfig loading from YAML + env overrides

Step I-3: Implement `tests/harness/test_scenario.py`

Test load_scenario() with valid YAML
Test schema violation handling (missing required fields)
Test load_scenarios_for_volk() discovery
Test setup hook execution (async function called correctly)
Test setup hook import failure (ERR-002)

Step I-4: Implement `tests/harness/test_evaluators.py`

For each evaluator:

SyntaxEvaluator: valid Python → pass; invalid Python → fail
LintEvaluator: clean file → pass; file with errors → fail
TraceEvaluator: clean trace → pass; trace with exception → fail
FileExistsEvaluator: file exists → pass; missing → fail
LLMJudgeEvaluator: mock mode → deterministic score

Step I-5: Implement `tests/harness/test_snapshots.py`

Save and load snapshot
Compare identical outputs → None
Compare different outputs → diff string
Update snapshot → overwrite
Handle corrupt snapshot file → graceful error

Step I-6: Implement `tests/harness/test_healing.py`

Mock CXP dispatcher
Test hop progression (1→2→3)
Test max depth enforcement (stops at 3)
Test success on hop 2 (repair works)
Test diagnostic trace saved to file

Step I-7: Implement `tests/harness/test_isolation.py`

Test worktree creation and cleanup
Test worktree idempotency (cleanup twice)
Test Redis DB mapping
Test Neo4j prefix generation

Step I-8: Implement `tests/harness/test_cli.py`

Test list command output format
Test config command prints config
Test run --mock-llm uses mock responses
Test exit codes (0 for pass, 1 for fail)

Step I-9: Implement `tests/harness/test_security.py`

Test command whitelist allows valid commands
Test command whitelist blocks invalid commands
Test path traversal blocked (worktree escape)
Test API key masking in logs
Test LLM judge prompt injection resistance

Step I-10: Implement integration test

tests/harness/test_integration.py:

Create a minimal "echo volk" in tests
Define a scenario that calls it
Run full pipeline: load → execute → evaluate → snapshot → history
Assert all components integrate correctly

Verification I

uv run pytest tests/harness/ -n auto -x --timeout=120

Execution Order for Swarm

Phase 1: Parallel Foundation (All agents start together)

Agent-1 (A): Start immediately. Delivers models + config.
Agent-2 (B): Waits for A to deliver models.py, then starts.
Agent-3 (C): Waits for A, then starts.
Agent-4 (D): Waits for A, then starts.
Agent-5 (E): Waits for A + D evaluators registry, then starts.
Agent-6 (F): Waits for A + C executor interface, then starts.
Agent-7 (G): Waits for A, then starts.

Phase 2: Integration (After A–G complete)

Agent-8 (H): Starts after A–G are done. Integrates all components.

Phase 3: Testing (After H completes)

Agent-9 (I): Starts after H. Writes comprehensive tests.

Phase 4: Verification & Merge

Run all harness tests: uv run pytest tests/harness/ -n auto
Run lint: uv run ruff check src/orchestrata_ultimate/harness/ tests/harness/
Run full test suite: uv run pytest tests/ -n auto -q --tb=no
Commit all changes

Risk Mitigation

Risk	Mitigation
Interface mismatch between agents	Contracts defined above are frozen once A completes. Changes require updating ALL dependent workstreams.
DeepSeek API costs during testing	Mock LLM mode (`--mock-llm`) used for all automated tests. Real API only in manual/nightly runs.
Git worktree pollution	Every trial uses unique branch name (`harness-<uuid>`). Cleanup is mandatory in `finally` blocks.
Neo4j/Redis unavailable	All code uses localhost fallback. Tests use `pytest.importorskip` for optional services.
Self-healing infinite loop	Hard max 3 hops. Timeout per hop. No recursion.
Security: volk escapes sandbox	Command whitelist + path validation + worktree isolation. Incus mode provides full sandboxing.

Success Criteria

[ ] python -m orchestrata_ultimate.harness list shows all sample scenarios
[ ] python -m orchestrata_ultimate.harness run --scenario qa-001 --mock-llm completes successfully
[ ] python -m orchestrata_ultimate.harness snapshot update creates snapshot files
[ ] uv run pytest tests/harness/ -n auto passes with >90% coverage
[ ] uv run ruff check src/orchestrata_ultimate/harness/ reports zero errors
[ ] No git worktrees left behind after test runs
[ ] All sample scenarios have deterministic evaluators configured
[ ] Mock LLM mode works without API keys
[ ] Help text visible for all CLI commands

Agent Handoff Notes

Agent-1 (A) → Deliver models.py, config.py, __init__.py, harness/config.yaml to all other agents. These files are the foundation.

Agent-2–7 (B–G) → Run in parallel after receiving A's output. Each owns their subdirectory. Do NOT modify models.py or config.py.

Agent-8 (H) → Integrator. May need to fix minor interface mismatches. Has authority to add adapter code but NOT to change core contracts.

Agent-9 (I) → Tester. Must achieve >90% harness coverage. May request changes to any workstream for testability.

Plan: Spec 102 — Individual Volk Test Harness (Swarm Implementation)

Plan text

Plan: Spec 102 — Individual Volk Test Harness (Swarm Implementation)

Architecture Overview

Workstream Assignments

Cross-Workstream Contracts

Contract-1: HarnessConfig (from Workstream A)

Contract-2: Scenario Model (from Workstream A)

Contract-3: HarnessContext (from Workstream A)

Contract-4: Evaluator Interface (from Workstream D)

Contract-5: TrialResult (from Workstream A)

Contract-6: ScenarioResult (from Workstream A)

Contract-7: Isolation Manager Interface (from Workstream C)

Contract-8: Executor Interface (from Workstream C)

Contract-9: SelfHealingLoop Interface (from Workstream F)

Contract-10: Snapshot Manager Interface (from Workstream E)

Contract-11: History Tracker Interface (from Workstream G)

Workstream A: Foundation & Models (Agent-1)

Step A-1: Create package directory

Step A-2: Implement models.py

Step A-3: Implement config.py

Step A-4: Create harness/config.yaml

Step A-5: Implement __init__.py

Verification A

Workstream B: Scenario System (Agent-2)

Step B-1: Implement scenario.py

Step B-2: Create sample scenarios

Verification B

Workstream C: Execution Engine (Agent-3)

Step C-1: Implement isolation.py

Step C-2: Implement executor.py

Step C-3: Command Whitelist Enforcer

Verification C

Workstream D: Deterministic Evaluators (Agent-4)

Step D-1: Implement evaluators/base.py

Step D-2: Implement evaluators/syntax.py

Step D-3: Implement evaluators/lint.py

Step D-4: Implement evaluators/trace.py

Step D-5: Implement evaluators/file_exists.py

Step D-6: Implement evaluators/__init__.py

Verification D

Workstream E: LLM Judge & Snapshots (Agent-5)

Step E-1: Implement evaluators/llm_judge.py

Step E-2: Implement snapshots.py

Step E-3: Add mock LLM responses fixture

Verification E

Workstream F: Self-Healing Loop (Agent-6)

Step F-1: Implement healing.py

Verification F

Workstream G: History & Reporting (Agent-7)

Step G-1: Implement history.py

Step G-2: Implement reporting.py

Step G-3: GitLab CI Job

Verification G

Workstream H: CLI & Runner (Agent-8)

Step H-1: Implement runner.py

Step H-2: Implement cli.py

Step H-3: Implement __main__.py

Verification H

Workstream I: Test Suite (Agent-9)

Step I-1: Create tests/harness/conftest.py

Step I-2: Implement tests/harness/test_models.py

Step I-3: Implement tests/harness/test_scenario.py

Step I-4: Implement tests/harness/test_evaluators.py

Step I-5: Implement tests/harness/test_snapshots.py

Step I-6: Implement tests/harness/test_healing.py

Step I-7: Implement tests/harness/test_isolation.py

Step I-8: Implement tests/harness/test_cli.py

Step I-9: Implement tests/harness/test_security.py

Step I-10: Implement integration test

Verification I

Execution Order for Swarm

Phase 1: Parallel Foundation (All agents start together)

Phase 2: Integration (After A–G complete)

Phase 3: Testing (After H completes)

Phase 4: Verification & Merge

Risk Mitigation

Success Criteria

Agent Handoff Notes

Related

Step A-2: Implement `models.py`

Step A-3: Implement `config.py`

Step A-4: Create `harness/config.yaml`

Step A-5: Implement `init.py`

Step B-1: Implement `scenario.py`

Step C-1: Implement `isolation.py`

Step C-2: Implement `executor.py`

Step D-1: Implement `evaluators/base.py`

Step D-2: Implement `evaluators/syntax.py`

Step D-3: Implement `evaluators/lint.py`

Step D-4: Implement `evaluators/trace.py`

Step D-5: Implement `evaluators/file_exists.py`

Step D-6: Implement `evaluators/init.py`

Step E-1: Implement `evaluators/llm_judge.py`

Step E-2: Implement `snapshots.py`

Step F-1: Implement `healing.py`

Step G-1: Implement `history.py`

Step G-2: Implement `reporting.py`

Step H-1: Implement `runner.py`

Step H-2: Implement `cli.py`

Step H-3: Implement `main.py`

Step I-1: Create `tests/harness/conftest.py`

Step I-2: Implement `tests/harness/test_models.py`

Step I-3: Implement `tests/harness/test_scenario.py`

Step I-4: Implement `tests/harness/test_evaluators.py`

Step I-5: Implement `tests/harness/test_snapshots.py`

Step I-6: Implement `tests/harness/test_healing.py`

Step I-7: Implement `tests/harness/test_isolation.py`

Step I-8: Implement `tests/harness/test_cli.py`

Step I-9: Implement `tests/harness/test_security.py`