241 lines
7.8 KiB
Python
241 lines
7.8 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Iterable
|
|
|
|
from band_aider.judgement.lexical_anchor_check import LexicalAnchorCheck
|
|
from band_aider.judgement.model2vec_code_parser import (
|
|
LexicalCheckFactory,
|
|
MODEL2VEC_SIMILARITY_THRESHOLD,
|
|
Model2VecSimilarityCheck,
|
|
SemanticCheckFactory,
|
|
)
|
|
from band_aider.judgement.relevant_beat_selector import RelevantBeatSelector
|
|
|
|
|
|
class _LexicalCheck(LexicalAnchorCheck):
|
|
"""Return the next configured lexical match from one selector scenario."""
|
|
|
|
def __init__(
|
|
self,
|
|
factory: "_LexicalFactory",
|
|
beats: Iterable[str],
|
|
) -> None:
|
|
"""Bind one check to its factory's stationary matches and current candidates."""
|
|
|
|
super().__init__(changes=[], anchors=(), candidate_beats=beats)
|
|
self.factory = factory
|
|
|
|
def __call__(
|
|
self,
|
|
anchors: Iterable[str] | None = None,
|
|
candidate_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Select the next configured candidate that remains in this check's scope."""
|
|
|
|
del anchors
|
|
beats = tuple(self.candidate_beats if candidate_beats is None else candidate_beats)
|
|
if self.factory.fail:
|
|
return False
|
|
for candidate in self.factory.matches:
|
|
if candidate in beats and candidate not in self.factory.selected:
|
|
self.factory.selected.add(candidate)
|
|
self.best_candidate = candidate
|
|
return True
|
|
return False
|
|
|
|
|
|
class _LexicalFactory:
|
|
"""Create nominal lexical checks sharing one stationary selection history."""
|
|
|
|
def __init__(self, matches: list[str], *, fail: bool = False) -> None:
|
|
"""Store deterministic lexical match order and an optional failure condition."""
|
|
|
|
self.matches = list(matches)
|
|
self.selected: set[str] = set()
|
|
self.fail = fail
|
|
|
|
def __call__(
|
|
self,
|
|
anchors: Iterable[str],
|
|
beats: Iterable[str],
|
|
) -> LexicalAnchorCheck:
|
|
"""Return one nominal check for the caller-supplied candidate scope."""
|
|
|
|
del anchors
|
|
return _LexicalCheck(self, beats)
|
|
|
|
|
|
class _SemanticCheck(Model2VecSimilarityCheck):
|
|
"""Return configured semantic matches without loading a local model artifact."""
|
|
|
|
def __init__(self, matches: list[tuple[str, float]], *, fail: bool = False) -> None:
|
|
"""Store deterministic scored candidates and a stationary failure case."""
|
|
|
|
super().__init__(changes=[], request_text="", searchable_beats=())
|
|
self.matches = list(matches)
|
|
self.selected: set[str] = set()
|
|
self.fail = fail
|
|
self.calls = 0
|
|
|
|
def __call__(
|
|
self,
|
|
request_text: str | None = None,
|
|
searchable_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Select the next configured semantic candidate from supplied beats."""
|
|
|
|
del request_text
|
|
self.calls += 1
|
|
beats = tuple(searchable_beats or ())
|
|
if self.fail:
|
|
return False
|
|
for candidate, score in self.matches:
|
|
if candidate in beats and candidate not in self.selected:
|
|
self.selected.add(candidate)
|
|
self.best_candidate = candidate
|
|
self.score = score
|
|
return True
|
|
return False
|
|
|
|
|
|
class _SemanticFactory:
|
|
"""Return one reusable nominal semantic check for a selector scenario."""
|
|
|
|
def __init__(self, check: _SemanticCheck) -> None:
|
|
"""Store the semantic check and a visible factory invocation count."""
|
|
|
|
self.check = check
|
|
self.calls = 0
|
|
|
|
def __call__(
|
|
self,
|
|
request_text: str,
|
|
beats: Iterable[str],
|
|
) -> Model2VecSimilarityCheck:
|
|
"""Return the scenario's semantic check without making an external call."""
|
|
|
|
del request_text, beats
|
|
self.calls += 1
|
|
return self.check
|
|
|
|
|
|
def _selector(
|
|
*,
|
|
lexical_matches: list[str] | None = None,
|
|
semantic_matches: list[tuple[str, float]] | None = None,
|
|
semantic_fail: bool = False,
|
|
) -> RelevantBeatSelector:
|
|
"""Return a selector with nominal checks over stationary scenario evidence."""
|
|
|
|
lexical_factory: LexicalCheckFactory = _LexicalFactory(lexical_matches or [])
|
|
semantic_factory: SemanticCheckFactory = _SemanticFactory(
|
|
_SemanticCheck(semantic_matches or [], fail=semantic_fail)
|
|
)
|
|
return RelevantBeatSelector(
|
|
anchors=("anchor",),
|
|
request_text="request",
|
|
lexical_check_factory=lexical_factory,
|
|
semantic_check_factory=semantic_factory,
|
|
)
|
|
|
|
|
|
def test_relevant_beat_selector_keeps_all_iterative_lexical_matches_in_extraction_order() -> None:
|
|
"""Verify lexical selection returns every configured match in match order."""
|
|
|
|
beats = ("third", "first", "second")
|
|
selector = _selector(lexical_matches=["first", "second"])
|
|
|
|
assert selector(beats) == ("first", "second")
|
|
|
|
|
|
def test_relevant_beat_selector_keeps_semantic_matches_until_existing_check_stops() -> None:
|
|
"""Verify semantic candidates remain selected until the nominal check rejects."""
|
|
|
|
selector = _selector(
|
|
semantic_matches=[
|
|
("low", 0.73),
|
|
("high", 0.91),
|
|
]
|
|
)
|
|
|
|
assert selector(("low", "high", "other")) == ("high", "low")
|
|
|
|
|
|
def test_relevant_beat_selector_dedupes_selected_beats_once() -> None:
|
|
"""Verify lexical and semantic agreement retains one complete source beat."""
|
|
|
|
selector = _selector(
|
|
lexical_matches=["shared"],
|
|
semantic_matches=[("shared", 0.95), ("semantic", 0.90)],
|
|
)
|
|
|
|
assert selector(("shared", "semantic", "shared")) == ("shared", "semantic")
|
|
|
|
|
|
def test_relevant_beat_selector_semantic_ties_preserve_extraction_order() -> None:
|
|
"""Verify equal semantic scores retain source-extraction order."""
|
|
|
|
selector = _selector(
|
|
semantic_matches=[
|
|
("second", 0.8),
|
|
("first", 0.8),
|
|
]
|
|
)
|
|
|
|
assert selector(("first", "second")) == ("first", "second")
|
|
|
|
|
|
def test_relevant_beat_selector_preserves_complete_long_and_many_beats() -> None:
|
|
"""Verify selection preserves complete long text and every configured beat."""
|
|
|
|
long_beat = "x" * 320
|
|
many = [f"beat {index}" for index in range(45)]
|
|
selector = _selector(lexical_matches=[long_beat, *many])
|
|
|
|
selected = selector((long_beat, *many))
|
|
|
|
assert selected[0] == long_beat
|
|
assert len(selected) == 46
|
|
|
|
|
|
def test_relevant_beat_selector_semantic_failure_preserves_lexical_matches() -> None:
|
|
"""Verify semantic rejection does not revoke earlier lexical admission."""
|
|
|
|
selector = _selector(lexical_matches=["lexical"], semantic_fail=True)
|
|
|
|
assert selector(("lexical", "semantic")) == ("lexical",)
|
|
|
|
|
|
def test_relevant_beat_selector_no_match_returns_empty_tuple() -> None:
|
|
"""Verify no matching complete beat produces no selected guidance."""
|
|
|
|
assert _selector()(("unmatched",)) == ()
|
|
|
|
|
|
def test_relevant_beat_selector_uses_existing_threshold_and_no_local_math() -> None:
|
|
"""Verify the selector exposes the shared semantic threshold by default."""
|
|
|
|
assert RelevantBeatSelector("a", "r").semantic_threshold == MODEL2VEC_SIMILARITY_THRESHOLD
|
|
assert MODEL2VEC_SIMILARITY_THRESHOLD == 0.72
|
|
|
|
|
|
def test_relevant_beat_selector_applies_constructor_owned_threshold() -> None:
|
|
"""Verify selector construction supplies its threshold to semantic checks."""
|
|
|
|
semantic_check = _SemanticCheck([("semantic", 0.9)])
|
|
lexical_factory: LexicalCheckFactory = _LexicalFactory([])
|
|
semantic_factory: SemanticCheckFactory = _SemanticFactory(semantic_check)
|
|
selector = RelevantBeatSelector(
|
|
anchors=("anchor",),
|
|
request_text="request",
|
|
lexical_check_factory=lexical_factory,
|
|
semantic_check_factory=semantic_factory,
|
|
semantic_threshold=0.81,
|
|
)
|
|
|
|
assert selector(("semantic",)) == ("semantic",)
|
|
assert semantic_check.threshold == 0.81
|