70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Callable
|
|
|
|
from band_aider.context.contex_bag import ContextBag
|
|
from band_aider.judgement.check import Check
|
|
from band_aider.judgement.mechanical_check import MechanicalCheck
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
|
|
|
|
@dataclass(init=False)
|
|
class PredicateCheck(MechanicalCheck):
|
|
"""PredicateCheck records the boolean result of one structural predicate."""
|
|
|
|
predicate: Callable[[], bool]
|
|
changes: list[tuple[Check, bool]] = field(default_factory=list)
|
|
|
|
def __init__(self, predicate: Callable[[], bool]) -> None:
|
|
"""Create a predicate-backed mechanical check with an empty change log."""
|
|
|
|
self.predicate = predicate
|
|
self.changes = []
|
|
|
|
def __call__(self, *args, **kwargs) -> bool:
|
|
"""Return and record the predicate result."""
|
|
|
|
result = bool(self.predicate())
|
|
self.changes.append((self, result))
|
|
return result
|
|
|
|
|
|
@dataclass(init=False)
|
|
class MechanicalLoopGate(LoopGate):
|
|
"""MechanicalLoopGate evaluates structural entry, retry, and continuation predicates."""
|
|
|
|
def __init__(
|
|
self,
|
|
context: ContextBag,
|
|
may_enter: Callable[[], bool],
|
|
should_continue: Callable[[], bool],
|
|
should_retry: Callable[[], bool] | None = None,
|
|
) -> None:
|
|
"""Create a gate from deterministic structural predicates."""
|
|
|
|
super().__init__(
|
|
context=context,
|
|
may_enter_check=PredicateCheck(may_enter),
|
|
should_retry_check=PredicateCheck(should_retry or (lambda: False)),
|
|
should_continue_check=PredicateCheck(should_continue),
|
|
)
|
|
|
|
def may_enter(self) -> bool:
|
|
"""Return whether the loop may enter based on structural checks."""
|
|
|
|
return self.may_enter_check()
|
|
|
|
def should_continue(self) -> bool:
|
|
"""Return whether the loop should request another child-work pass."""
|
|
|
|
return self.should_continue_check()
|
|
|
|
def should_retry(self) -> bool:
|
|
"""Return whether the loop should retry the current scope."""
|
|
|
|
return self.should_retry_check()
|