146 lines
7.3 KiB
Python
146 lines
7.3 KiB
Python
"""Shared choice resolution for targeted human intervention in curses and plain CLI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import curses
|
|
from dataclasses import dataclass, field
|
|
from collections.abc import Callable
|
|
from posixpath import normpath
|
|
from typing import TextIO
|
|
|
|
from band_aider.ui.dto.human_intervention import CycleInterruptCommandDTO
|
|
|
|
|
|
# Publicly selectable LLM owners; deterministic checks and transport wrappers are excluded.
|
|
DECISION_CHOICES = tuple(sorted((
|
|
"AlphaBlankSourceCorrectionDecision", "AlphaEditAcceptanceDecision", "AlphaSourcePreflightDecision",
|
|
"BetaContractGuidanceDecision", "BootstrapAcceptanceCheck", "BootstrapSourcePreflightDecision",
|
|
"BroadOrNarrowCheck", "CachedPatchGeneralizationCheck", "CodeEditOrLeaveIt", "CrossCuttingConcernCheck",
|
|
"CurrentCotCompletedCheck", "DeadEndDecision", "FileBlameCandidatePrecedenceDecision",
|
|
"FileBlameDecision", "FutureGuidanceApplicability", "FutureGuidanceRelocalization",
|
|
"GlobalRelevantCheck", "LocalizationReview", "PromptLocalization",
|
|
"HumanFeedbackReasonableVibeCheck", "InspectionWorkNeededDecision", "IntentCheck",
|
|
"LegacyFileSalienceCheck", "PredictiveFailureSourceCandidateSelectionDecision", "PromptToStoriesCheck",
|
|
"RepositoryIdiomCheck", "RetryVibeCheck", "SearchReplaceTokenGenerator", "SingleTaskClosureCheck",
|
|
"SourceRemediationGuidanceCheck", "TargetedAlphaOrBetaRoutingDecision",
|
|
"TargetedCollectionAlphaOrBetaRoutingDecision", "TightenDesignWorkCheck",
|
|
)))
|
|
FIELDS = ("decision", "file", "strength")
|
|
|
|
|
|
def matching_choices(prefix: str, choices: tuple[str, ...]) -> tuple[str, ...]:
|
|
"""Filter display choices by prefix without changing their canonical identities."""
|
|
return tuple(filter(lambda choice: choice.casefold().startswith(prefix.casefold()), choices))
|
|
|
|
|
|
def resolve_choice(value: str, choices: tuple[str, ...], default: str = "") -> str:
|
|
"""Accept a blank default, complete identity, or uniquely identifiable prefix."""
|
|
# Blank selectors carry no restriction, while strength defaults to guide.
|
|
if not value:
|
|
return default
|
|
matches = matching_choices(value, choices)
|
|
# A complete identity takes precedence over longer names sharing its prefix.
|
|
if value in choices:
|
|
return value
|
|
# A unique prefix resolves to the same canonical value in both interfaces.
|
|
if len(matches) == 1:
|
|
return matches[0]
|
|
raise ValueError("No matching choice." if not matches else "Ambiguous choice; type more letters or select a row.")
|
|
|
|
|
|
@dataclass
|
|
class InterventionSelection:
|
|
"""Own unsent selector text and dropdown navigation; use selection.command() to resolve it."""
|
|
|
|
files: tuple[str, ...] = ()
|
|
values: dict[str, str] = field(default_factory=lambda: dict(decision="", file="", strength="guide"))
|
|
row: int = 0
|
|
|
|
def saved_cue(self, cues: tuple[CycleInterruptCommandDTO, ...]) -> CycleInterruptCommandDTO | None:
|
|
"""Find acknowledged feedback for the selected decision/file without committing draft selectors."""
|
|
return next(filter(lambda cue: cue.check_name == self.values["decision"]
|
|
and normpath(cue.file) == normpath(self.values["file"]), cues), None)
|
|
|
|
def choices(self, name: str) -> tuple[str, ...]:
|
|
"""Return one selector's complete canonical options."""
|
|
return {"decision": DECISION_CHOICES, "file": self.files, "strength": ("guide", "replace")}[name]
|
|
|
|
def command(self) -> CycleInterruptCommandDTO:
|
|
"""Resolve all fields atomically before creating a one-shot command."""
|
|
decision = resolve_choice(self.values["decision"], DECISION_CHOICES)
|
|
file = resolve_choice(self.values["file"], self.files)
|
|
strength = resolve_choice(self.values["strength"], ("guide", "replace"), "guide")
|
|
self.values.update(decision=decision, file=file, strength=strength)
|
|
return CycleInterruptCommandDTO(decision, file, strength)
|
|
|
|
def handle(self, key: int, name: str) -> str:
|
|
"""Edit a prefix or select a visible dropdown row without submitting an intervention."""
|
|
matches = matching_choices(self.values[name], self.choices(name))
|
|
# Navigation selects a row; typing refines the prefix and resets row selection.
|
|
if key in (curses.KEY_UP, curses.KEY_DOWN):
|
|
self.row = min(max(0, self.row + (1 if key == curses.KEY_DOWN else -1)), max(0, len(matches) - 1))
|
|
# Confirmation commits only an available row, never arbitrary text.
|
|
elif key in (10, 13, curses.KEY_ENTER):
|
|
# The operator must refine a prefix that currently has no candidate.
|
|
if matches:
|
|
self.values[name] = matches[min(self.row, len(matches) - 1)]
|
|
self.row = 0
|
|
else:
|
|
return "No matching choice."
|
|
# Clearing restores the selector's unrestricted or Guide default.
|
|
elif key == 27:
|
|
self.values[name] = "guide" if name == "strength" else ""
|
|
self.row = 0
|
|
# Backspace refines the existing text without committing a target.
|
|
elif key in (curses.KEY_BACKSPACE, 127, 8):
|
|
self.values[name] = self.values[name][:-1]
|
|
self.row = 0
|
|
# Typing replaces a committed choice or extends an unfinished prefix.
|
|
elif 32 <= key <= 126:
|
|
self.values[name] = chr(key) if self.values[name] in self.choices(name) else self.values[name] + chr(key)
|
|
self.row = 0
|
|
return "Enter selects; Escape clears; F5 requests the next match"
|
|
|
|
|
|
@dataclass
|
|
class ChoiceCompleter:
|
|
"""Supply readline's nominal completion callback; bind completer.complete temporarily."""
|
|
|
|
choices: tuple[str, ...]
|
|
|
|
def complete(self, text: str, state: int) -> str | None:
|
|
"""Return a requested match without discarding ambiguous or unmatched input."""
|
|
matches = matching_choices(text, self.choices)
|
|
return matches[state] if state < len(matches) else None
|
|
|
|
|
|
def read_choice(label: str, choices: tuple[str, ...], default: str, input_func: Callable[[str], str], output: TextIO) -> str:
|
|
"""Resolve a CLI selection with temporary shell-style completion and restore readline."""
|
|
import readline
|
|
|
|
previous = readline.get_completer()
|
|
delimiters = readline.get_completer_delims()
|
|
readline.set_completer(ChoiceCompleter(choices).complete)
|
|
readline.set_completer_delims("")
|
|
readline.parse_and_bind("tab: complete")
|
|
try:
|
|
selected: str | None = None
|
|
# Ambiguous input remains operator-owned until an explicit choice resolves it.
|
|
while selected is None:
|
|
value = input_func(f"{label} [{default or 'any'}; type a prefix, Tab lists matches]: ")
|
|
try:
|
|
selected = resolve_choice(value, choices, default)
|
|
except ValueError as error:
|
|
print(str(error), file=output)
|
|
return selected
|
|
finally:
|
|
readline.set_completer(previous)
|
|
readline.set_completer_delims(delimiters)
|
|
|
|
|
|
def read_target(files: tuple[str, ...], input_func: Callable[[str], str], output: TextIO) -> CycleInterruptCommandDTO:
|
|
"""Read the three intervention controls without entering a second terminal application."""
|
|
decision = read_choice("Decision/check", DECISION_CHOICES, "", input_func, output)
|
|
file = read_choice("File", files, "", input_func, output)
|
|
strength = read_choice("Response", ("guide", "replace"), "guide", input_func, output)
|
|
return CycleInterruptCommandDTO(decision, file, strength)
|