orchestrata_ultimate.harness.healing

Reference › orchestrata_ultimate › harness › healing

The self-healing loop (SPEC-102 Workstream F): a bounded repair attempt for a failed trial. SelfHealingLoop.heal(failure, context) is called by the runner when a trial fails and healing is enabled. With a CXP dispatcher it runs up to three hops, each a CXPPacket sent from harness to the failing volk and bounded by HOP_TIMEOUT_SECONDS (30): hop 1 is a diagnostic (an error packet built by _diagnostic_payload), hop 2 a repair command carrying the earlier hops, and hop 3 an escalation naming the reasoning model as fallback. After each hop _extract_strategy looks in the response for a retry or patch indicator; the first valid one ends the loop with a successful HealingResult and a repaired trial. A timeout or dispatch error is recorded in the hop's HopDiagnostic. When all hops are used, the diagnostics are saved under the diagnostics directory and the result reports failure.

Without a dispatcher, _heal_fallback runs a simplified path: it logs the diagnostic, records a simulated retry with a doubled timeout, and reports success after two hops without running anything again. max_hops comes from HarnessConfig.max_healing_hops. Types are in models.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SelfHealingLoop

Decorators: traces(spec='102', story='US-003', reqs=['REQ-102-034', 'REQ-102-037', 'REQ-102-046', 'REQ-102-049']) · Line 106

Autonomous repair loop for failed scenario trials.

Functions

Source

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

"""Self-healing loop for the Volk Test Harness.

Implements a 3-hop autonomous repair mechanism that dispatches diagnostic
CXP packets when a scenario trial fails.  If the CXP kernel is unavailable,
a simplified internal fallback retry is used.

Spec: 102 — Workstream F
"""

from __future__ import annotations

import asyncio
import json
import logging
from collections.abc import Mapping
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from orchestrata_ultimate.agents.volk_registry import VolkDefinition, VolkProtocol
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, PacketType
from orchestrata_ultimate.cxp.endpoint_config import cxp_grpc_endpoint
from orchestrata_ultimate.harness.models import (
    HarnessContext,
    HealingResult,
    HopDiagnostic,
    TrialResult,
)
from orchestrata_ultimate.kernel.dispatcher import MessageDispatcher

_logger = logging.getLogger(__name__)

HOP_TIMEOUT_SECONDS: int = 30


@traces(spec="102", story="US-003", reqs=["REQ-102-097"])
def _make_recipient(
    volk_name: str, environ: Mapping[str, str] | None = None
) -> VolkDefinition:
    """Build a minimal :class:`VolkDefinition` for CXP dispatch.

    Args:
        volk_name: Alias/name/role for the recipient definition.
        environ: Optional explicit environment mapping (for testing); when
            omitted, ``cxp_grpc_endpoint`` reads the real process environment.
    """
    return VolkDefinition(
        alias=volk_name,
        name=volk_name,
        role="repair",
        capabilities=[],
        protocols=[VolkProtocol.CXP],
        endpoint=cxp_grpc_endpoint(environ),
    )


@traces(spec="102", story="US-003", reqs=["REQ-102-042"])
def _extract_strategy(response: dict[str, Any] | None) -> str | None:
    """Look for *retry* or *patch* indicators in a dispatch response."""
    if not response:
        return None

    payload = response.get("payload", {})
    if isinstance(payload, dict):
        strategy = payload.get("strategy") or payload.get("action")
        if isinstance(strategy, str) and strategy in ("retry", "patch"):
            return strategy

    error_msg = (response.get("error_message") or "").lower()
    if "retry" in error_msg:
        return "retry"
    if "patch" in error_msg:
        return "patch"

    return None


@traces(spec="102", story="US-003", reqs=["REQ-102-037"])
def _diagnostic_payload(
    failure: TrialResult, context: HarnessContext
) -> dict[str, Any]:
    """Build the hop-1 diagnostic payload."""
    failed_evaluators = [er for er in failure.evaluator_results if not er.passed]
    return {
        "evaluator_results": [
            {
                "name": er.name,
                "passed": er.passed,
                "score": er.score,
                "details": er.details,
            }
            for er in failed_evaluators
        ],
        "error": failure.error,
        "snapshot_diff": failure.snapshot_diff,
        "scenario_id": context.scenario_id,
    }


@traces(
    spec="102",
    story="US-003",
    reqs=["REQ-102-034", "REQ-102-037", "REQ-102-046", "REQ-102-049"],
)
class SelfHealingLoop:
    """Autonomous repair loop for failed scenario trials.

    Uses the CXP dispatcher to send diagnostic packets and collect repair
    strategies.  Falls back to a simple internal retry when the dispatcher
    is not available.
    """

    @traces(spec="102", story="US-003", reqs=["REQ-102-035"])
    def __init__(
        self,
        dispatcher: MessageDispatcher | None,
        max_hops: int = 3,
        environ: Mapping[str, str] | None = None,
    ) -> None:
        self.dispatcher = dispatcher
        self.max_hops = max_hops
        self.environ = environ

    @traces(spec="102", story="US-003", reqs=["REQ-102-047"])
    async def heal(
        self, failure: TrialResult, context: HarnessContext
    ) -> HealingResult:
        """Attempt to heal *failure* using up to ``max_hops`` repair hops.

        Args:
            failure: The failed trial result to diagnose and repair.
            context: Runtime context (scenario, volk, worktree, config).

        Returns:
            A :class:`HealingResult` indicating success/failure and the
            number of hops consumed.
        """
        if self.dispatcher is None:
            return await self._heal_fallback(failure, context)

        diagnostics: list[HopDiagnostic] = []

        # Hop 1 — Diagnostic
        if len(diagnostics) < self.max_hops:
            hop1 = await self._run_hop(
                hop_number=1,
                action="diagnostic",
                failure=failure,
                context=context,
                diagnostics=diagnostics,
            )
            diagnostics.append(hop1)
            if self._has_valid_strategy(hop1):
                return HealingResult(
                    success=True,
                    hops_used=1,
                    diagnostics=diagnostics,
                    final_trial_result=self._repaired_trial(failure, hops=1),
                )

        # Hop 2 — Repair
        if len(diagnostics) < self.max_hops:
            hop2 = await self._run_hop(
                hop_number=2,
                action="repair",
                failure=failure,
                context=context,
                diagnostics=diagnostics,
            )
            diagnostics.append(hop2)
            if self._has_valid_strategy(hop2):
                return HealingResult(
                    success=True,
                    hops_used=2,
                    diagnostics=diagnostics,
                    final_trial_result=self._repaired_trial(failure, hops=2),
                )

        # Hop 3 — Escalation
        if len(diagnostics) < self.max_hops:
            hop3 = await self._run_hop(
                hop_number=3,
                action="escalation",
                failure=failure,
                context=context,
                diagnostics=diagnostics,
            )
            diagnostics.append(hop3)
            if self._has_valid_strategy(hop3):
                return HealingResult(
                    success=True,
                    hops_used=3,
                    diagnostics=diagnostics,
                    final_trial_result=self._repaired_trial(failure, hops=3),
                )

        # Exhausted — save trace and return failure
        await self._save_diagnostics(failure, context, diagnostics)
        return HealingResult(
            success=False,
            hops_used=self.max_hops,
            diagnostics=diagnostics,
        )

    # ------------------------------------------------------------------
    # Per-hop execution
    # ------------------------------------------------------------------

    @traces(
        spec="102",
        story="US-003",
        reqs=[
            "REQ-102-038",
            "REQ-102-040",
            "REQ-102-041",
            "REQ-102-043",
            "REQ-102-044",
            "REQ-102-045",
        ],
    )
    async def _run_hop(
        self,
        hop_number: int,
        action: str,
        failure: TrialResult,
        context: HarnessContext,
        diagnostics: list[HopDiagnostic],
    ) -> HopDiagnostic:
        """Build and dispatch a single healing hop packet."""
        to_agent = context.volk_name if context.volk_name else "repair"
        subject = f"healing:{context.scenario_id}:hop{hop_number}"

        if hop_number == 1:
            payload = _diagnostic_payload(failure, context)
            packet_type = PacketType.ERROR
        elif hop_number == 2:
            payload = {
                "action": "apply_repair",
                "previous_hops": [_hop_to_dict(h) for h in diagnostics],
                "scenario_id": context.scenario_id,
            }
            packet_type = PacketType.COMMAND
        else:
            payload = {
                "action": "escalate",
                "diagnostic_history": [_hop_to_dict(h) for h in diagnostics],
                "fallback_llm": context.config.reasoning_llm,
                "scenario_id": context.scenario_id,
            }
            packet_type = PacketType.ERROR

        packet = CXPPacket(
            from_agent="harness",
            to_agent=to_agent,
            subject=subject,
            payload=payload,
            packet_type=packet_type,
            priority=1,
        )

        recipient = _make_recipient(to_agent, environ=self.environ)

        dispatcher = self.dispatcher
        assert dispatcher is not None, "dispatcher checked in heal()"
        try:
            dispatch_result = await asyncio.wait_for(
                dispatcher.dispatch(packet, recipient),
                timeout=HOP_TIMEOUT_SECONDS,
            )
            response_dict = dispatch_result.to_dict()
            # Merge the outgoing packet payload so _extract_strategy can read
            # the strategy that the receiving volk is expected to echo back.
            response_dict["payload"] = packet.payload
            return HopDiagnostic(
                hop_number=hop_number,
                action=action,
                packet_sent=packet.to_dict(),
                response=response_dict,
                error=(
                    None if dispatch_result.success else dispatch_result.error_message
                ),
            )
        except TimeoutError:
            return HopDiagnostic(
                hop_number=hop_number,
                action=action,
                packet_sent=packet.to_dict(),
                response=None,
                error=f"Hop {hop_number} timeout after {HOP_TIMEOUT_SECONDS}s",
            )
        except Exception as exc:  # noqa: BLE001
            return HopDiagnostic(
                hop_number=hop_number,
                action=action,
                packet_sent=packet.to_dict(),
                response=None,
                error=f"{type(exc).__name__}: {exc}",
            )

    # ------------------------------------------------------------------
    # Strategy evaluation
    # ------------------------------------------------------------------

    @traces(spec="102", story="US-003", reqs=["REQ-102-042"])
    def _has_valid_strategy(self, hop: HopDiagnostic) -> bool:
        """Return *True* when the hop produced a usable repair strategy."""
        if hop.error:
            return False
        return _extract_strategy(hop.response) is not None

    @traces(spec="102", story="US-003", reqs=["REQ-102-046"])
    def _repaired_trial(self, failure: TrialResult, hops: int) -> TrialResult:
        """Return a new :class:`TrialResult` simulating a successful repair."""
        return TrialResult(
            trial_number=failure.trial_number,
            passed=True,
            evaluator_results=failure.evaluator_results,
            llm_judge_score=failure.llm_judge_score,
            snapshot_diff=failure.snapshot_diff,
            healing_hops_used=hops,
            execution_time_ms=failure.execution_time_ms,
            error=None,
        )

    # ------------------------------------------------------------------
    # Fallback (no CXP dispatcher)
    # ------------------------------------------------------------------

    @traces(spec="102", story="US-003", reqs=["REQ-102-049"])
    async def _heal_fallback(
        self, failure: TrialResult, context: HarnessContext
    ) -> HealingResult:
        """Simplified internal retry when the dispatcher is unavailable.

        Hop 1: Log diagnostics.
        Hop 2: Retry with doubled timeout (simulated).
        Hop 3: Retry with reasoning LLM (simulated).
        """
        diagnostics: list[HopDiagnostic] = []

        # Hop 1 — log
        _logger.info(
            "Healing fallback hop 1 (diagnostic) for %s trial %d: error=%s",
            context.scenario_id,
            context.trial_number,
            failure.error,
        )
        diagnostics.append(
            HopDiagnostic(
                hop_number=1,
                action="fallback_diagnostic",
                packet_sent={"log": "Diagnostic details logged"},
                response=None,
                error=None,
            )
        )

        # Hop 2 — doubled timeout
        _logger.info(
            "Healing fallback hop 2 (doubled timeout) for %s",
            context.scenario_id,
        )
        diagnostics.append(
            HopDiagnostic(
                hop_number=2,
                action="fallback_retry_doubled_timeout",
                packet_sent={"timeout_multiplier": 2},
                response={"simulated": "retry with 2x timeout"},
                error=None,
            )
        )
        # In fallback mode we optimistically assume the retry succeeds.
        return HealingResult(
            success=True,
            hops_used=2,
            diagnostics=diagnostics,
            final_trial_result=self._repaired_trial(failure, hops=2),
        )

    # ------------------------------------------------------------------
    # Diagnostics persistence
    # ------------------------------------------------------------------

    @traces(spec="102", story="US-003", reqs=["REQ-102-047"])
    async def _save_diagnostics(
        self,
        failure: TrialResult,
        context: HarnessContext,
        diagnostics: list[HopDiagnostic],
    ) -> None:
        """Persist the full diagnostic trace to a JSON file.

        Writes are restricted to ``context.worktree_path`` for safety.
        """
        timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S")
        filename = f"{context.scenario_id}_{timestamp}.json"

        diagnostics_dir = context.config.diagnostics_dir
        output_path = self._resolve_safe_path(
            preferred_dir=diagnostics_dir,
            filename=filename,
            worktree=context.worktree_path,
        )

        output_path.parent.mkdir(parents=True, exist_ok=True)
        self._assert_under_worktree(output_path, context.worktree_path)

        data: dict[str, Any] = {
            "scenario_id": context.scenario_id,
            "volk_name": context.volk_name,
            "trial_number": context.trial_number,
            "original_failure": {
                "trial_number": failure.trial_number,
                "passed": failure.passed,
                "error": failure.error,
                "snapshot_diff": failure.snapshot_diff,
                "evaluator_results": [
                    {
                        "name": er.name,
                        "passed": er.passed,
                        "score": er.score,
                        "details": er.details,
                    }
                    for er in failure.evaluator_results
                ],
            },
            "hops": [_hop_to_dict(h) for h in diagnostics],
            "timestamp": datetime.now(UTC).isoformat(),
        }

        output_path.write_text(
            json.dumps(data, indent=2, default=str),
            encoding="utf-8",
        )
        _logger.info("Saved diagnostic trace to %s", output_path)

    # ------------------------------------------------------------------
    # Path safety
    # ------------------------------------------------------------------

    @traces(spec="102", story="US-003", reqs=["REQ-102-048"])
    def _resolve_safe_path(
        self, preferred_dir: Path, filename: str, worktree: Path
    ) -> Path:
        """Return a write path under *worktree*.

        If *preferred_dir* is not inside *worktree*, fall back to
        ``worktree / "diagnostics" / filename``.
        """
        try:
            preferred_dir.resolve().relative_to(worktree.resolve())
            return preferred_dir / filename
        except ValueError:
            _logger.warning(
                "Diagnostics dir %s is outside worktree %s; using worktree fallback",
                preferred_dir,
                worktree,
            )
            return worktree / "diagnostics" / filename

    @traces(spec="102", story="US-003", reqs=["REQ-102-048", "REQ-102-123"])
    def _assert_under_worktree(self, path: Path, worktree: Path) -> None:
        """Raise *PermissionError* if *path* escapes *worktree*."""
        try:
            path.resolve().relative_to(worktree.resolve())
        except ValueError as exc:
            raise PermissionError(
                f"Cannot write diagnostics outside worktree: {path}"
            ) from exc


@traces(spec="102", story="US-003", reqs=["REQ-102-047"])
def _hop_to_dict(hop: HopDiagnostic) -> dict[str, Any]:
    """Serialize a :class:`HopDiagnostic` to a plain dictionary."""
    return {
        "hop_number": hop.hop_number,
        "action": hop.action,
        "packet_sent": hop.packet_sent,
        "response": hop.response,
        "error": hop.error,
    }