band-aider-pub/band_aider/ai_client/escalation_utils.py

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only

"""Stateless model and human operations; callers own escalation policy and progress."""

import hashlib
import logging
import time
from dataclasses import replace
from collections.abc import Callable
from uuid import uuid4

from band_aider.ai_client.completion_client import CompletionClient, HumanResponse, OpenAIConfig
from band_aider.ai_client.cycle_record import CycleRecord
from band_aider.ai_client.token_budget_estimator import TokenBudgetEstimator
from band_aider.judgement import prompt_formatter
from band_aider.judgement.prompt_formatter import PromptFormatterBuilder
from band_aider.ui.dto.human_intervention import CycleInterruptCommandDTO, HumanInterventionRequestDTO, HumanInterventionResponseDTO
from band_aider.ui.dto.model_escalation import ModelEscalationDTO
from band_aider.ui.status_publisher import HumanCommandReceiver, StatusPublisher


class EscalationDeferred(RuntimeError):
    """Return control to the model policy, e.g. ``raise EscalationDeferred('skip')``."""

    def __init__(self, message: str, *, suppress: bool = False) -> None:
        """Carry the operator's suppression choice to the owning escalation policy."""

        super().__init__(message)
        self.suppress = suppress


class EscalationAborted(RuntimeError):
    """Report an explicit human abort without admitting a judgment."""


def model_configs(clients: tuple[CompletionClient, ...]) -> tuple[OpenAIConfig, ...]:
    """Describe the supplied provider tiers without duplicating an advertised config chain."""

    primary = clients[0].openai_configs
    return primary if len(primary) >= len(clients) else tuple(map(
        lambda client: client.openai_configs[0], filter(lambda client: bool(client.openai_configs), clients),
    ))


def publish(
    prompt: str, index: int, state: str, *, clients: tuple[CompletionClient, ...],
    check_name: str, event_publisher: StatusPublisher, record: CycleRecord | None = None,
) -> None:
    """Publish supplied tier and repetition evidence without reading or advancing counters."""

    configs = model_configs(clients)
    config = configs[index] if index < len(configs) else OpenAIConfig(model="unknown", api_base="unknown")
    event_publisher.publish(ModelEscalationDTO(
        trigger_name=check_name, occurred_at=time.time(), prompt_preview=prompt,
        reason=f"Judgment attempt uses configured tier {index}.",
        request_md5=hashlib.md5(prompt.encode()).hexdigest(),
        local_model=configs[0].model if configs else "unknown",
        active_model=config.model, config_index=index, config_count=len(clients),
        state=state, active_endpoint=config.api_base,
        repeat_count=record.repeat_count if record else 0,
        response_md5=record.response_md5 if record else "",
        response_tail=record.response_tail if record else "",
        tokens_escalated=max(0, int(TokenBudgetEstimator().estimate(prompt))) if index else 0,
    ))


def complete_tier(
    prompt: str, index: int, *, clients: tuple[CompletionClient, ...], check_name: str,
    event_publisher: StatusPublisher, logger: logging.Logger | None = None,
) -> str:
    """Complete the caller-selected tier and publish telemetry; never select the next tier."""

    client = clients[index]
    publish(
        prompt, index, "escalating" if index else "baseline", clients=clients,
        check_name=check_name, event_publisher=event_publisher,
    )
    return client.complete(prompt) if logger is None else client.complete(prompt, logger=logger)


def guided_prompt(prompt: str, guidance: str, builder: PromptFormatterBuilder) -> str:
    """Preserve human guidance verbatim at the existing highest-authority boundary."""

    return prompt_formatter.format_prompt(
        builder, "HUMAN GUIDANCE",
        lambda: "\n\n".join((
            prompt_formatter.format_prompt(builder, "ORIGINAL REQUEST", prompt, depth=2, escape=False),
            prompt_formatter.format_prompt(
                builder, "AUTHORITATIVE GUIDANCE",
                "Human guidance (highest priority; authoritative instruction that overrides all prior guidance when conflicts exist):\n" + guidance,
                depth=2,
            ),
        )), escape=False,
    )


def human_response(
    prompt: str, *, clients: tuple[CompletionClient, ...], check_name: str,
    prompt_formatter_builder: PromptFormatterBuilder, event_publisher: StatusPublisher,
    command_receiver: HumanCommandReceiver, direct: bool, previous_response: str,
    ask: str = "", validation_errors: str = "", draft: str = "",
    suppressed: bool = False,
    manual_target: CycleInterruptCommandDTO | None = None,
    files: tuple[str, ...] = (),
    release_manual: Callable[[], None] | None = None,
    pop_cue: Callable[[], CycleInterruptCommandDTO | None] | None = None,
    logger: logging.Logger | None = None,
) -> str:
    """Perform a caller-selected hint or dictation exchange, retaining no state between calls."""

    # Suppressed intervention returns control before publishing a screen or waiting for input.
    if suppressed:
        raise EscalationDeferred(f"Intervention suppressed for {check_name}.", suppress=True)
    cue: CycleInterruptCommandDTO | None = None
    # Apply queued clears before consuming text; the worker remains its sole owner.
    if pop_cue is not None:
        command_receiver.consume_pending()
        cue = pop_cue()
    # Stored feedback chooses the same guide or replacement submission path as live input.
    if cue is not None:
        direct = cue.strength == "replace"
    configs = model_configs(clients)
    response: str | None = None
    # Skipping the optional hint opens dictation during this exchange without calling a model.
    while response is None:
        request = HumanInterventionRequestDTO(
            request_id=str(uuid4()), check_name=check_name,
            original_prompt=ask or prompt,
            repeated_response_tail="" if direct else previous_response,
            request_md5=hashlib.md5(prompt.encode()).hexdigest(),
            response_md5=hashlib.md5(previous_response.encode()).hexdigest(),
            local_model=configs[0].model if configs else "unknown",
            escalated_model=configs[-1].model if configs else "unknown",
            mode="direct_response" if direct else "hint",
            validation_errors=validation_errors, draft=draft if direct else "",
            guide_only=manual_target is not None and manual_target.strength == "guide",
            files=files,
        )
        # A matching cue replaces the input exchange without publishing a stop screen.
        if cue is not None:
            command = HumanInterventionResponseDTO(request.request_id, cue.guidance, "resend")
            cue = None
        # Missing cues preserve the existing correlated human exchange.
        else:
            event_publisher.publish(request)
            command = command_receiver.wait_for(request.request_id)
        # Only an explicit submission can supply a response to the owning validator.
        if command.action == "abort":
            # The caller owns releasing its selected request, including explicit cancellation.
            if release_manual is not None:
                release_manual()
            raise EscalationAborted(f"Human aborted {check_name}.")
        # Deferral resumes model work; suppression additionally disables later automatic prompts.
        elif command.action in {"defer", "suppress"}:
            # Discard the request without promoting an unused draft into an instruction.
            if release_manual is not None:
                release_manual()
            raise EscalationDeferred(f"Human {command.action} for {check_name}.", suppress=command.action == "suppress")
        elif command.action == "resend" and direct:
            response = HumanResponse(command.guidance)
            command = replace(command, guidance="")
            # Human text crosses to validation without a provider invocation.
            if release_manual is not None:
                release_manual()
        elif command.action == "resend":
            guided = guided_prompt(prompt, command.guidance, prompt_formatter_builder)
            command = replace(command, guidance="")
            # Consume before dispatch so even a failed provider call cannot replay advice.
            if release_manual is not None:
                release_manual()
            response = complete_tier(
                guided, len(clients) - 1, clients=clients, check_name=check_name,
                event_publisher=event_publisher, logger=logger,
            )
        elif command.action == "retry" and not direct:
            # A manually selected guide-only exchange never escalates to dictation.
            if manual_target is not None:
                # Release only the matching request; a newer selection belongs to its caller.
                if release_manual is not None:
                    release_manual()
                raise EscalationDeferred(f"Human skipped guidance for {check_name}.")
            direct = True
        else:
            # An unsupported action cannot leave an abandoned manual exchange active.
            if release_manual is not None:
                release_manual()
            raise EscalationDeferred(f"Unsupported human action {command.action!r} for {request.mode}.")
    return response