232 lines
11 KiB
Python
232 lines
11 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import cast
|
|
|
|
from band_aider.ai_client.json_completion import load_json_completion
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.validation_utils import (
|
|
ITEMS,
|
|
MechanicalValidationError,
|
|
ValidationError,
|
|
ValidationLocation,
|
|
ValidationResult,
|
|
SemanticValidationError,
|
|
validate_shape,
|
|
)
|
|
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
|
|
from band_aider.judgement.code_based_reasoning import CodeBasedReasoning
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.fidelity import Fidelity
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.prompt_formatter import format_prompt
|
|
from band_aider.judgement.work_definition_constants import Work
|
|
from band_aider.judgement.stabilization_decision import StabilizationDecision
|
|
|
|
|
|
class SourceRemediationGuidanceCheck(StabilizationDecision):
|
|
"""Derive repair work; ``check()`` validates response citations before its work_needed reaches Alpha."""
|
|
|
|
_VALID_RESPONSE_REFERENCES: list[object] = [
|
|
{
|
|
"work_needed": "one compact ordered instruction grounded in current failure evidence",
|
|
"reason": ["exact current selected-source line"],
|
|
},
|
|
]
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""Return every JSON reference accepted by source-remediation guidance."""
|
|
|
|
return self._VALID_RESPONSE_REFERENCES
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
client: OpenAICompatibleClient,
|
|
file_manager: FileManager,
|
|
request: AiderEditRequest,
|
|
original_cot: str,
|
|
source: ClassMeta,
|
|
failure_evidence: str,
|
|
precedent_diffs: str,
|
|
readonly_sources: list[ClassMeta] | None = None,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
"""Initialize one source-remediation guidance judgement."""
|
|
|
|
super().__init__(
|
|
client=client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
logger=logger or logging.getLogger(__name__),
|
|
)
|
|
self._source = source
|
|
self._reason_check = CodeBasedReasoning(
|
|
file_manager=file_manager, files=[source], logger=self.logger,
|
|
baseline=file_manager.pristine_sources,
|
|
)
|
|
renderer_prompt = (
|
|
"ORIGINAL COT: IMMUTABLE STABILIZATION TARGET\n"
|
|
+ format_prompt(
|
|
self._prompt_formatter_builder,
|
|
"ORIGINAL STABILIZATION GOAL",
|
|
original_cot.strip(),
|
|
depth=2,
|
|
)
|
|
+ "\n\n"
|
|
"ROLE AND EDIT BOUNDARIES\n"
|
|
"Identify only source work owned by the selected remediation file."
|
|
)
|
|
renderer = ContextualFileRenderer.build(
|
|
backlog_guidance=request.guidance,
|
|
file_manager=file_manager,
|
|
client=client,
|
|
editable_files=[source],
|
|
readonly_files=list(readonly_sources or []),
|
|
prompt=renderer_prompt,
|
|
fault=failure_evidence,
|
|
logger=self.logger.getChild("contextual-file-renderer"),
|
|
prompt_formatter_builder=self._prompt_formatter_builder,
|
|
)
|
|
readonly_context = lambda: "\n\n".join(map(
|
|
lambda related: renderer.cheap_read(
|
|
related,
|
|
Fidelity.AUTO | Fidelity.CDC,
|
|
),
|
|
readonly_sources or [],
|
|
)) or "None."
|
|
section = self._prompt_formatter_builder.generate
|
|
prompt = lambda: (
|
|
"Determine the source remediation needed to satisfy the immutable stabilization goal.\n"
|
|
"The current source, failed test, and precedent diffs describe state or deviation.\n"
|
|
"They may diagnose the required edit but may not replace, weaken, contradict, or reverse the goal.\n"
|
|
"Do not restore a removed API merely because stale callers or failures expect it.\n"
|
|
f"{Work.NECESSITY}\n\n"
|
|
f"{Work.SOURCE_AUTHORITY}\n\n"
|
|
"Ground remediation in a concrete current-file divergence localized by this failure. "
|
|
"When source work is needed, work_needed must be a compact ordered instruction: "
|
|
"1. Identify the exact current operation or behavior, "
|
|
"2. State the smallest source change that aligns it with the COT, "
|
|
"3. State the focused verification.\n"
|
|
"Do not make a generic edit assignment, revive goal-aligned removed behavior, or repair unrelated compiler debt.\n"
|
|
f"{self._reason_check.render_baseline()}\n\n"
|
|
"For ordinary citations, reason must be a nonempty array of all exact current selected-source lines relied on for every part of the guidance, including relevant conditions, operations, and results; one complete nonblank line per item, without line numbers or diff markers.\n"
|
|
"Every ordinary item must match current selected source: items are a conjunction, so one invalid item rejects the complete response; readonly evidence cannot satisfy this static check, which cannot judge relevance or completeness.\n"
|
|
"Only a successfully read blank selected file permits a nonempty array of nonblank explanations instead of citations; work_needed must still give concrete restoration or repair guidance, never no-work. Missing or unreadable files do not qualify.\n"
|
|
"Return strict JSON only with this shape:\n"
|
|
f"{json.dumps(self.valid_response_references[0], separators=(',', ':'))}\n\n"
|
|
"ORIGINAL COT: IMMUTABLE STABILIZATION TARGET\n"
|
|
f"{section('ORIGINAL STABILIZATION GOAL', escape=True)(original_cot.strip())}\n\n"
|
|
+ section('SELECTED SOURCE')(lambda: "\n\n".join((
|
|
section('LOCATION', escape=True)(source.location),
|
|
section('SUMMARY', escape=True)(source.summary),
|
|
)))
|
|
+ "\n\n"
|
|
"CURRENT SOURCE EVIDENCE\n"
|
|
f"{section('CURRENT SOURCE EVIDENCE', escape=True)(renderer.cheap_read(
|
|
source,
|
|
Fidelity.AUTO | Fidelity.CDC | Fidelity.DIFF,
|
|
))}\n\n"
|
|
"READONLY REMEDIATION CONTEXT\n"
|
|
f"{section('READONLY REMEDIATION CONTEXT', escape=True)(readonly_context)}\n\n"
|
|
"DEPENDENCY REFERENCE CARDS\n"
|
|
f"{section('DEPENDENCY REFERENCE CARDS', escape=True)(renderer.get_reference_card_evidence())}\n\n"
|
|
"FAILURE EVIDENCE\n"
|
|
f"{section('FAILURE EVIDENCE', escape=True)(failure_evidence)}\n\n"
|
|
"CHANGED PRECEDENT DIFFS\n"
|
|
f"{section('CHANGED PRECEDENT DIFFS', escape=True)(precedent_diffs or 'None.')}"
|
|
)
|
|
self.prompt = section("SOURCE REMEDIATION GUIDANCE")(prompt)
|
|
|
|
def _decide(
|
|
self,
|
|
response: object | None = None,
|
|
validation: ValidationResult | None = None,
|
|
) -> tuple[object | None, ValidationResult]:
|
|
"""Request, validate, and record one source-remediation instruction."""
|
|
|
|
# Response correction preserves the composed decision, including localized guidance.
|
|
base_prompt = self.prompt
|
|
# Add only the preceding rejection when this is a correction attempt.
|
|
# Keep the original request and only the latest rejected response in each correction.
|
|
if validation is None:
|
|
prompt = base_prompt
|
|
else:
|
|
self.logger.warning("Source remediation retrying response=%s errors=%s", self.escalation.last_response, validation.stringify())
|
|
prompt = super()._format_prompt_w_feedback(
|
|
validation,
|
|
base_prompt=base_prompt,
|
|
) + (
|
|
"\nReassess the whole remediation judgment and return a complete replacement reason array "
|
|
"retaining all supporting evidence still relied on, not just corrected items. Every item must pass."
|
|
)
|
|
root = ValidationLocation([])
|
|
try:
|
|
data = load_json_completion(
|
|
self.escalation,
|
|
self.render_prompt_for_turn(prompt),
|
|
prompt_formatter_builder=self._prompt_formatter_builder,
|
|
)
|
|
except json.JSONDecodeError as error:
|
|
return error.doc, ValidationResult(errors=(ValidationError(
|
|
location=root,
|
|
error=str(error),
|
|
),))
|
|
|
|
validation = validate_shape(
|
|
data,
|
|
self.valid_response_references,
|
|
location=root,
|
|
)
|
|
# Keep incomplete remediation text in the parent-owned correction loop.
|
|
if not validation.valid:
|
|
return data, validation
|
|
response = cast("dict[str, object]", data)
|
|
work_needed = cast(str, response["work_needed"])
|
|
reason = cast(list[str], response["reason"])
|
|
# Require a concrete ordered remediation rather than letting empty output stop Alpha work.
|
|
if not work_needed.strip():
|
|
return data, ValidationResult(errors=(SemanticValidationError(
|
|
location=root.child("work_needed"),
|
|
error="must be a nonempty string.",
|
|
),))
|
|
|
|
# Even blank-file repair must supply an explanation array.
|
|
if not reason:
|
|
return data, ValidationResult(errors=(ValidationError(
|
|
location=root.child("reason"), error="must be a nonempty array of strings.",
|
|
),))
|
|
# Only a successful repository read can establish the blank-file exception.
|
|
try:
|
|
content = self._file_manager.repo_exec.read_text(self._source.resolve_path(self._file_manager))
|
|
except Exception as error:
|
|
self.logger.exception("Cannot read selected remediation source %s.", self._source.location)
|
|
return data, ValidationResult(errors=(MechanicalValidationError(
|
|
location=root.child("reason"), error=f"Cannot read selected source: {error!r}. The blank-file exception does not apply.",
|
|
),))
|
|
errors: list[ValidationError] = []
|
|
# Blank remediation still requires concrete repair guidance and nonblank explanations.
|
|
if not content.strip() and not any("<DELETED>" in item or "</DELETED>" in item for item in reason):
|
|
errors.extend(ValidationError(
|
|
location=root.child("reason").child(ITEMS).child(str(index)),
|
|
error="must be a nonblank explanation for the blank selected file.",
|
|
) for index, item in enumerate(reason) if not item.strip())
|
|
elif not self._reason_check(reason):
|
|
errors.extend(self._reason_check.validation.errors)
|
|
# Publish no guidance until every supplied reason passes.
|
|
if errors:
|
|
return data, ValidationResult(errors=tuple(errors))
|
|
|
|
self.outcome = "guidance"
|
|
self.reason = "\n".join(reason)
|
|
self.additional = {"work_needed": work_needed}
|
|
self.changes.append((self, True))
|
|
self.logger.info("Source remediation validated reason=%s work_needed=%s", reason, work_needed)
|
|
return data, ValidationResult()
|