186 lines
11 KiB
Python
186 lines
11 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
|
|
from band_aider.context import validation_utils
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.validation_utils import (
|
|
ITEMS,
|
|
MechanicalValidationError,
|
|
ValidationError,
|
|
ValidationLocation,
|
|
ValidationResult,
|
|
)
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.reasoning_check import ReasoningCheck
|
|
|
|
|
|
class CodeBasedReasoning(ReasoningCheck):
|
|
"""Validate live code citations within a caller-supplied ClassMeta scope.
|
|
|
|
Bind a file manager, metadata scope, and optional documented failure lines.
|
|
Example: ``check([current_line], negative_reason=[documented_obsolete_line])``
|
|
verifies presence and documented absence in that scope. Each call refreshes
|
|
``validation``; the caller owns any response correction.
|
|
"""
|
|
|
|
DELETED_INSTRUCTION = (
|
|
"An additional reason form is <DELETED>exact baseline passage</DELETED>. "
|
|
"Put the tags around the entire reason item. The enclosed nonblank line or contiguous block "
|
|
"must match the supplied baseline verbatim, including indentation and newlines, and be absent "
|
|
"verbatim from that same current file. Do not trim or reformat its contents. "
|
|
"Ordinary reason items still require one complete current source line each. "
|
|
"Deletion is evidence, not proof of a defect or of correctness: judge it against the immutable goal. "
|
|
"Do not restore goal-aligned removals merely because callers still expect them; consider caller migration. "
|
|
"A missing passage does not by itself prove that its entire symbol was removed."
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
file_manager: FileManager,
|
|
files: Sequence[ClassMeta],
|
|
evidence_lines: Sequence[str] = (),
|
|
baseline: Mapping[str, str] | None = None,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
"""Bind the caller's immutable baseline and file scope while leaving current reads fresh."""
|
|
|
|
super().__init__(changes=[])
|
|
self._file_manager = file_manager
|
|
self._files = tuple(files)
|
|
self._evidence_lines = tuple(evidence_lines)
|
|
baseline = baseline or {}
|
|
self._baseline = dict(map(lambda source: (source.location, baseline[source.location]),
|
|
filter(lambda source: source.location in baseline, self._files)))
|
|
self._logger = logger or logging.getLogger(__name__)
|
|
self.validation = ValidationResult()
|
|
|
|
def render_baseline(self) -> str:
|
|
"""Present the exact scoped baseline and shared citation contract to the judging model."""
|
|
|
|
passages = map(lambda source: f"BASELINE FILE: {source.location}\n{self._baseline[source.location]}",
|
|
filter(lambda source: source.location in self._baseline, self._files))
|
|
return self.DELETED_INSTRUCTION + "\n\n" + "\n\n".join(passages)
|
|
|
|
def __call__(self, reason: list[str], negative_reason: list[str] | None = None) -> bool:
|
|
"""Validate both citation buckets against freshly read code, retaining all correction errors."""
|
|
|
|
buckets = {"reason": reason, "negative_reason": [] if negative_reason is None else negative_reason}
|
|
lines: set[str] = set()
|
|
current: dict[str, str] = {}
|
|
errors: list[ValidationError] = []
|
|
locations = tuple(map(lambda source: source.location, self._files))
|
|
has_deleted = any("<DELETED>" in item for item in reason if isinstance(item, str))
|
|
# An unresolved scope can establish neither current code nor absence.
|
|
if not self._files:
|
|
errors.append(MechanicalValidationError(
|
|
location=ValidationLocation(["reason"]), error="No live file scope was supplied; citations cannot be verified.",
|
|
))
|
|
# Traceback display prefixes are presentation, not part of the cited source line.
|
|
evidence = set(filter(None, map(
|
|
lambda line: re.sub(r"^\s*>\s*", "", line).strip(),
|
|
"\n".join(self._evidence_lines).splitlines(),
|
|
)))
|
|
# Negative reasoning requires an actual documented observation.
|
|
if negative_reason and not evidence:
|
|
errors.append(MechanicalValidationError(
|
|
location=ValidationLocation(["negative_reason"]),
|
|
error="Negative citations require nonblank documented failure lines; missing evidence cannot prove resolved work.",
|
|
))
|
|
# Source retrieval belongs to this check and must observe every new invocation.
|
|
for source in self._files:
|
|
try:
|
|
# Negative evidence must distinguish failed reads from a successfully read empty file.
|
|
if has_deleted:
|
|
content = self._file_manager.repo_exec.read_bytes(source.resolve_path(self._file_manager)).decode("utf-8")
|
|
# Existing negative citations retain their documented-line normalization.
|
|
elif negative_reason:
|
|
content = self._file_manager.repo_exec.read_text(source.resolve_path(self._file_manager), encoding="utf-8")
|
|
# Ordinary citations retain the existing ClassMeta read contract.
|
|
else:
|
|
content = source.get_source(self._file_manager)
|
|
current[source.location] = content
|
|
lines.update(filter(None, map(str.strip, content.splitlines())))
|
|
except Exception as error:
|
|
self._logger.exception("Cannot retrieve live code for %s.", source.location)
|
|
errors.append(MechanicalValidationError(
|
|
location=ValidationLocation(["negative_reason" if negative_reason else "reason"]),
|
|
error=f"Cannot retrieve live code for {source.location}: {error!r}. No checkpoint evidence was substituted.",
|
|
))
|
|
|
|
# Both buckets are validated; an invalid collection cannot conceal faults in the other.
|
|
for name, citations in buckets.items():
|
|
root = ValidationLocation([name])
|
|
shape = validation_utils.validate_shape(citations, [["exact source line"]], location=root)
|
|
errors.extend(shape.errors)
|
|
# Malformed members cannot be interpreted as code, but the other bucket still needs checking.
|
|
if not shape.valid:
|
|
continue
|
|
# Negative evidence is optional; positive code must always support the verdict.
|
|
if name == "reason" and not citations:
|
|
errors.append(MechanicalValidationError(
|
|
location=root,
|
|
error="Supply a nonempty array of exact current code lines; empty reason cannot establish a verdict, even with negative_reason.",
|
|
))
|
|
# Validate every member so a correct line cannot conceal an unsupported citation.
|
|
for index, text in enumerate(citations):
|
|
location = root.child(ITEMS).child(str(index))
|
|
# Tagged passages retain exact whitespace and establish absence only in their baseline file.
|
|
if name == "reason" and ("<DELETED>" in text or "</DELETED>" in text):
|
|
payload = text.removeprefix("<DELETED>").removesuffix("</DELETED>")
|
|
complete = text.startswith("<DELETED>") and text.endswith("</DELETED>")
|
|
valid = complete and bool(payload.strip()) and "<DELETED>" not in payload and "</DELETED>" not in payload
|
|
matches = tuple(filter(lambda path: path in self._baseline and payload in self._baseline[path], current))
|
|
present = any(payload in current[path] for path in matches)
|
|
# Both historical presence and current absence must be established in the same scope.
|
|
if not valid or not matches or present:
|
|
errors.append(MechanicalValidationError(
|
|
location=location,
|
|
error=("Deleted citation is still present in live code. " if present
|
|
else "Deleted citation must be a nonblank exact baseline passage absent from the corresponding current file. ")
|
|
+ self.DELETED_INSTRUCTION,
|
|
))
|
|
continue
|
|
citation = text.strip()
|
|
# A traceback pointer is presentation, not source that can establish absence.
|
|
if name == "negative_reason":
|
|
citation = re.sub(r"^\s*>\s*", "", text).strip()
|
|
# Each member must cite one actual source line rather than prose or a block.
|
|
if not citation or len(text.splitlines()) != 1:
|
|
errors.append(MechanicalValidationError(
|
|
location=location,
|
|
error=f"{text!r} must be one nonblank code line; put separate code lines in separate array items.",
|
|
))
|
|
# Positive evidence must describe code that exists now.
|
|
elif name == "reason" and citation not in lines:
|
|
errors.append(MechanicalValidationError(
|
|
location=location,
|
|
error=f"{text!r} was not found as a complete line in live code at {locations!r}. Derive the resulting code from PRE plus DIFF; omit diff markers, line numbers, and explanations. PRE-only lines and removed lines are not current code. Correct this citation for either verdict.",
|
|
))
|
|
# Negative evidence must be documented in the supplied observation.
|
|
elif name == "negative_reason" and citation not in evidence:
|
|
errors.append(MechanicalValidationError(
|
|
location=location,
|
|
error=f"{text!r} is not a complete line in the supplied traceback. Cite a nonblank source line documented there, without line numbers or explanations; omit negative_reason if the trace contains no usable source lines.",
|
|
))
|
|
# Display normalization must not hide a literal source operator that is still present.
|
|
elif name == "negative_reason" and (text.strip() in lines or citation in lines):
|
|
errors.append(MechanicalValidationError(
|
|
location=location,
|
|
error=f"{text!r} is still present in live code at {locations!r}. Reassess the judgment or cite documented code that is actually absent; reason must separately cite current code supporting the verdict.",
|
|
))
|
|
|
|
self.validation = ValidationResult(errors=tuple(errors))
|
|
self.changes.append((self, self.validation.valid))
|
|
self._logger.info(
|
|
"CodeBasedReasoning valid=%s files=%s reason=%s negative_reason=%s errors=%s",
|
|
self.validation.valid, locations, reason, negative_reason, self.validation.stringify(),
|
|
)
|
|
return self.validation.valid
|