orchestrata_ultimate.harness.isolation

Reference › orchestrata_ultimate › harness › isolation

Per-trial isolation for the harness (SPEC-102 US-006), so trials cannot contaminate each other or standing infrastructure. IsolationManager is the abstract base: async create(test_id) returns an absolute path, and async cleanup(test_id) destroys the environment and is idempotent. It is used by the runner and the cli.

WorktreeIsolation resolves the repository root with git, creates a branch harness-<test_id> from the current HEAD and adds a worktree under .harness/worktrees/<test_id>. IncusIsolation requires the incus executable (it raises RuntimeError naming the worktree alternative if absent), launches an ephemeral container named oh-trial-<slug> from the image in INCUS_HARNESS_IMAGE (default images:ubuntu/22.04), and bind-mounts a per-trial host directory under .harness/incus/ at /workspace; the returned path is the host directory. Cleanup force-deletes the container and removes the directory. Container names are sanitised to lowercase letters, digits and hyphens, at most 63 characters; the prefix keeps trial containers apart from standing ones. Failures surface incus's own stderr.

Two helpers assign the other resources: get_redis_db_for_trial maps a zero-based trial number to the Redis database of the same number, and get_graph_prefix returns the test_id as the graph resource prefix.

At a glance

Class diagram

Dependency diagram

API

Classes

IsolationManager (abstract)

Bases: ABC · Line 19

Abstract base for trial isolation strategies.

WorktreeIsolation

Bases: IsolationManager · Decorators: traces(spec='102', story='US-006', reqs=['REQ-102-079', 'REQ-102-081', 'REQ-102-083']) · Line 34

Isolation backed by a git worktree under .harness/worktrees/.

IncusIsolation

Bases: IsolationManager · Decorators: traces(spec='102', story='US-006', reqs=['REQ-102-084', 'REQ-102-085', 'REQ-102-086']) · Line 137

Isolation backed by an Incus container.

Functions

Source

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

"""Trial isolation via git worktrees and Incus containers."""

from __future__ import annotations

import logging
import os
import re
import shutil
import subprocess
from abc import ABC, abstractmethod
from collections.abc import Mapping
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces

_logger = logging.getLogger(__name__)


class IsolationManager(ABC):
    """Abstract base for trial isolation strategies."""

    @abstractmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-079", "REQ-102-084"])
    async def create(self, test_id: str) -> Path:
        """Create an isolated environment and return its absolute path."""

    @abstractmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-081"])
    async def cleanup(self, test_id: str) -> None:
        """Destroy the isolated environment. Idempotent — no-op if already gone."""


@traces(spec="102", story="US-006", reqs=["REQ-102-079", "REQ-102-081", "REQ-102-083"])
class WorktreeIsolation(IsolationManager):
    """Isolation backed by a git worktree under ``.harness/worktrees/``."""

    @traces(spec="102", story="US-006", reqs=["REQ-102-079"])
    async def create(self, test_id: str) -> Path:
        """Create a git worktree for *test_id* and return its absolute path.

        Steps:
        1. Resolve the repository root via ``git rev-parse --show-toplevel``.
        2. Create a branch named ``harness-<test_id>`` from the current HEAD.
        3. Add a worktree at ``<repo>/.harness/worktrees/<test_id>`` pointing to
           the new branch.
        """
        repo_root = self._git_toplevel()
        worktree_path = repo_root / ".harness" / "worktrees" / test_id
        branch_name = f"harness-{test_id}"

        worktree_path.parent.mkdir(parents=True, exist_ok=True)

        # Create branch from current HEAD
        self._run_git(["git", "branch", branch_name], cwd=repo_root, check=True)

        # Add worktree
        self._run_git(
            ["git", "worktree", "add", str(worktree_path), branch_name],
            cwd=repo_root,
            check=True,
        )

        return worktree_path.resolve()

    @traces(spec="102", story="US-006", reqs=["REQ-102-081"])
    async def cleanup(self, test_id: str) -> None:
        """Remove the worktree and delete its branch. Silently succeeds if
        either has already been removed."""
        try:
            repo_root = self._git_toplevel()
        except subprocess.CalledProcessError:
            _logger.warning("Not inside a git repository; skipping worktree cleanup.")
            return

        worktree_path = repo_root / ".harness" / "worktrees" / test_id
        branch_name = f"harness-{test_id}"

        # Remove worktree if it still exists
        if worktree_path.exists():
            self._run_git(
                ["git", "worktree", "remove", "--force", str(worktree_path)],
                cwd=repo_root,
                check=False,
            )
            self._run_git(
                ["git", "worktree", "prune"],
                cwd=repo_root,
                check=False,
            )

        # Delete branch if it still exists
        branch_list = self._run_git(
            ["git", "branch", "--list", branch_name],
            cwd=repo_root,
            check=False,
            capture_output=True,
            text=True,
        )
        if branch_name in (branch_list.stdout or ""):
            self._run_git(
                ["git", "branch", "-D", branch_name],
                cwd=repo_root,
                check=False,
            )

    @staticmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-079"])
    def _git_toplevel() -> Path:
        result = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True,
            text=True,
            check=True,
        )
        return Path(result.stdout.strip())

    @staticmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-079", "REQ-102-081"])
    def _run_git(
        cmd: list[str],
        *,
        cwd: Path | None = None,
        check: bool = True,
        capture_output: bool = True,
        text: bool = True,
    ) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            cmd,
            cwd=cwd,
            check=check,
            capture_output=capture_output,
            text=text,
        )


@traces(spec="102", story="US-006", reqs=["REQ-102-084", "REQ-102-085", "REQ-102-086"])
class IncusIsolation(IsolationManager):
    """Isolation backed by an Incus container.

    ``create`` launches an ephemeral container named ``oh-trial-<slug>`` from a
    base image (``INCUS_HARNESS_IMAGE``, default ``images:ubuntu/22.04``) and
    bind-mounts a per-trial host directory into it at ``/workspace``. The returned
    path is that host directory: the caller stages files there (visible inside the
    container) and drives the container with ``incus exec``. The ``oh-trial-``
    prefix guarantees trial containers never collide with standing infrastructure.
    """

    _NAME_PREFIX = "oh-trial-"
    _MOUNT_POINT = "/workspace"

    @traces(spec="102", story="US-006", reqs=["REQ-102-084"])
    async def create(
        self, test_id: str, *, env: Mapping[str, str] | None = None
    ) -> Path:
        env = env or os.environ
        if shutil.which("incus", path=env.get("PATH")) is None:
            raise RuntimeError(
                "incus executable not found; install Incus or use "
                "WorktreeIsolation (isolation_mode='worktree')."
            )
        container = self._container_name(test_id)
        host_dir = self._host_dir(test_id)
        host_dir.mkdir(parents=True, exist_ok=True)

        image = env.get("INCUS_HARNESS_IMAGE", "images:ubuntu/22.04")
        # --ephemeral: the container is auto-deleted when stopped, so an
        # interrupted trial cannot leave a lingering instance.
        self._run_incus(
            ["incus", "launch", image, container, "--ephemeral"], check=True, env=env
        )
        self._run_incus(
            [
                "incus",
                "config",
                "device",
                "add",
                container,
                "workspace",
                "disk",
                f"source={host_dir}",
                f"path={self._MOUNT_POINT}",
            ],
            check=True,
            env=env,
        )
        return host_dir.resolve()

    @traces(spec="102", story="US-006", reqs=["REQ-102-084"])
    async def cleanup(
        self, test_id: str, *, env: Mapping[str, str] | None = None
    ) -> None:
        """Destroy the container and its host directory. Idempotent — deleting a
        container that is already gone is swallowed."""
        env = env or os.environ
        if shutil.which("incus", path=env.get("PATH")) is not None:
            container = self._container_name(test_id)
            self._run_incus(
                ["incus", "delete", container, "--force"], check=False, env=env
            )
        shutil.rmtree(self._host_dir(test_id), ignore_errors=True)

    @classmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-084"])
    def _container_name(cls, test_id: str) -> str:
        """Derive a valid, collision-safe Incus instance name from *test_id*.

        Incus names allow only lowercase letters, digits and hyphens, must start
        with a letter, and are capped at 63 chars.
        """
        slug = re.sub(r"[^a-z0-9-]+", "-", test_id.lower()).strip("-")
        slug = re.sub(r"-{2,}", "-", slug)
        name = f"{cls._NAME_PREFIX}{slug}" if slug else cls._NAME_PREFIX.rstrip("-")
        return name[:63].rstrip("-")

    @staticmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-084"])
    def _host_dir(test_id: str) -> Path:
        try:
            repo_root = WorktreeIsolation._git_toplevel()
        except subprocess.CalledProcessError:
            repo_root = Path.cwd()
        return repo_root / ".harness" / "incus" / test_id

    @staticmethod
    @traces(spec="102", story="US-006", reqs=["REQ-102-084"])
    def _run_incus(
        cmd: list[str],
        *,
        check: bool = True,
        env: Mapping[str, str] | None = None,
    ) -> subprocess.CompletedProcess[str]:
        # capture_output hides incus' own diagnostics, so on failure re-raise with
        # the daemon's stderr (e.g. "The requested image couldn't be found")
        # instead of an opaque CalledProcessError.
        proc = subprocess.run(cmd, capture_output=True, text=True, check=False, env=env)
        if check and proc.returncode != 0:
            detail = (proc.stderr or proc.stdout or "").strip()
            raise RuntimeError(f"incus command failed ({' '.join(cmd)}): {detail}")
        return proc


@traces(spec="102", story="US-006", reqs=["REQ-102-092"])
def get_redis_db_for_trial(trial_number: int) -> int:
    """Map a trial number to an isolated Redis DB index.

    Trial 0 → DB 0, trial 1 → DB 1, etc.
    """
    return trial_number


@traces(spec="102", story="US-006", reqs=["REQ-102-093"])
def get_graph_prefix(test_id: str) -> str:
    """Return the Neo4j resource prefix for a given test run.

    The prefix is simply the *test_id* itself.
    """
    return test_id