260 lines
13 KiB
Python
260 lines
13 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import hashlib
|
|
import logging
|
|
from functools import partial
|
|
from itertools import chain
|
|
from posixpath import normpath
|
|
|
|
from band_aider.ai_client import escalation_utils
|
|
from band_aider.ai_client import completion_client
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.ai_client.cycle_record import CycleRecord
|
|
from band_aider.ai_client.escalation_utils import (
|
|
EscalationAborted as CycleJackerAborted,
|
|
EscalationDeferred as CycleJackerDeferred,
|
|
)
|
|
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
|
|
from band_aider.judgement.prompt_formatter import PromptFormatterBuilder
|
|
from band_aider.ui.dto.human_feedback import HumanFeedbackKind
|
|
from band_aider.ui.dto.human_intervention import CycleInterruptCommandDTO, InterventionStatusDTO
|
|
from band_aider.ui.status_publisher import (
|
|
DeferredHumanCommandReceiver,
|
|
HumanCommandReceiver,
|
|
NullStatusPublisher,
|
|
StatusPublisher,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CycleState:
|
|
"""Own run-local manual selection, saved cues, and suppression; use cycle.arm(target, publisher)."""
|
|
|
|
human_interrupt: bool = False
|
|
intervention_suppressed: bool = False
|
|
target: CycleInterruptCommandDTO = field(default_factory=CycleInterruptCommandDTO)
|
|
cues: dict[str, dict[str, CycleInterruptCommandDTO]] = field(default_factory=dict)
|
|
|
|
@property
|
|
def status(self) -> InterventionStatusDTO:
|
|
"""Project the worker's complete selection and unconsumed cues as immutable UI facts."""
|
|
return InterventionStatusDTO(self.target, self.human_interrupt, tuple(chain.from_iterable(
|
|
map(lambda entries: entries.values(), self.cues.values()),
|
|
)))
|
|
|
|
def arm(self, target: CycleInterruptCommandDTO, publisher: StatusPublisher) -> None:
|
|
"""Replace the unconsumed selection and acknowledge it to the operator."""
|
|
self.target = target
|
|
# One decision/file key owns one cue, regardless of its guide or replace mode.
|
|
if target.guidance:
|
|
self.cues.setdefault(target.check_name, {})[normpath(target.file)] = target
|
|
self.human_interrupt = True
|
|
self.intervention_suppressed = False
|
|
publisher.publish(self.status)
|
|
|
|
def pop_cue(self, check_name: str, files: tuple[str, ...], publisher: StatusPublisher) -> CycleInterruptCommandDTO | None:
|
|
"""Consume one exact decision/file cue only at its human feedback boundary."""
|
|
# A suppression command received at this boundary also protects saved human input.
|
|
if self.intervention_suppressed:
|
|
return None
|
|
entries = self.cues.get(check_name, {})
|
|
file = next(filter(lambda path: path in entries, map(normpath, files)), None)
|
|
# Near matches leave every cue available for its actual owner.
|
|
if file is None:
|
|
return None
|
|
cue = entries.pop(file)
|
|
publisher.publish(self.status)
|
|
return cue
|
|
|
|
def clear_cues(self, target: CycleInterruptCommandDTO | None, publisher: StatusPublisher) -> None:
|
|
"""Discard stored feedback without cancelling the existing manual stop selection."""
|
|
# An explicit all-cues action owns the entire run-local repertoire.
|
|
if target is None:
|
|
self.cues.clear()
|
|
# A keyed clear cannot remove another decision's or file's feedback.
|
|
else:
|
|
self.cues.get(target.check_name, {}).pop(normpath(target.file), None)
|
|
publisher.publish(self.status)
|
|
|
|
def matching(self, check_name: str, files: tuple[str, ...]) -> CycleInterruptCommandDTO | None:
|
|
"""Select only an armed exact decision/file match without consuming it."""
|
|
target = self.target
|
|
matches = (not target.check_name or target.check_name == check_name) and (not target.file or normpath(target.file) in map(normpath, files))
|
|
return target if self.human_interrupt and matches else None
|
|
|
|
def consume(self, target: CycleInterruptCommandDTO, publisher: StatusPublisher) -> None:
|
|
"""Release the submitted selection without clearing a newer operator request."""
|
|
# A response for an older exchange cannot consume a rearmed selection.
|
|
if self.target is target:
|
|
self.human_interrupt = False
|
|
self.target = CycleInterruptCommandDTO()
|
|
publisher.publish(InterventionStatusDTO(target, False, self.status.cues))
|
|
|
|
|
|
cycle = CycleState()
|
|
|
|
|
|
@dataclass(init=False)
|
|
class CycleJacker(CompletionClient):
|
|
"""Escalate repeated pairs in equal response blocks, probing the local client on each call."""
|
|
|
|
event_publisher: StatusPublisher
|
|
command_receiver: HumanCommandReceiver
|
|
repeat_threshold: int
|
|
response_tail_length: int
|
|
feedback_store: HumanFeedbackStore
|
|
prompt_formatter_builder: PromptFormatterBuilder
|
|
_inner: CompletionClient = field(init=False, repr=False)
|
|
|
|
def __init__(
|
|
self,
|
|
inner: CompletionClient,
|
|
*,
|
|
feedback_store: HumanFeedbackStore,
|
|
event_publisher: StatusPublisher | None = None,
|
|
command_receiver: HumanCommandReceiver | None = None,
|
|
escalation_clients: tuple[CompletionClient, ...] = (),
|
|
repeat_threshold: int = 12,
|
|
# TODO: Deliberate with a human whether to remove this arbitrary response-history cap.
|
|
response_tail_length: int = 1000,
|
|
prompt_formatter_builder: PromptFormatterBuilder,
|
|
) -> None:
|
|
"""Bind cycle persistence and a positive response allowance shared by every tier."""
|
|
|
|
# Every model and human tier must have at least one response slot.
|
|
if repeat_threshold < 1:
|
|
raise ValueError("repeat_threshold must be positive.")
|
|
self._inner = inner
|
|
self.event_publisher = event_publisher or NullStatusPublisher()
|
|
self.command_receiver = command_receiver or DeferredHumanCommandReceiver()
|
|
self.repeat_threshold = repeat_threshold
|
|
self.response_tail_length = response_tail_length
|
|
self.feedback_store = feedback_store
|
|
self.prompt_formatter_builder = prompt_formatter_builder
|
|
self.escalation_clients = escalation_clients
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient:
|
|
return self._inner
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
return self.inner.openai_configs
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Detect an exact pair and delegate cloud or human recovery when needed."""
|
|
|
|
self.command_receiver.consume_pending()
|
|
scope = completion_client.completion_scope.get()
|
|
check_name = scope.check_name or type(self).__name__
|
|
selected = cycle.matching(check_name, scope.files)
|
|
# Direct replacement supplies the answer without first buying a model response.
|
|
if selected is not None and selected.strength == "replace":
|
|
try:
|
|
return escalation_utils.human_response(
|
|
prompt, clients=(self.inner, *self.escalation_clients), check_name=check_name,
|
|
prompt_formatter_builder=self.prompt_formatter_builder,
|
|
event_publisher=self.event_publisher, command_receiver=self.command_receiver,
|
|
direct=True, previous_response="", manual_target=selected, files=scope.files, logger=logger,
|
|
release_manual=partial(cycle.consume, selected, self.event_publisher),
|
|
pop_cue=partial(cycle.pop_cue, check_name, scope.files, self.event_publisher),
|
|
)
|
|
except CycleJackerDeferred as error:
|
|
cycle.intervention_suppressed = cycle.intervention_suppressed or error.suppress
|
|
response = self.inner.complete(prompt) if logger is None else self.inner.complete(prompt, logger=logger)
|
|
self.command_receiver.consume_pending()
|
|
local_model = self.openai_configs[0].model if self.openai_configs else "unknown"
|
|
record = self._record(prompt, response, local_model)
|
|
clients = (self.inner, *self.escalation_clients)
|
|
# Each tier keeps the same allowance; the optional human tier also wraps back to local.
|
|
client_index = ((record.repeat_count - 1) // self.repeat_threshold) % (len(clients) + 1)
|
|
selected = cycle.matching(check_name, scope.files)
|
|
interrupted = selected is not None
|
|
# Explicit F5 requests override automatic-intervention suppression.
|
|
if interrupted:
|
|
cycle.intervention_suppressed = False
|
|
# New pairs and the first block of each repeated circuit stay on the local client.
|
|
if not interrupted and client_index == 0:
|
|
return response
|
|
request_id = f"{record.request_md5}:{record.response_md5}"
|
|
feedback = self.feedback_store.latest_feedback(request_id)
|
|
configs = escalation_utils.model_configs(clients)
|
|
highest = len(self.escalation_clients)
|
|
# Previously approved feedback remains valid for the noncritical cycle decorator.
|
|
if not interrupted and feedback is not None and feedback.kind is HumanFeedbackKind.Ignore:
|
|
return response
|
|
human_requested = interrupted or (feedback is not None and feedback.kind is HumanFeedbackKind.Ask)
|
|
# Explicit intervention bypasses ordinary exact-pair recovery without changing its counter policy.
|
|
if not human_requested:
|
|
# Retain the detector's diagnostic anchor while sharing recovery mechanics.
|
|
if logger is not None:
|
|
logger.warning(
|
|
"CycleJacker detected repeated completion request_md5=%s response_md5=%s repeat_count=%s client_index=%s attempt=%s/%s.",
|
|
record.request_md5, record.response_md5, record.repeat_count, client_index,
|
|
(record.repeat_count - 1) % self.repeat_threshold + 1, self.repeat_threshold,
|
|
)
|
|
escalation_utils.publish(
|
|
prompt, 0, "detected", clients=clients, check_name=check_name,
|
|
event_publisher=self.event_publisher, record=record,
|
|
)
|
|
# Exhaust the selected cloud's allowance before advancing to another tier or human help.
|
|
if client_index <= highest:
|
|
escalated = escalation_utils.complete_tier(
|
|
prompt, client_index, clients=clients, check_name=check_name,
|
|
event_publisher=self.event_publisher, logger=logger,
|
|
)
|
|
# Changed bytes reset only this exact-pair detector, never stabilization progression.
|
|
if hashlib.md5(escalated.encode()).hexdigest() != record.response_md5:
|
|
resolved = self._record(prompt, escalated, configs[client_index].model if client_index < len(configs) else "unknown")
|
|
escalation_utils.publish(
|
|
prompt, client_index, "resolved", clients=clients, check_name=check_name,
|
|
event_publisher=self.event_publisher, record=resolved,
|
|
)
|
|
return escalated
|
|
# Suppression also applies to ordinary renderer intervention, not just judgment calls.
|
|
if not cycle.intervention_suppressed:
|
|
escalation_utils.publish(
|
|
prompt, highest, "human_required", clients=clients, check_name=check_name,
|
|
event_publisher=self.event_publisher, record=record,
|
|
)
|
|
try:
|
|
return escalation_utils.human_response(
|
|
prompt, clients=clients, check_name=check_name,
|
|
prompt_formatter_builder=self.prompt_formatter_builder,
|
|
event_publisher=self.event_publisher, command_receiver=self.command_receiver,
|
|
direct=(selected is not None and selected.strength == "replace") or (not human_requested and (record.repeat_count - 1) % self.repeat_threshold > 0),
|
|
previous_response=response,
|
|
manual_target=selected, files=scope.files,
|
|
release_manual=partial(cycle.consume, selected, self.event_publisher) if selected is not None else None,
|
|
pop_cue=partial(cycle.pop_cue, check_name, scope.files, self.event_publisher),
|
|
suppressed=cycle.intervention_suppressed, logger=logger,
|
|
)
|
|
except CycleJackerDeferred as error:
|
|
cycle.intervention_suppressed = cycle.intervention_suppressed or error.suppress
|
|
self.feedback_store.reset_cycle(record.request_md5)
|
|
# The local response already obtained is the first response of the resumed circuit.
|
|
self._record(prompt, response, local_model)
|
|
# The consumer still validates that response; deferral supplies no verdict.
|
|
if logger is not None:
|
|
logger.info("%s Continuing ordinary completion retries.", error)
|
|
return response
|
|
|
|
def _record(
|
|
self,
|
|
prompt: str,
|
|
response: str,
|
|
model: str,
|
|
) -> CycleRecord:
|
|
"""Persist the exact pair with the existing cycle detector's response preview."""
|
|
|
|
return self.feedback_store.record_cycle(
|
|
request_md5=hashlib.md5(prompt.encode()).hexdigest(),
|
|
response_md5=hashlib.md5(response.encode()).hexdigest(),
|
|
response_tail=response[-self.response_tail_length:],
|
|
model=model,
|
|
)
|