scripts-termination-criteria

#!/usr/bin/env python3 """ CODITECT Ralph Loop Termination Criteria (H.8.6.5)

Defines and evaluates termination criteria for autonomous agent loops. Determines when a loop should stop based on budget, iterations, health, time limits, and goal achievement signals.

Usage: from scripts.core.ralph_wiggum.termination_criteria import ( TerminationCriteria, TerminationResult, TerminationReason, )

criteria = TerminationCriteria()
result = criteria.evaluate(
    current_iteration=5,
    max_iterations=10,
    total_cost=12.50,
    max_cost=50.0,
    ...
)
if result.should_terminate:
    print(f"Terminating: {result.reason}")

Author: CODITECT Framework Version: 1.0.0 Created: 2026-02-16 Task Reference: H.8.6.5 ADR References: ADR-108, ADR-110, ADR-111 """

import logging from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any, Dict, List, Optional

logger = logging.getLogger("ralph-loop-orchestrator")

class TerminationReason(Enum): """Reasons for loop termination.""" GOAL_ACHIEVED = "goal_achieved" MAX_ITERATIONS = "max_iterations" BUDGET_EXHAUSTED = "budget_exhausted" MAX_DURATION = "max_duration" HEALTH_FAILURE = "health_failure" CONSECUTIVE_ERRORS = "consecutive_errors" CIRCUIT_BREAKER_OPEN = "circuit_breaker_open" EXPLICIT_STOP = "explicit_stop" NO_PROGRESS = "no_progress"

@dataclass class TerminationResult: """Result of termination evaluation.""" should_terminate: bool = False reason: str = "" details: str = "" iteration: int = 0 budget_used: float = 0.0 budget_remaining: float = 0.0 duration_minutes: float = 0.0

def to_dict(self) -> Dict[str, Any]:
    return asdict(self)

class TerminationCriteria: """ Evaluates termination criteria for autonomous loops.

Criteria (checked in priority order):
1. Health failure (FAILING or TERMINATED state)
2. Consecutive errors exceeding threshold
3. Budget exhausted (>= max_cost)
4. Max iterations reached
5. Max duration exceeded
6. No progress detected (3+ iterations with no new completed items)
"""

def evaluate(
    self,
    current_iteration: int,
    max_iterations: int,
    total_cost: float,
    max_cost: float,
    started_at: str,
    max_duration_minutes: int,
    consecutive_errors: int = 0,
    max_consecutive_errors: int = 3,
    last_health_state: str = "healthy",
    completed_items: Optional[List[str]] = None,
    goal: str = "",
    circuit_breaker_open: bool = False,
) -> TerminationResult:
    """
    Evaluate all termination criteria.

    Returns TerminationResult indicating whether the loop should stop.
    """
    completed_items = completed_items or []

    # Calculate duration
    duration_minutes = 0.0
    if started_at:
        try:
            start = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
            now = datetime.now(timezone.utc)
            duration_minutes = (now - start).total_seconds() / 60
        except (ValueError, TypeError):
            pass

    budget_remaining = max(0, max_cost - total_cost)
    budget_used = total_cost / max_cost if max_cost > 0 else 0

    base = {
        "iteration": current_iteration,
        "budget_used": round(total_cost, 2),
        "budget_remaining": round(budget_remaining, 2),
        "duration_minutes": round(duration_minutes, 1),
    }

    # 1. Circuit breaker open
    if circuit_breaker_open:
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.CIRCUIT_BREAKER_OPEN.value,
            details="Circuit breaker is open — too many failures",
            **base,
        )

    # 2. Health failure
    if last_health_state in ("failing", "terminated"):
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.HEALTH_FAILURE.value,
            details=f"Agent health state: {last_health_state}",
            **base,
        )

    # 3. Consecutive errors
    if consecutive_errors >= max_consecutive_errors:
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.CONSECUTIVE_ERRORS.value,
            details=f"{consecutive_errors} consecutive errors (threshold: {max_consecutive_errors})",
            **base,
        )

    # 4. Budget exhausted
    if total_cost >= max_cost and max_cost > 0:
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.BUDGET_EXHAUSTED.value,
            details=f"${total_cost:.2f} >= ${max_cost:.2f} budget",
            **base,
        )

    # 5. Max iterations
    if current_iteration >= max_iterations:
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.MAX_ITERATIONS.value,
            details=f"Iteration {current_iteration} >= {max_iterations} max",
            **base,
        )

    # 6. Max duration
    if duration_minutes >= max_duration_minutes and max_duration_minutes > 0:
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.MAX_DURATION.value,
            details=f"{duration_minutes:.0f} minutes >= {max_duration_minutes} max",
            **base,
        )

    # 7. No progress detection (only after 3+ iterations)
    if current_iteration >= 3 and not completed_items:
        return TerminationResult(
            should_terminate=True,
            reason=TerminationReason.NO_PROGRESS.value,
            details=f"No completed items after {current_iteration} iterations",
            **base,
        )

    # Continue — no termination criteria met
    return TerminationResult(
        should_terminate=False,
        reason="",
        details="All criteria within bounds",
        **base,
    )