orchestrata_ultimate.harness.snapshots

Reference › orchestrata_ultimate › harness › snapshots

Regression detection by snapshot (SPEC-102 US-004), used by the runner. SnapshotManager stores one text file per scenario in the snapshot directory, named after the scenario id with path separators replaced. Each file has a header (snapshot version, scenario id, timestamp, git commit) and, after a --- line, a deterministic JSON body produced by _serialize, which first normalises special types in _transform_value. save and update write the file, exists tests for it, and compare returns None when there is no baseline or the output is identical, and otherwise a unified diff. A trial whose diff is not None counts as a regression and fails. The snapshot update command of the cli refreshes stored baselines.

At a glance

Class diagram

Dependency diagram

API

Classes

SnapshotManager

Decorators: traces(spec='102', story='US-004', reqs=['REQ-102-050', 'REQ-102-053', 'REQ-102-055']) · Line 20

Manage text-based, versioned snapshots for scenario outputs.

Functions

Source

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

"""Snapshot serialization and comparison for the Volk Test Harness."""

from __future__ import annotations

import base64
import difflib
import hashlib
import json
from datetime import UTC, date, datetime
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces

_SNAPSHOT_VERSION = "v1"
_LONG_STRING_THRESHOLD = 10_240


@traces(spec="102", story="US-004", reqs=["REQ-102-050", "REQ-102-053", "REQ-102-055"])
class SnapshotManager:
    """Manage text-based, versioned snapshots for scenario outputs."""

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

    # ------------------------------------------------------------------ #
    # Public API
    # ------------------------------------------------------------------ #

    @traces(spec="102", story="US-004", reqs=["REQ-102-050"])
    def save(self, scenario_id: str, output: Any, git_commit: str = "") -> Path:
        """Serialize *output* and write a new snapshot file.

        Returns the path to the saved snapshot.
        """
        path = self._snapshot_path(scenario_id)
        serialized = _serialize(output)
        header = self._build_header(scenario_id, git_commit)
        path.write_text(header + "\n" + serialized + "\n", encoding="utf-8")
        return path

    @traces(spec="102", story="US-004", reqs=["REQ-102-053", "REQ-102-054"])
    def compare(self, scenario_id: str, output: Any) -> str | None:
        """Compare *output* against the stored snapshot.

        Returns ``None`` when:
        - no stored snapshot exists (no baseline),
        - or the outputs are identical.

        Returns a unified diff string when the outputs differ.
        """
        path = self._snapshot_path(scenario_id)
        if not path.exists():
            return None

        stored = self._read_snapshot_body(path)
        if stored is None:
            return None

        current = _serialize(output)
        if stored == current:
            return None

        return self._unified_diff(
            stored.splitlines(keepends=True),
            current.splitlines(keepends=True),
            scenario_id,
        )

    @traces(spec="102", story="US-004", reqs=["REQ-102-055"])
    def update(self, scenario_id: str, output: Any, git_commit: str = "") -> Path:
        """Overwrite an existing snapshot with the current *output*.

        Returns the path to the updated snapshot.
        """
        return self.save(scenario_id, output, git_commit)

    @traces(spec="102", story="US-004", reqs=["REQ-102-051"])
    def exists(self, scenario_id: str) -> bool:
        """Return whether a snapshot exists for *scenario_id*."""
        return self._snapshot_path(scenario_id).exists()

    # ------------------------------------------------------------------ #
    # Helpers
    # ------------------------------------------------------------------ #

    @traces(spec="102", story="US-004", reqs=["REQ-102-051"])
    def _snapshot_path(self, scenario_id: str) -> Path:
        """Return the filesystem path for a given scenario snapshot."""
        safe_id = scenario_id.replace("/", "_").replace("\\", "_")
        return self.snapshot_dir / f"{safe_id}.snap"

    @traces(spec="102", story="US-004", reqs=["REQ-102-052"])
    def _build_header(self, scenario_id: str, git_commit: str) -> str:
        """Build the snapshot header lines."""
        timestamp = datetime.now(UTC).isoformat()
        lines = [
            f"# snapshot:{_SNAPSHOT_VERSION}",
            f"# scenario_id: {scenario_id}",
            f"# timestamp: {timestamp}",
            f"# git_commit: {git_commit}",
            "---",
        ]
        return "\n".join(lines)

    @traces(spec="102", story="US-004", reqs=["REQ-102-118"])
    def _read_snapshot_body(self, path: Path) -> str | None:
        """Read the JSON body from a snapshot file, skipping the header."""
        try:
            text = path.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            return None

        # Find the ``---`` delimiter that separates header from body.
        lines = text.splitlines(keepends=True)
        delim_idx: int | None = None
        for i, line in enumerate(lines):
            if line.rstrip("\n") == "---":
                delim_idx = i
                break

        if delim_idx is None:
            return None

        body = "".join(lines[delim_idx + 1 :])
        return body.strip("\n")

    @traces(spec="102", story="US-004", reqs=["REQ-102-054"])
    def _unified_diff(
        self,
        old_lines: list[str],
        new_lines: list[str],
        scenario_id: str,
    ) -> str:
        """Generate a unified diff between two line lists."""
        return "".join(
            difflib.unified_diff(
                old_lines,
                new_lines,
                fromfile=f"{scenario_id}.snap",
                tofile=f"{scenario_id}.current",
            ),
        )


# ---------------------------------------------------------------------- #
# Serialization helpers
# ---------------------------------------------------------------------- #


@traces(spec="102", story="US-004", reqs=["REQ-102-056"])
def _serialize(value: Any) -> str:
    """Serialize *value* to a deterministic JSON string."""
    return json.dumps(
        _transform_value(value),
        sort_keys=True,
        indent=2,
        ensure_ascii=False,
    )


@traces(spec="102", story="US-004", reqs=["REQ-102-056"])
def _transform_value(value: Any) -> Any:
    """Recursively transform special types before JSON serialization."""
    if isinstance(value, (datetime, date)):
        return value.isoformat()

    if isinstance(value, bytes):
        return {
            "__binary__": True,
            "data": base64.b64encode(value).decode("ascii"),
        }

    if isinstance(value, str) and len(value) > _LONG_STRING_THRESHOLD:
        return {
            "__truncated__": True,
            "length": len(value),
            "hash": hashlib.sha256(value.encode("utf-8")).hexdigest(),
            "preview": value[:256],
        }

    if isinstance(value, dict):
        return {k: _transform_value(v) for k, v in value.items()}

    if isinstance(value, (list, tuple)):
        return [_transform_value(item) for item in value]

    return value