193 lines
8 KiB
Python
193 lines
8 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Construct failure-source candidates through decision-owned prediction."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import cast
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.validation_utils import (
|
|
MechanicalValidationError,
|
|
SemanticValidationError,
|
|
ValidationError,
|
|
ValidationLocation,
|
|
ValidationResult,
|
|
validate_shape,
|
|
)
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.source_needed_to_understand_failure_check import SourceNeededToUnderstandFailureCheck
|
|
from band_aider.judgement.failure_source_candidate_selection_strategy import (
|
|
FailureCandidateScope,
|
|
FailureSourceCandidateSelectionStrategy,
|
|
)
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.stabilization_decision import StabilizationDecision
|
|
|
|
|
|
SOURCE_LOCATION_RESPONSE_SHAPES: list[object] = [{
|
|
"source_locations": ["repo-relative or absolute source path"],
|
|
}]
|
|
|
|
|
|
class PredictiveFailureSourceCandidateSelectionDecision(StabilizationDecision):
|
|
"""Extract and validate one complete set of source locations from a trace.
|
|
|
|
Example: ``decision()`` retries a rejected location response until every
|
|
returned path identifies a current source supported by the supplied trace.
|
|
"""
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""Return the strict predictive source-location response shape."""
|
|
|
|
return SOURCE_LOCATION_RESPONSE_SHAPES
|
|
|
|
def __init__(self, *, file_manager: FileManager, failure: FailingTest, candidates: list[ClassMeta]) -> None:
|
|
"""Bind the target inventory and immutable failure evidence to one decision."""
|
|
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[],
|
|
readonly_files=[],
|
|
failure_error_message=failure.error_message,
|
|
failure_stack_trace=list(failure.stack_trace),
|
|
failure_raw_output=failure.raw_output,
|
|
)
|
|
super().__init__(
|
|
client=file_manager.client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
disambiguation=failure.test.location,
|
|
)
|
|
self._candidate_by_location = dict(map(lambda source: (source.location, source), candidates))
|
|
self.sources: list[ClassMeta] = []
|
|
self._evidence = "\n".join(map(str, failure.stack_trace)).replace("\\", "/")
|
|
section = self._prompt_formatter_builder.generate
|
|
self.prompt = section("FAILURE SOURCE CANDIDATE SELECTION")(lambda: "\n\n".join((
|
|
"Extract every production source file identified by this failure trace. "
|
|
"Return only files supported by the trace, normalize each result to a "
|
|
"repo-relative or absolute path, and do not return tests or invented paths.",
|
|
'Return strict JSON only: {"source_locations": ["path"]}.',
|
|
section("FAILURE STACK TRACE", escape=True)(self._evidence),
|
|
)))
|
|
|
|
def _decide(
|
|
self,
|
|
response: object | None = None,
|
|
validation: ValidationResult | None = None,
|
|
) -> tuple[object | None, ValidationResult]:
|
|
"""Request locations and reject the complete response until all are current."""
|
|
|
|
# Add only the latest validation rejection to the stationary selection request.
|
|
prompt = (
|
|
self.prompt
|
|
if validation is None
|
|
else super()._format_prompt_w_feedback(validation)
|
|
)
|
|
root = ValidationLocation([])
|
|
try:
|
|
data = json.loads(self.escalation.complete(
|
|
self.render_prompt_for_turn(prompt),
|
|
disambiguation=self.disambiguation,
|
|
).strip())
|
|
except json.JSONDecodeError as error:
|
|
return error.doc, ValidationResult(errors=(ValidationError(
|
|
location=root,
|
|
error=str(error),
|
|
),))
|
|
|
|
result = validate_shape(
|
|
data,
|
|
self.valid_response_references,
|
|
location=root,
|
|
)
|
|
# Preserve structural validation before reading the required locations.
|
|
if not result.valid:
|
|
return data, result
|
|
payload = cast("dict[str, object]", data)
|
|
locations = cast("list[str]", payload["source_locations"])
|
|
errors: list[ValidationError] = []
|
|
selected: dict[str, ClassMeta] = {}
|
|
# Require predictive selection to identify at least one source owner.
|
|
if not locations:
|
|
errors.append(SemanticValidationError(
|
|
location=root.child("source_locations"),
|
|
error="must contain at least one source location.",
|
|
))
|
|
# Validate every prediction as one current, trace-supported target source.
|
|
for index, location in enumerate(locations):
|
|
try:
|
|
canonical = self._file_manager.source_location(location)
|
|
except ValueError:
|
|
canonical = ""
|
|
source = self._candidate_by_location.get(canonical)
|
|
# Predictions cannot expand the preselected trace-supported source set.
|
|
if source is None:
|
|
errors.append(MechanicalValidationError(
|
|
location=root.child("source_locations").child(str(index)),
|
|
error=f"{location!r} is not an admitted trace-supported managed source.",
|
|
))
|
|
else:
|
|
selected.setdefault(source.location, source)
|
|
missing = self._candidate_by_location.keys() - selected.keys()
|
|
# Omitting a trace-supported source would change downstream ownership authority.
|
|
if missing:
|
|
errors.append(SemanticValidationError(
|
|
location=root.child("source_locations"),
|
|
error=f"must include every admitted source; missing: {', '.join(sorted(missing))}.",
|
|
))
|
|
# Keep every invalid prediction in the decision-owned correction loop.
|
|
if errors:
|
|
return data, ValidationResult(errors=tuple(errors))
|
|
|
|
self.sources = list(selected.values())
|
|
self.outcome = "selected"
|
|
self.reason = "Every predicted location is current and supported by the failure trace."
|
|
self.additional = {
|
|
"source_locations": [source.location for source in self.sources],
|
|
}
|
|
self.changes.append((self, True))
|
|
return data, ValidationResult()
|
|
|
|
|
|
class PredictiveFailureSourceCandidateSelectionStrategy(FailureSourceCandidateSelectionStrategy):
|
|
"""Construct trace-supported candidates through a correcting LLM decision.
|
|
|
|
Example: ``strategy.candidates_for_failure(file_manager=manager, failure=failing, scope=FailureCandidateScope.BOTH)`` returns
|
|
only after every predicted location validates against the current manager.
|
|
"""
|
|
|
|
def candidates_for_failure(
|
|
self,
|
|
*,
|
|
file_manager: FileManager,
|
|
failure: FailingTest,
|
|
scope: FailureCandidateScope,
|
|
) -> list[ClassMeta]:
|
|
"""Combine mechanical test matches with a complete validated source prediction."""
|
|
|
|
check = SourceNeededToUnderstandFailureCheck(
|
|
file_manager=file_manager,
|
|
scope=scope,
|
|
failure_error_message="",
|
|
failure_stack_trace=list(failure.stack_trace),
|
|
failure_raw_output="",
|
|
)
|
|
check()
|
|
sources = list(filter(lambda candidate: candidate.content_type is ContentType.Source, check.matching_files))
|
|
tests = list(filter(lambda candidate: candidate.content_type is ContentType.Test, check.matching_files))
|
|
# Tests are mechanically admitted; source extraction only runs for a nonempty source set.
|
|
if sources:
|
|
decision = PredictiveFailureSourceCandidateSelectionDecision(
|
|
file_manager=file_manager,
|
|
failure=failure,
|
|
candidates=sources,
|
|
)
|
|
decision()
|
|
sources = decision.sources
|
|
return [*sources, *tests]
|