atlas-iac/services/hermes/scripts/cli_lane_failover.py

#!/usr/bin/env python3
"""Bounded, fail-closed cross-provider failover for one capacity boundary."""

from __future__ import annotations

import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from cli_lane_board import _board_call
from cli_lane_config import ProcessResult, Route
from cli_lane_health import classify_capacity_failure, record_provider_failure
from cli_lane_metrics import (
    record_provider_fallback,
    record_route_decision,
    record_router_selection_failure,
)
from cli_lane_prompt import build_prompt, git_handoff
from cli_lane_provider import run_provider
from cli_lane_records import _persist_candidate
from cli_lane_routing import select_route


@dataclass
class FailoverOutcome:
    """Where one capacity boundary left the claim's route and result."""

    route: Route
    result: ProcessResult
    candidate_file: Path | None
    unavailable_provider: str | None
    blocked: bool


def _block_transient(
    kanban_db: Any, board: str, task_id: str, run_id: Any, reason: str
) -> None:
    """Park one card as retryable so it re-enters when providers recover."""
    _board_call(
        kanban_db,
        board,
        lambda fresh: kanban_db.block_task(
            fresh,
            task_id,
            reason=reason,
            kind="transient",
            expected_run_id=run_id,
        ),
    )


def _routed_or_blocked(
    kanban_db: Any,
    board: str,
    task_id: str,
    run_id: Any,
    select: Callable[[], Route],
) -> Route | None:
    """Turn a Switchyard outage into a transient block, never a capability one."""
    try:
        return select()
    except RuntimeError as error:
        record_router_selection_failure()
        _block_transient(
            kanban_db,
            board,
            task_id,
            run_id,
            "Switchyard route selection is unavailable; the card retries "
            f"when the router recovers: {error}",
        )
        return None


def capacity_failover(
    kanban_db: Any,
    *,
    board: str,
    task_id: str,
    run_id: Any,
    assignee: str,
    comment: Callable[[str], None],
    context: str,
    workspace: Path,
    state: dict[str, Any],
    state_file: Path,
    log_path: Path,
    heartbeat: Callable[[str], bool],
    deadline: float,
    route: Route,
    result: ProcessResult,
    candidate_file: Path | None,
    goal_turn: int,
) -> FailoverOutcome:
    """Apply the capacity policy once: record, fail closed, or fail over."""
    failure_class = classify_capacity_failure(result.output)
    record_provider_failure(route.provider, failure_class)
    record_route_decision(
        route.provider, route.effort, route.classifier, "capacity-failure"
    )
    if assignee != "cli-auto":
        # Fail closed: an operator pinned this provider, so a capacity or
        # auth failure must wait for that provider, never switch away.
        _block_transient(
            kanban_db,
            board,
            task_id,
            run_id,
            f"Manually pinned provider {route.provider} hit a "
            f"{failure_class} failure; the explicit {assignee} lane never "
            "switches providers. Retry after recovery or reassign to cli-auto.",
        )
        return FailoverOutcome(route, result, candidate_file, None, True)
    retry_context = (
        context
        + "\n\nRouting boundary: the first provider failed from capacity/authentication. "
        + "Select the alternate hosted provider at an appropriate effort."
    )
    alternate = "claude" if route.provider == "codex" else "codex"
    fallback = _routed_or_blocked(
        kanban_db,
        board,
        task_id,
        run_id,
        lambda: select_route(retry_context, f"cli-{alternate}-{route.effort}"),
    )
    if fallback is None:
        return FailoverOutcome(route, result, candidate_file, route.provider, True)
    record_provider_fallback(route.provider, fallback.provider, failure_class)
    comment(
        f"Provider fallback: {route.provider} -> {fallback.provider} "
        f"after a {failure_class} failure; Jetson reclassified the retry boundary.",
    )
    fallback_result = run_provider(
        fallback,
        build_prompt(
            context,
            workspace,
            git_handoff(workspace, result.output),
        ),
        workspace,
        state,
        state_file,
        log_path,
        heartbeat,
        max(1, int(deadline - time.monotonic())),
    )
    if fallback_result.capacity_failure:
        # Bounded failover depth: two hosted providers, never a third
        # metered or local path.
        fallback_class = classify_capacity_failure(fallback_result.output)
        record_provider_failure(fallback.provider, fallback_class)
        record_route_decision(
            fallback.provider, fallback.effort, fallback.classifier, "capacity-failure"
        )
        _block_transient(
            kanban_db,
            board,
            task_id,
            run_id,
            "Both hosted providers failed at this boundary "
            f"({route.provider}: {failure_class}; {fallback.provider}: "
            f"{fallback_class}); the lane never falls to a metered or "
            "local path. Waiting for provider recovery.",
        )
        return FailoverOutcome(
            fallback, fallback_result, candidate_file, route.provider, True
        )
    if fallback_result.structured:
        candidate_file = _persist_candidate(
            state,
            state_file,
            dict(fallback_result.structured),
            route=fallback,
            returncode=fallback_result.returncode,
            goal_turn=goal_turn,
        )
    return FailoverOutcome(
        fallback, fallback_result, candidate_file, route.provider, False
    )