orchestrata_ultimate.harness.history

Reference › orchestrata_ultimate › harness › history

Durable history of harness runs (SPEC-102 US-002), used by the runner. HistoryTracker is the abstract base with one async method, record(result).

JSONHistoryTracker appends each ScenarioResult to a monthly file named by year and month in the history directory. It reads the existing list (treating a corrupt or non-list file as empty, with a warning), appends the new entry and writes the file atomically through a temporary file and os.replace. The files are what reporting reads.

GraphHistoryTracker syncs results to the operational knowledge graph. It connects lazily and resolves its URI from the argument, then KG_ROUTING_URI or KG_URI in the environment or a dotenv file; with no URI the connection raises. record merges a Scenario, a Volk (the volk name resolved to its canonical alias by _canonical_volk_alias), a Run and a Project node with EVALUATED_BY, RESULT_OF and has-scenario relationships, then creates a Regression node for each trial that carries a snapshot diff. Because record must return None, a failed write is logged and stored in last_write_error for the caller to inspect. The runner treats graph recording as optional. Types are in models.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

HistoryTracker (abstract)

Bases: ABC · Line 102

Abstract base for history tracking backends.

JSONHistoryTracker

Bases: HistoryTracker · Decorators: traces(spec='102', story='US-002', reqs=['REQ-102-019', 'REQ-102-020', 'REQ-102-021']) · Line 112

Append results to monthly JSON files with atomic writes.

GraphHistoryTracker

Bases: HistoryTracker · Decorators: traces(spec='102', story='US-002', reqs=['REQ-102-023', 'REQ-102-024', 'REQ-102-026']) · Line 210

Sync results to graph knowledge graph.

Functions

Source

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

"""History tracking for Volk Test Harness — JSON and graph backends."""

from __future__ import annotations

import json
import logging
import os
import tempfile
from abc import ABC, abstractmethod
from collections.abc import Mapping
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import BoltSyncRepository, GraphRepositoryConfig
from orchestrata_ultimate.harness.models import ScenarioResult
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    LABEL_PROJECT,
    LABEL_SCENARIO,
    LABEL_VOLK,
    REL_HAS_SCENARIO,
)

_logger = logging.getLogger(__name__)

# Directory-style volk names -- as used by harness scenario YAML configs
# (``harness/scenarios/<name>/*.yaml``, the ``volk:`` field) and the
# ``agents/volks/<name>`` package layout -- mapped to their canonical KG
# alias (``canonical_schema.CANONICAL_MERGE_KEYS[LABEL_VOLK] == "alias"``).
# Without this, ``ScenarioResult.volk_name`` (e.g. ``"junior"``, ``"qa"``)
# flowed straight into ``MERGE (v:Volk {alias: ...})`` and minted a
# duplicate lowercase Volk node beside the canonical ``'J:Junior'``/
# ``'U:Qa'``/``'U:Ar'`` node already synced by ``kernel/pipeline_kg.py``'s
# boot-time sync (session-254 connectivity audit, task 3).
_VOLK_NAME_TO_ALIAS: dict[str, str] = {
    "analyst": "U:BA",
    "architect": "U:Ar",
    "codedirector": "U:Cd",
    "compliance": "U:Cp",
    "concierge": "U:Cx",
    "documentor": "U:Do",
    "junior": "J:Junior",
    "pm": "U:PM",
    "qa": "U:Qa",
    "senior_dev": "U:SD",
    "ux": "U:Ux",
}


@traces(spec="102", story="US-002", reqs=["REQ-102-024"])
def _canonical_volk_alias(volk_name: str) -> str:
    """Resolve a harness ``volk_name`` (directory-style, e.g. ``"qa"``) to
    its canonical KG alias (e.g. ``"U:Qa"``).

    Values that already look like a canonical alias (contain ``:``) pass
    through unchanged; unknown names also pass through unchanged rather than
    raising, so a volk added to the roster without yet being added to
    ``_VOLK_NAME_TO_ALIAS`` degrades to a plain (non-duplicating-for-known-
    volks) alias instead of a hard failure.
    """
    if ":" in volk_name:
        return volk_name
    return _VOLK_NAME_TO_ALIAS.get(volk_name, volk_name)


@traces(spec="102", story="US-002", reqs=["REQ-102-020"])
def _result_to_dict(result: ScenarioResult) -> dict[str, Any]:
    """Serialize a ScenarioResult to a plain dict suitable for JSON."""
    trials = []
    for trial in result.trials:
        t: dict[str, Any] = {
            "trial_number": trial.trial_number,
            "passed": trial.passed,
            "evaluator_results": [
                {
                    "name": er.name,
                    "passed": er.passed,
                    "score": er.score,
                    "details": er.details,
                }
                for er in trial.evaluator_results
            ],
            "llm_judge_score": trial.llm_judge_score,
            "snapshot_diff": trial.snapshot_diff,
            "healing_hops_used": trial.healing_hops_used,
            "execution_time_ms": trial.execution_time_ms,
            "error": trial.error,
        }
        trials.append(t)

    return {
        "scenario_id": result.scenario_id,
        "volk": result.volk_name,
        "passed": result.passed,
        "aggregate_score": result.aggregate_score,
        "trials": trials,
        "timestamp": result.timestamp.isoformat(),
        "git_commit": result.git_commit,
    }


class HistoryTracker(ABC):
    """Abstract base for history tracking backends."""

    @abstractmethod
    @traces(spec="102", story="US-002", reqs=["REQ-102-019"])
    async def record(self, result: ScenarioResult) -> None:
        """Persist a scenario result."""


@traces(spec="102", story="US-002", reqs=["REQ-102-019", "REQ-102-020", "REQ-102-021"])
class JSONHistoryTracker(HistoryTracker):
    """Append results to monthly JSON files with atomic writes."""

    def __init__(self, history_dir: Path) -> None:
        self.history_dir = history_dir
        self.history_dir.mkdir(parents=True, exist_ok=True)

    @traces(spec="102", story="US-002", reqs=["REQ-102-022"])
    async def record(self, result: ScenarioResult) -> None:
        """Append *result* to the current month's JSON file."""
        month_key = result.timestamp.strftime("%Y-%m")
        target = self.history_dir / f"{month_key}.json"

        entry = _result_to_dict(result)

        records: list[dict[str, Any]] = []
        if target.exists():
            try:
                records = json.loads(target.read_text(encoding="utf-8"))
                if not isinstance(records, list):
                    records = []
            except (json.JSONDecodeError, OSError) as exc:
                _logger.warning("Corrupt history file %s: %s", target, exc)
                records = []

        records.append(entry)

        payload = json.dumps(records, indent=2, ensure_ascii=False)

        # Atomic write: write to temp file in same directory, then rename
        fd, tmp_path = tempfile.mkstemp(
            dir=self.history_dir, prefix=f".{month_key}", suffix=".tmp"
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as fh:
                fh.write(payload)
            os.replace(tmp_path, target)
        except Exception:
            os.unlink(tmp_path)
            raise


# The URI only. NEO4J_URI/NEO4J_USERNAME/NEO4J_PASSWORD were retired with the
# Neo4j containers on 2026-07-27; every remaining graph endpoint negotiates no
# auth, so there is no credential left to recover from .env.
_GRAPH_DOTENV_KEYS = ("KG_ROUTING_URI", "KG_URI")
_GRAPH_DOTENV_PATHS = (".env", "/opt/orchestrata/.env")


def _read_dotenv_graph_vars(
    *,
    keys: tuple[str, ...] = _GRAPH_DOTENV_KEYS,
    paths: tuple[str, ...] = _GRAPH_DOTENV_PATHS,
    environ: Mapping[str, str] | None = None,
) -> dict[str, str]:
    """Read credential *keys* from a local ``.env`` file, if present.

    Mirrors the hand-rolled parser in ``llm/client.py``'s ``_load_dotenv``:
    values are returned as a local mapping (never written into
    ``os.environ``) and only for keys not already set in *environ* (the real
    process environment by default). Harness runs launched outside pytest
    (whose ``tests/conftest.py`` loads ``.env`` via ``python-dotenv`` before
    any test runs) otherwise silently fall back to an empty password, which
    the Neo4j server rejects with "Unsupported authentication token, missing
    key `credentials`" instead of a plain auth failure.

    *keys*, *paths*, and *environ* are overridable (rather than hardcoded to
    ``os.environ``) purely so tests can exercise this parser against scratch
    ``.env`` files and key names without ever mutating the real process
    environment.
    """
    if environ is None:
        environ = os.environ
    found: dict[str, str] = {}
    for candidate in paths:
        path = Path(candidate)
        if not path.exists():
            continue
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
        except OSError as exc:
            _logger.debug("Graph history .env read failed for %s: %s", candidate, exc)
            continue
        for line in lines:
            stripped = line.strip()
            if not stripped or stripped.startswith("#") or "=" not in stripped:
                continue
            key, _, value = stripped.partition("=")
            key = key.strip()
            if key not in keys or key in environ or key in found:
                continue
            value = value.strip().strip('"').strip("'")
            if value:
                found[key] = value
    return found


@traces(spec="102", story="US-002", reqs=["REQ-102-023", "REQ-102-024", "REQ-102-026"])
class GraphHistoryTracker(HistoryTracker):
    """Sync results to graph knowledge graph."""

    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
    ) -> None:
        dotenv_vars = _read_dotenv_graph_vars()
        # No fallback URI: localhost never runs the graph in this infra (Incus
        # containers), so a default would mask misconfiguration with a
        # guaranteed-dead endpoint. Unset → _connect raises.
        self._uri = (
            uri
            or os.environ.get("KG_ROUTING_URI")
            or os.environ.get("KG_URI")
            or dotenv_vars.get("KG_ROUTING_URI")
            or dotenv_vars.get("KG_URI")
        )
        # No credentials: every remaining graph endpoint negotiates no auth.
        self._user = username or ""
        self._password = password or ""
        self._repo: BoltSyncRepository | None = None
        #: str(exc) of the most recent failed write, None when the last
        #: ``record`` call wrote cleanly. ``record`` returns None by
        #: protocol, so this is how a swallowed failure reaches the caller.
        self.last_write_error: str | None = None

    @traces(spec="102", story="US-002", reqs=["REQ-102-114"])
    def _connect(self) -> None:
        """Lazy-connect to graph."""
        if self._repo is not None:
            return
        if not self._uri:
            raise RuntimeError(
                "KG_ROUTING_URI (or KG_URI) is not set (constructor arg, environment, or .env) — "
                "GraphHistoryTracker cannot connect"
            )
        repo = BoltSyncRepository(
            GraphRepositoryConfig(uri=self._uri, auth=(self._user, self._password))
        )
        repo.verify_connectivity()
        self._repo = repo

    def _close(self) -> None:
        """Close driver if open."""
        if self._repo:
            self._repo.close()
            self._repo = None

    @traces(spec="102", story="US-002", reqs=["REQ-102-025", "REQ-102-114"])
    async def record(self, result: ScenarioResult) -> None:
        """Persist *result* to graph, creating/merging nodes and relationships.

        ``record`` implements the :class:`HistoryTracker` protocol and so must
        return None — it cannot report a write failure through its return
        value the way the other graph writers now do. Instead it records the
        failure on :attr:`last_write_error` (None when the last call wrote
        cleanly). Without that, a swallowed regression write left the caller
        unable to distinguish "no regressions" from "the regression write
        failed", which is how contention on the shared graph surfaces later as
        an unexplained empty read-back.
        """
        self.last_write_error = None
        try:
            self._connect()
        except Exception as exc:
            _logger.warning("Graph history sync skipped: %s", exc)
            self.last_write_error = str(exc)
            return

        timestamp_iso = result.timestamp.isoformat()

        # Project slug resolution mirrors the runtime writers'
        # ORCHESTRATA_PROJECT_ID / "orchestrata-ultimate" default
        # (tools/context_tools.py), since this tracker has no project-dir
        # context of its own to resolve against.
        project_id = os.environ.get("ORCHESTRATA_PROJECT_ID", "orchestrata-ultimate")

        # Build Cypher parameters
        params: dict[str, Any] = {
            "scenario_id": result.scenario_id,
            "volk_name": _canonical_volk_alias(result.volk_name),
            "passed": result.passed,
            "aggregate_score": result.aggregate_score,
            "execution_time_ms": result.execution_time_ms,
            "timestamp": timestamp_iso,
            "git_commit": result.git_commit,
            "project_id": project_id,
        }

        cypher = (
            f"MERGE (s:{LABEL_SCENARIO} {{id: $scenario_id}}) "
            f"MERGE (v:{LABEL_VOLK} {{alias: $volk_name}}) "
            "MERGE (r:Run {timestamp: $timestamp}) "
            "SET s.passed = $passed, "
            "    s.aggregate_score = $aggregate_score, "
            "    s.execution_time_ms = $execution_time_ms, "
            "    s.git_commit = $git_commit "
            "SET v.name = coalesce(v.name, $volk_name) "
            "SET r.passed = $passed, "
            "    r.aggregate_score = $aggregate_score, "
            "    r.execution_time_ms = $execution_time_ms, "
            "    r.git_commit = $git_commit "
            "MERGE (s)-[:EVALUATED_BY]->(v) "
            "MERGE (r)-[:RESULT_OF]->(s) "
            f"MERGE (p:{LABEL_PROJECT} {{slug: $project_id}}) "
            f"MERGE (p)-[:{REL_HAS_SCENARIO}]->(s) "
        )

        repo = self._repo
        if repo is None:
            _logger.error("Graph driver is not connected")
            return
        try:
            repo.execute_write(cypher, params)
        except Exception as exc:
            _logger.error("Graph history write failed: %s", exc)
            self.last_write_error = str(exc)
            return

        # Record regressions for any trial with a snapshot diff
        for trial in result.trials:
            if trial.snapshot_diff:
                reg_params: dict[str, Any] = {
                    "scenario_id": result.scenario_id,
                    "diff": trial.snapshot_diff,
                    "timestamp": timestamp_iso,
                    "trial_number": trial.trial_number,
                }
                reg_cypher = (
                    "MERGE (s:Scenario {id: $scenario_id}) "
                    "CREATE (reg:Regression {scenario_id: $scenario_id, diff: $diff, "
                    "timestamp: $timestamp, trial_number: $trial_number}) "
                    "MERGE (reg)-[:DETECTED_IN]->(s) "
                )
                try:
                    repo.execute_write(reg_cypher, reg_params)
                except Exception as exc:
                    _logger.error("Graph regression write failed: %s", exc)
                    self.last_write_error = str(exc)