1094 lines
53 KiB
Python
1094 lines
53 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import json
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
import random
|
|
from typing import cast, Sequence
|
|
|
|
from band_aider.ai_client.llm_logging import current_logger, set_current_logger
|
|
from band_aider.ai_client.token_budget_estimator import TokenBudgetEstimator
|
|
from band_aider.context import provenance, token_budget_state
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.edited_test_inventory import EditedTestInventory
|
|
from band_aider.context.validation_receipts import ValidationReceipts
|
|
from band_aider.context.memento import Memento
|
|
from band_aider.context.provenance import DeadEndError
|
|
from band_aider.context.token_budget_state import DEFAULT_MAX_PROMPT_TOKENS
|
|
from band_aider.judgement.dead_end_decision import DeadEndDecisionBuilder
|
|
from band_aider.judgement.contextual_cot_renderer import ContextualCotRenderer
|
|
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
|
|
from band_aider.judgement.edit_request import (
|
|
AiderEditRequest,
|
|
AiderEditRequestMask,
|
|
AiderFailureMode,
|
|
)
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.prompt_localization import PromptLocalization
|
|
from band_aider.judgement.file_patcher import DirectCodeEditFailed, FilePatcher, QuickRetryError
|
|
from band_aider.judgement.prompt_formatter import PromptFormatterBuilder
|
|
from band_aider.judgement.search_replace_token_generator import (
|
|
ContextWideningRequired,
|
|
RepairEvidenceBuilder,
|
|
SearchReplaceTokenGenerator,
|
|
)
|
|
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult, EditorRecord
|
|
from band_aider.strategies.compiler import (
|
|
BlackboardCompiler,
|
|
CompileAttempt,
|
|
CompileError,
|
|
CompilerFactory,
|
|
TestRunCompileError,
|
|
TypeCheckError,
|
|
)
|
|
from band_aider.strategies.language_strategy_set import LanguageStrategySet
|
|
from band_aider.ui.dto.direct_code_editor_status import DirectCodeEditorStatusDTO
|
|
from band_aider.ui.status import publish_supreme_status
|
|
|
|
|
|
INCREASE_VISIBLE_ERR_EVERY_N_BLOCKS = 3
|
|
|
|
|
|
def _diagnostic_identities(
|
|
errors: Sequence[CompileError],
|
|
*,
|
|
with_coordinates: bool,
|
|
) -> tuple[str, ...]:
|
|
"""Return deterministic diagnostic identities for presentation-only movement labels."""
|
|
|
|
identities: list[str] = []
|
|
# Preserve reported order so substituted failures remain visibly distinct.
|
|
for error in errors:
|
|
value = error
|
|
# Remove only known structured source coordinates for coordinate-independent comparison.
|
|
if isinstance(error, TypeCheckError) and not with_coordinates:
|
|
value = replace(error, path=None, line=None, column=None)
|
|
identities.append(f"{type(error).__module__}.{type(error).__qualname__}:{value!r}")
|
|
return tuple(identities)
|
|
|
|
|
|
def _diagnostic_movement(
|
|
previous_errors: Sequence[CompileError] | None,
|
|
errors: Sequence[CompileError],
|
|
) -> str:
|
|
"""Classify diagnostic movement without supplying a repair-loop decision signal."""
|
|
|
|
# The absence of blocking static or runtime semantic failures is the clearest terminal state.
|
|
if not errors:
|
|
return "clean"
|
|
# The first observed check result has no prior failure set to compare.
|
|
if previous_errors is None:
|
|
return "initial"
|
|
if _diagnostic_identities(previous_errors, with_coordinates=True) == _diagnostic_identities(
|
|
errors, with_coordinates=True
|
|
):
|
|
return "same diagnostic set"
|
|
if _diagnostic_identities(previous_errors, with_coordinates=False) == _diagnostic_identities(
|
|
errors, with_coordinates=False
|
|
):
|
|
return "coordinates changed"
|
|
return "diagnostics changed"
|
|
|
|
class DirectCodeEditor(CodeEditor):
|
|
"""Apply SEARCH/REPLACE repairs with snapshots and compiler-owned test inventory.
|
|
|
|
Callers supply the edit request and bind the runtime FileManager before editing.
|
|
Each call scopes its compiler's inventory and contains retry failures locally.
|
|
|
|
Example::
|
|
|
|
editor.file_manager = file_manager
|
|
result = editor.edit(request)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
token_generator: SearchReplaceTokenGenerator,
|
|
prompt_formatter_builder: PromptFormatterBuilder,
|
|
file_patcher: FilePatcher,
|
|
repo_root: Path,
|
|
) -> None:
|
|
"""Initialize the direct editor with collaborators and repository state."""
|
|
|
|
self.token_generator = token_generator
|
|
self.prompt_formatter_builder = prompt_formatter_builder
|
|
self.file_patcher = file_patcher
|
|
self.repo_root = Path(str(repo_root))
|
|
self._file_manager: FileManager | None = None
|
|
self.attempts_since_acceptance = 0
|
|
self.error_permit = 1
|
|
|
|
@property
|
|
def file_manager(self) -> FileManager | None:
|
|
"""Return the runtime FileManager used for diff-aware prompt rendering."""
|
|
|
|
return self._file_manager
|
|
|
|
@file_manager.setter
|
|
def file_manager(self, value: FileManager | None) -> None:
|
|
"""Bind the runtime FileManager to DirectCodeEditor-owned collaborators."""
|
|
|
|
self._file_manager = value
|
|
self.token_generator.file_manager = value
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Return None because DirectCodeEditor is a concrete editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Prepare scope, request SEARCH/REPLACE output, and apply it directly."""
|
|
|
|
self.attempts_since_acceptance = 0
|
|
self.error_permit = 1
|
|
request.validation_receipts = ValidationReceipts()
|
|
retained_records = list(request.editor_records)
|
|
previous_logger = current_logger()
|
|
if logger is not None:
|
|
set_current_logger(logger)
|
|
logger.info(
|
|
"DirectCodeEditor starting edit: editable=%s readonly=%s.",
|
|
len(request.editable_files),
|
|
len(request.readonly_files),
|
|
)
|
|
try:
|
|
editable, readonly, _ = self.prepare_edit_scope(request)
|
|
if not editable:
|
|
raise DirectCodeEditFailed("No editable files were available for direct editing.")
|
|
if request.context is None:
|
|
raise DirectCodeEditFailed("Compilation requires an owning workflow context.")
|
|
if self.file_manager is None:
|
|
raise DirectCodeEditFailed("Compilation requires a bound FileManager.")
|
|
strategy_set = LanguageStrategySet.get(self.file_manager)
|
|
compiler_factory = cast(CompilerFactory, strategy_set[CompilerFactory])
|
|
outer_mementos = self._capture_mementos(editable, stack=True)
|
|
# Scope the initial compiler's inventory across every candidate compiler.
|
|
try:
|
|
compiler = cast(
|
|
BlackboardCompiler,
|
|
compiler_factory.build(
|
|
CompileAttempt(
|
|
request=request,
|
|
mementos=outer_mementos,
|
|
frontier_provider=request.frontier_provider,
|
|
)
|
|
),
|
|
)
|
|
with compiler.get_blackboard() as blackboard:
|
|
changed_paths, patch_summary, _ = self._edit_with_repairs(
|
|
request,
|
|
editable,
|
|
readonly,
|
|
compiler=compiler,
|
|
compiler_factory=compiler_factory,
|
|
blackboard=blackboard,
|
|
outer_mementos=outer_mementos,
|
|
logger=logger,
|
|
)
|
|
# Only Alpha remediation exports execution evidence to semantic acceptance.
|
|
if request.is_remediation:
|
|
request.validation_receipts = ValidationReceipts(blackboard.receipts, blackboard.accepted_generation)
|
|
(logger or logging.getLogger(__name__)).info("Validation receipts after editing:\n%s", request.validation_receipts.render())
|
|
except DeadEndError as error:
|
|
request.validation_receipts = ValidationReceipts()
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, outer_mementos)))
|
|
# Historical revival must receive the candidate before its owning scope restores bytes.
|
|
if rejected_diff:
|
|
error.message += "\n\n" + RepairEvidenceBuilder.compiler_rejection(str(error), rejected_diff)
|
|
# Preserve no uncommitted edit when phase ownership redirects this item.
|
|
for memento in reversed(outer_mementos):
|
|
memento.rollback()
|
|
request.editor_records = retained_records
|
|
raise
|
|
except Exception as error:
|
|
request.validation_receipts = ValidationReceipts()
|
|
# Inventory cleanup can fail after snapshots were already resolved.
|
|
unresolved = filter(lambda memento: memento.repo_exec.exists(memento.path), outer_mementos)
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, unresolved)))
|
|
# Retain failed candidate evidence for the caller even when editing raises.
|
|
if rejected_diff:
|
|
error.add_note(RepairEvidenceBuilder.compiler_rejection(str(error), rejected_diff))
|
|
# Restore the complete episode when construction, editing, or cleanup fails.
|
|
for memento in reversed(outer_mementos):
|
|
memento.rollback()
|
|
request.editor_records = retained_records
|
|
raise
|
|
summary = self._summary(changed_paths, patch_summary)
|
|
if logger is not None:
|
|
logger.info("DirectCodeEditor edit completed: changed=%s.", len(changed_paths))
|
|
return CodeEditorResult(changed_paths, summary, list(request.editor_records))
|
|
except DeadEndError:
|
|
raise
|
|
except Exception as error:
|
|
if logger is not None:
|
|
logger.exception("DirectCodeEditor edit failed.")
|
|
# An interrupted remediation remains owned by the enclosing workflow.
|
|
if request.is_remediation:
|
|
raise
|
|
request.editor_records = retained_records
|
|
return CodeEditorResult(
|
|
[], self._failure_summary("\n\n".join((str(error), *getattr(error, "__notes__", ())))),
|
|
list(retained_records),
|
|
)
|
|
finally:
|
|
publish_supreme_status(direct_code_editor=None)
|
|
progress_logger = current_logger() or logging.getLogger(__name__)
|
|
# Mirror the inactive UI transition in the established runtime log stream.
|
|
progress_logger.info("DirectCodeEditor progress: inactive")
|
|
if logger is not None:
|
|
set_current_logger(previous_logger)
|
|
|
|
def prepare_edit_scope(
|
|
self,
|
|
request: AiderEditRequest,
|
|
) -> tuple[list[Path], list[Path], str]:
|
|
"""Validate edit scope after callers have preflighted required new files."""
|
|
|
|
editable = [self._absolute_path(path) for path in request.editable_files]
|
|
readonly = [self._absolute_path(path) for path in request.readonly_files]
|
|
if self.file_manager is not None:
|
|
editable = [path for path in editable if self._is_managed_path(path)]
|
|
readonly = [path for path in readonly if self._is_managed_path(path)]
|
|
readonly = [
|
|
path
|
|
for path in readonly
|
|
if path not in editable
|
|
]
|
|
return editable, readonly, f"Prepared {len(editable)} editable file(s) and {len(readonly)} readonly file(s)."
|
|
|
|
def _is_managed_path(self, path: Path) -> bool:
|
|
"""Return whether FileManager discovered one path as source or test input."""
|
|
|
|
file_manager = self.file_manager
|
|
if file_manager is None:
|
|
return True
|
|
location = self._relative_or_absolute(path).as_posix()
|
|
return location in file_manager.src_index or location in file_manager.test_index
|
|
|
|
def _absolute_path(self, path: Path) -> Path:
|
|
"""Resolve one path against the repository root when needed."""
|
|
|
|
return Path(str(path)) if path.is_absolute() else self.repo_root / path
|
|
|
|
def _relative_or_absolute(self, path: Path) -> Path:
|
|
"""Return a repo-relative path when the file lives inside the repository."""
|
|
|
|
absolute = self._absolute_path(path)
|
|
try:
|
|
return absolute.relative_to(self.repo_root)
|
|
except ValueError:
|
|
return absolute
|
|
|
|
def _summary(self, changed_paths: list[Path], patch_summary: str) -> str:
|
|
"""Render the complete direct-edit summary used by existing caller logs."""
|
|
|
|
changed_lines = [
|
|
f"- {path.relative_to(self.repo_root).as_posix()}"
|
|
for path in changed_paths
|
|
]
|
|
return "\n".join(
|
|
[
|
|
"DIRECT EDIT",
|
|
"Changed files:",
|
|
*changed_lines,
|
|
"",
|
|
"Patch summary:",
|
|
patch_summary,
|
|
]
|
|
).strip()
|
|
|
|
def _edit_with_repairs(
|
|
self,
|
|
request: AiderEditRequest,
|
|
editable: list[Path],
|
|
readonly: list[Path],
|
|
*,
|
|
compiler: BlackboardCompiler,
|
|
compiler_factory: CompilerFactory,
|
|
blackboard: EditedTestInventory,
|
|
outer_mementos: list[Memento],
|
|
logger: logging.Logger | None,
|
|
) -> tuple[list[Path], str, str]:
|
|
"""Search static and runtime semantic results while preserving original source."""
|
|
|
|
file_manager = self.file_manager
|
|
# Direct repair requires the FileManager already validated by the public edit boundary.
|
|
if file_manager is None:
|
|
raise DirectCodeEditFailed("Compilation requires a bound FileManager.")
|
|
raw_output = ""
|
|
retained_records = list(request.editor_records)
|
|
remediation = request.is_remediation
|
|
presenter = ContextualFileRenderer.build(
|
|
file_manager=file_manager,
|
|
client=self.token_generator.clients[0],
|
|
editable_files=editable,
|
|
readonly_files=readonly,
|
|
prompt=(
|
|
"ORIGINAL COT: IMMUTABLE STABILIZATION TARGET\n"
|
|
+ self.token_generator.render_section(
|
|
"ORIGINAL STABILIZATION GOAL",
|
|
depth=3,
|
|
)(file_manager.prompt.strip())
|
|
+ "\n\n"
|
|
"ROLE AND EDIT BOUNDARIES\n"
|
|
f"{request.authoritative_prompt.strip()}"
|
|
),
|
|
pinned_context=request.pinned_context,
|
|
backlog_guidance=request.guidance,
|
|
failure_error_message="",
|
|
failure_stack_trace=(),
|
|
failure_raw_output="",
|
|
fault="",
|
|
logger=(logger or logging.getLogger(__name__)).getChild("contextual-file-renderer"),
|
|
repair_episode=True,
|
|
prompt_formatter_builder=self.prompt_formatter_builder,
|
|
)
|
|
cot_renderer = ContextualCotRenderer(
|
|
client=self.token_generator.clients[0],
|
|
current_cot=file_manager.prompt,
|
|
prompt_formatter_builder=self.prompt_formatter_builder,
|
|
logger=(logger or logging.getLogger(__name__)).getChild(
|
|
"contextual-cot-renderer"
|
|
),
|
|
)
|
|
try:
|
|
current_errors = compiler.compile(editable)
|
|
current_blocking_errors = tuple(current_errors)
|
|
current_error_text = self._render_compiler_errors(current_blocking_errors)
|
|
current_error_count = len(current_blocking_errors)
|
|
semantic_default = (
|
|
AiderFailureMode.SEMANTIC
|
|
if AiderEditRequestMask(
|
|
request, AiderFailureMode.SEMANTIC
|
|
).has_semantic_failure
|
|
else AiderFailureMode.DEFAULT
|
|
)
|
|
workflow_level = AiderFailureMode.attempt_level(
|
|
current_errors,
|
|
default=semantic_default,
|
|
)
|
|
last_rejected_feedback = ""
|
|
patch_repair_feedback = ""
|
|
mechanical_candidate_number = 0
|
|
repair_turn = 0
|
|
unapplied_attempts = 0
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="initial compile",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=None,
|
|
)
|
|
# Search one accepted repair baseline at a time until all checks pass.
|
|
while True:
|
|
# Retain earlier accepted edits; monitor whether their history unduly influences later work.
|
|
# Rejected candidates never join this history. Whole-attempt rollback restores its entry list.
|
|
presenter.backlog_guidance = tuple(request.guidance)
|
|
# Widen the next prompt after each full group of unaccepted repair blocks.
|
|
if (
|
|
self.attempts_since_acceptance
|
|
% INCREASE_VISIBLE_ERR_EVERY_N_BLOCKS
|
|
== INCREASE_VISIBLE_ERR_EVERY_N_BLOCKS - 1
|
|
):
|
|
self.error_permit *= 2
|
|
current_error_text = self._render_compiler_errors(current_blocking_errors)
|
|
request_mask = AiderEditRequestMask(request, workflow_level)
|
|
visible_request = request_mask.projected()
|
|
focused_failure = "\n".join(filter(None, (
|
|
current_error_text,
|
|
visible_request.failure_error_message,
|
|
*visible_request.failure_stack_trace,
|
|
visible_request.failure_raw_output,
|
|
)))
|
|
presenter.set_failure(focused_failure)
|
|
feedback = (
|
|
RepairEvidenceBuilder(self.token_generator, request_mask)
|
|
.with_reference_cards(presenter.get_reference_card_text())
|
|
.with_rejected_candidate(last_rejected_feedback)
|
|
.with_patch_feedback(patch_repair_feedback)
|
|
.render(current_errors, current_error_text)
|
|
)
|
|
base_prompt = self.token_generator.render_prompt(
|
|
request_mask,
|
|
editable_files=editable,
|
|
readonly_files=readonly,
|
|
renderer=presenter,
|
|
repair_feedback=feedback,
|
|
)
|
|
turn_prompt = cot_renderer.render(request_mask, base_prompt)
|
|
turn_prompt = PromptLocalization(
|
|
client=self.token_generator.clients[0], file_manager=file_manager,
|
|
request=visible_request, owner=type(self).__name__, logger=logger,
|
|
).localize(turn_prompt)
|
|
estimated_tokens = TokenBudgetEstimator(
|
|
tokens_per_word=token_budget_state.get_token_budget_settings().tokens_per_word,
|
|
).estimate(turn_prompt)
|
|
visible_error_count = len(current_blocking_errors[:self.error_permit])
|
|
# Oversized editor requests expose complete diagnostics without selecting more source.
|
|
if estimated_tokens > DEFAULT_MAX_PROMPT_TOKENS:
|
|
complete_errors = self._render_compiler_errors(
|
|
current_blocking_errors, all_errors=True,
|
|
)
|
|
complete_failure = "\n".join((
|
|
"Run information:", request.failure_run_info,
|
|
f"Error message: {request.failure_error_message}",
|
|
"Failure evidence:", *request.failure_stack_trace,
|
|
"Raw failure output:", request.failure_raw_output,
|
|
))
|
|
complete_sections = (
|
|
self.token_generator.render_section(
|
|
"CURRENT COMPILER STATE", depth=2, escape=True,
|
|
)(complete_errors or "None."),
|
|
self.token_generator.render_section(
|
|
"FAILURE EVIDENCE", depth=2, escape=True,
|
|
)(complete_failure),
|
|
)
|
|
# Keep the composed source exact and avoid repeating evidence already present in full.
|
|
turn_prompt = "\n\n".join((
|
|
turn_prompt,
|
|
*filter(lambda section: section not in turn_prompt, complete_sections),
|
|
))
|
|
visible_error_count = len(current_blocking_errors)
|
|
if logger is not None:
|
|
logger.info("DirectCodeEditor requesting patch.")
|
|
repair_turn += 1
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="awaiting model",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
visible_error_count=visible_error_count,
|
|
previous_errors=current_errors,
|
|
)
|
|
mementos = self._capture_mementos(editable, stack=True)
|
|
# Use the captured inner rollback boundary while generating this candidate.
|
|
patch_outcome = "not applied"
|
|
patch_summary = ""
|
|
unresolved_patch_issue_count = 0
|
|
candidate_records: list[EditorRecord] = []
|
|
try:
|
|
raw_output = self.token_generator.generate(
|
|
request,
|
|
base_prompt=turn_prompt,
|
|
editable_files=editable,
|
|
logger=logger,
|
|
unapplied_attempts=unapplied_attempts,
|
|
)
|
|
except ContextWideningRequired as error:
|
|
# Release the unused candidate boundary without unwinding accepted work.
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="expanding context",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
decision_branch="retrying",
|
|
decision_reason=str(error),
|
|
)
|
|
self.attempts_since_acceptance += 1
|
|
presenter.refine(focused_failure, stale_files=[])
|
|
unapplied_attempts = 0
|
|
# No growth or whole-file saturation cannot exhaust correction attempts.
|
|
continue
|
|
except Exception:
|
|
# The outer edit boundary captures the complete candidate before rollback.
|
|
raise
|
|
|
|
no_patch = self._is_no_patch(raw_output)
|
|
empty_response = all(
|
|
not line.strip() or line.strip().startswith("SUMMARY:")
|
|
for line in raw_output.splitlines()
|
|
)
|
|
# Only attributed NO_PATCH can request admission to leave remediation.
|
|
if remediation and no_patch:
|
|
# Keep the accepted episode baseline while discarding this empty candidate.
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
result: dict[str, object] = {"file": "", "reason": raw_output}
|
|
fault = "Return NO_PATCH with a JSON file and concrete causal reason, or produce a grounded repair."
|
|
# Parse the attribution supplied by the explicit control response.
|
|
try:
|
|
value = json.loads(self._no_patch_reason(raw_output))
|
|
except (ValueError, TypeError):
|
|
value = None
|
|
# Reject free prose and extra control fields at the model-output boundary.
|
|
if isinstance(value, dict) and set(value) == {"file", "reason"}:
|
|
result = value
|
|
# Retained history supplies the available restart destinations.
|
|
if provenance.current() is not None:
|
|
admission = DeadEndDecisionBuilder.for_direct_code_editor(
|
|
client=self.token_generator.clients[0],
|
|
file_manager=file_manager,
|
|
request=request,
|
|
last_result={"decision": type(self).__name__, **result},
|
|
completed_prompt=base_prompt,
|
|
).build()
|
|
admission()
|
|
fault = admission.reason or fault
|
|
patch_repair_feedback = (
|
|
"A no-edit response does not resolve the incoming failure. No restart was admitted. "
|
|
+ fault + "\nPrevious response:\n" + raw_output
|
|
)
|
|
self.attempts_since_acceptance += 1
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome="no patch",
|
|
decision_branch="retrying",
|
|
decision_reason=fault,
|
|
)
|
|
|
|
# Blank remediation output requests another attempt without attribution admission.
|
|
elif remediation and empty_response:
|
|
# Discard this empty candidate while retaining the accepted episode baseline.
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
fault = "Return NO_PATCH with a JSON file and concrete causal reason, or produce a grounded repair."
|
|
patch_repair_feedback = (
|
|
"A no-edit response does not resolve the incoming failure. No restart was admitted. "
|
|
+ fault + "\nPrevious response:\n" + raw_output
|
|
)
|
|
self.attempts_since_acceptance += 1
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome="no patch",
|
|
decision_branch="retrying",
|
|
decision_reason=fault,
|
|
)
|
|
|
|
# Ordinary no-patch responses terminate at the direct-editor protocol boundary.
|
|
elif no_patch:
|
|
no_patch_reason = self._no_patch_reason(raw_output)
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, outer_mementos)))
|
|
rejected_feedback = RepairEvidenceBuilder.compiler_rejection(
|
|
"\n".join(map(lambda error: error.message, current_errors)), rejected_diff,
|
|
)
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome="no patch",
|
|
decision_branch="stopped",
|
|
decision_reason="model returned NO_PATCH",
|
|
)
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
for memento in reversed(outer_mementos):
|
|
memento.rollback()
|
|
blackboard.clear_receipts()
|
|
request.editor_records = retained_records
|
|
return [], f"No patch was applied: {no_patch_reason}\n\n{rejected_feedback}", ""
|
|
|
|
# Treat an empty or summary-only response as an ordinary no-change result.
|
|
elif empty_response:
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, outer_mementos)))
|
|
rejected_feedback = RepairEvidenceBuilder.compiler_rejection(
|
|
"\n".join(map(lambda error: error.message, current_errors)), rejected_diff,
|
|
)
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome="no change",
|
|
decision_branch="stopped",
|
|
decision_reason="model response did not contain an applicable patch",
|
|
)
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
for memento in reversed(outer_mementos):
|
|
memento.rollback()
|
|
blackboard.clear_receipts()
|
|
request.editor_records = retained_records
|
|
return [], "\n\n".join(filter(None, (raw_output.strip(), rejected_feedback))), raw_output
|
|
|
|
else:
|
|
try:
|
|
patch_result = self.file_patcher.apply(
|
|
raw_output,
|
|
editable_files=editable,
|
|
readonly_files=readonly,
|
|
)
|
|
patch_summary = patch_result.summary
|
|
candidate_records = patch_result.records
|
|
patch_outcome = "applied"
|
|
except QuickRetryError as error:
|
|
candidate_records = error.records
|
|
# Retry an unapplied response; retain actual mutations as a partial candidate.
|
|
if not any(memento.diff for memento in mementos):
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="applying patch",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome="not applied",
|
|
unresolved_patch_issue_count=len(error.issues),
|
|
decision_branch="retrying",
|
|
decision_reason=error.decision_reason,
|
|
)
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
patch_repair_feedback = error.retry_feedback
|
|
self.attempts_since_acceptance += 1
|
|
unapplied_attempts += 1
|
|
else:
|
|
patch_summary = "\n".join(
|
|
[
|
|
"PARTIAL PATCH APPLIED",
|
|
error.summary or "",
|
|
"Unresolved patch issues:",
|
|
*[f"- {issue}" for issue in error.issues],
|
|
]
|
|
).strip()
|
|
patch_outcome = "partial"
|
|
unresolved_patch_issue_count = len(error.issues)
|
|
except DirectCodeEditFailed as error:
|
|
rejected_diff = "\n".join(filter(None, map(lambda memento: memento.diff, mementos)))
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="applying patch",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome="not applied",
|
|
unresolved_patch_issue_count=1,
|
|
decision_branch="retrying",
|
|
decision_reason="patch application did not apply",
|
|
)
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
patch_repair_feedback = RepairEvidenceBuilder.patch_rejection(
|
|
"PATCH APPLICATION DID NOT APPLY", str(error), raw_output
|
|
)
|
|
last_rejected_feedback = RepairEvidenceBuilder.compiler_rejection(str(error), rejected_diff)
|
|
self.attempts_since_acceptance += 1
|
|
unapplied_attempts += 1
|
|
except Exception:
|
|
# Preserve the failing candidate for the outer edit boundary's evidence capture.
|
|
raise
|
|
|
|
# Validate only candidates that survived patch application.
|
|
if patch_outcome in {"applied", "partial"}:
|
|
unapplied_attempts = 0
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="applying patch",
|
|
errors=current_errors,
|
|
blocking_errors=current_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome=patch_outcome,
|
|
unresolved_patch_issue_count=unresolved_patch_issue_count,
|
|
)
|
|
|
|
# Score the candidate against the currently committed repair baseline.
|
|
candidate_errors = self._compile_errors(
|
|
request,
|
|
compiler_factory,
|
|
outer_mementos,
|
|
editable,
|
|
blackboard=blackboard,
|
|
candidate_mementos=mementos,
|
|
previous_errors=current_errors,
|
|
)
|
|
candidate_blocking_errors = tuple(candidate_errors)
|
|
candidate_error_text = self._render_compiler_errors(candidate_blocking_errors)
|
|
candidate_error_count = len(candidate_blocking_errors)
|
|
candidate_phase = "targeted tests" if candidate_blocking_errors and all(
|
|
isinstance(error, TestRunCompileError) for error in candidate_blocking_errors
|
|
) else "candidate compile"
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase=candidate_phase,
|
|
errors=candidate_errors,
|
|
blocking_errors=candidate_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome=patch_outcome,
|
|
unresolved_patch_issue_count=unresolved_patch_issue_count,
|
|
)
|
|
# A clean compiler result cannot discharge remediation without a net change.
|
|
if (
|
|
not candidate_errors
|
|
and remediation
|
|
and not any(map(lambda memento: bool(memento.diff), outer_mementos))
|
|
):
|
|
blackboard.discard_candidate_results()
|
|
# An unchanged candidate retains no accepted runtime observation.
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
patch_repair_feedback = (
|
|
"The patch produced no net change and did not resolve the incoming failure. "
|
|
"Provide a grounded repair or NO_PATCH with another implicated file and a concrete causal reason."
|
|
)
|
|
self.attempts_since_acceptance += 1
|
|
# Only a clean, eligible candidate completes the outer edit episode.
|
|
elif not candidate_errors:
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=candidate_errors,
|
|
blocking_errors=candidate_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome=patch_outcome,
|
|
unresolved_patch_issue_count=unresolved_patch_issue_count,
|
|
decision_branch="accepted",
|
|
decision_reason="candidate has no blocking static or runtime semantic errors",
|
|
)
|
|
blackboard.commit_candidate_results()
|
|
for memento in reversed(mementos):
|
|
memento.commit()
|
|
changed_paths = [
|
|
memento.file_path for memento in outer_mementos if memento.diff
|
|
]
|
|
for memento in reversed(outer_mementos):
|
|
memento.commit()
|
|
request.editor_records = [*request.editor_records, *candidate_records]
|
|
self.attempts_since_acceptance = 0
|
|
self.error_permit = 1
|
|
return changed_paths, patch_summary, raw_output
|
|
|
|
else:
|
|
candidate_level = AiderFailureMode.attempt_level(
|
|
candidate_errors,
|
|
default=workflow_level,
|
|
)
|
|
workflow_level = max(workflow_level, candidate_level)
|
|
# Count only candidates that remain in the mechanical search.
|
|
if workflow_level is not AiderFailureMode.SEMANTIC:
|
|
mechanical_candidate_number += 1
|
|
# Commit static-clean runtime semantic observations regardless of test outcome.
|
|
if candidate_level is AiderFailureMode.SEMANTIC:
|
|
accepts = True
|
|
# Never anneal after this edit episode has reached semantic validation.
|
|
elif workflow_level is AiderFailureMode.SEMANTIC:
|
|
accepts = False
|
|
# Always establish the first mechanically invalid repair baseline.
|
|
elif current_error_count == 0:
|
|
accepts = True
|
|
# Admit a strict static-diagnostic reduction without annealing.
|
|
elif candidate_error_count < current_error_count:
|
|
accepts = True
|
|
# Let annealing decide every static candidate that did not improve.
|
|
else:
|
|
accepts = random.random() < max(
|
|
1 / (mechanical_candidate_number + 1), 0.10
|
|
)
|
|
# Commit an accepted search step; otherwise restore the prior baseline first.
|
|
if accepts:
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=candidate_errors,
|
|
blocking_errors=candidate_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome=patch_outcome,
|
|
unresolved_patch_issue_count=unresolved_patch_issue_count,
|
|
decision_branch="accepted baseline",
|
|
decision_reason="existing repair-loop acceptance branch selected this candidate",
|
|
)
|
|
stale_files = self._changed_class_metas(mementos)
|
|
blackboard.commit_candidate_results()
|
|
for memento in reversed(mementos):
|
|
memento.commit()
|
|
request.editor_records = [*request.editor_records, *candidate_records]
|
|
current_errors = candidate_errors
|
|
current_blocking_errors = candidate_blocking_errors
|
|
current_error_text = candidate_error_text
|
|
current_error_count = candidate_error_count
|
|
last_rejected_feedback = ""
|
|
patch_repair_feedback = ""
|
|
self.attempts_since_acceptance = 0
|
|
self.error_permit = 1
|
|
current_error_text = self._render_compiler_errors(current_blocking_errors)
|
|
presenter.backlog_guidance = tuple(request.guidance)
|
|
presenter.refine(current_error_text, stale_files=stale_files)
|
|
else:
|
|
self._publish_progress(
|
|
editable=editable,
|
|
repair_turn=repair_turn,
|
|
phase="decision",
|
|
errors=candidate_errors,
|
|
blocking_errors=candidate_blocking_errors,
|
|
previous_errors=current_errors,
|
|
patch_outcome=patch_outcome,
|
|
unresolved_patch_issue_count=unresolved_patch_issue_count,
|
|
decision_branch="rejected baseline",
|
|
decision_reason="existing repair-loop acceptance branch rejected this candidate",
|
|
)
|
|
rejected_diffs: list[str] = []
|
|
# Preserve only actual candidate mutations before Memento restores the baseline.
|
|
for memento in mementos:
|
|
candidate_diff = memento.diff.strip()
|
|
# Include only actual mutations in the rejected-candidate exhibit.
|
|
if candidate_diff:
|
|
rejected_diffs.append(candidate_diff)
|
|
rejected_candidate_diff = "\n".join(rejected_diffs)
|
|
blackboard.discard_candidate_results()
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
last_rejected_feedback = RepairEvidenceBuilder.compiler_rejection(
|
|
candidate_error_text, rejected_candidate_diff
|
|
)
|
|
patch_repair_feedback = ""
|
|
self.attempts_since_acceptance += 1
|
|
presenter.refine(current_error_text, stale_files=[])
|
|
except DeadEndError:
|
|
raise
|
|
except Exception:
|
|
# The public edit boundary captures evidence and owns complete episode rollback.
|
|
raise
|
|
|
|
def _compile_errors(
|
|
self,
|
|
request: AiderEditRequest,
|
|
compiler_factory: CompilerFactory,
|
|
mementos: list[Memento],
|
|
editable: list[Path],
|
|
*,
|
|
blackboard: EditedTestInventory,
|
|
candidate_mementos: Sequence[Memento] = (),
|
|
previous_errors: Sequence[CompileError] = (),
|
|
) -> list[CompileError]:
|
|
"""Run configured static and runtime semantic checks for one candidate."""
|
|
|
|
compiler = compiler_factory.build(
|
|
CompileAttempt(
|
|
request=request,
|
|
mementos=mementos,
|
|
frontier_provider=request.frontier_provider,
|
|
candidate_mementos=candidate_mementos,
|
|
previous_errors=previous_errors,
|
|
blackboard=blackboard,
|
|
)
|
|
)
|
|
return compiler.compile(editable)
|
|
|
|
def _publish_progress(
|
|
self,
|
|
*,
|
|
editable: Sequence[Path],
|
|
repair_turn: int,
|
|
phase: str,
|
|
errors: Sequence[CompileError],
|
|
blocking_errors: Sequence[CompileError],
|
|
previous_errors: Sequence[CompileError] | None,
|
|
visible_error_count: int | None = None,
|
|
patch_outcome: str | None = None,
|
|
unresolved_patch_issue_count: int = 0,
|
|
decision_branch: str | None = None,
|
|
decision_reason: str | None = None,
|
|
) -> None:
|
|
"""Publish one UI snapshot from static and runtime semantic results already observed."""
|
|
|
|
targeted_errors = tuple(
|
|
error for error in errors if isinstance(error, TestRunCompileError)
|
|
)
|
|
named_targeted_failures = tuple(
|
|
f"FAIL {error.test_name}"
|
|
for error in targeted_errors
|
|
if error.test_name
|
|
)
|
|
unattributed_targeted_execution_failure_count = sum(
|
|
1 for error in targeted_errors if not error.test_name
|
|
)
|
|
status = DirectCodeEditorStatusDTO(
|
|
editable_path=self._relative_or_absolute(editable[0]).as_posix(),
|
|
repair_turn=repair_turn,
|
|
phase=phase,
|
|
raw_compiler_error_count=len(errors),
|
|
blocking_compiler_error_count=len(blocking_errors),
|
|
# Awaiting-model events carry the count actually sent in that composed prompt.
|
|
model_visible_compiler_error_count=(
|
|
len(blocking_errors[:self.error_permit])
|
|
if visible_error_count is None else visible_error_count
|
|
),
|
|
static_type_check_error_count=sum(
|
|
1 for error in errors if isinstance(error, TypeCheckError)
|
|
),
|
|
unresolved_targeted_test_failure_count=len(targeted_errors),
|
|
named_targeted_failures=named_targeted_failures,
|
|
unattributed_targeted_execution_failure_count=(
|
|
unattributed_targeted_execution_failure_count
|
|
),
|
|
diagnostic_movement=_diagnostic_movement(previous_errors, errors),
|
|
patch_outcome=patch_outcome,
|
|
unresolved_patch_issue_count=unresolved_patch_issue_count,
|
|
decision_branch=decision_branch,
|
|
decision_reason=decision_reason,
|
|
)
|
|
publish_supreme_status(direct_code_editor=status)
|
|
progress_logger = current_logger() or logging.getLogger(__name__)
|
|
# Mirror every bounded UI snapshot in the existing runtime log stream.
|
|
progress_logger.info(
|
|
"DirectCodeEditor progress: target=%s repair_turn=%s phase=%s "
|
|
"blocking_remaining=%s raw=%s shown_to_model=%s static_type_check=%s "
|
|
"targeted_failures=%s named_targeted_failures=%s "
|
|
"unattributed_targeted_execution_failures=%s movement=%s "
|
|
"patch_outcome=%s unresolved_patch_issues=%s "
|
|
"decision_branch=%s decision_reason=%s",
|
|
status.editable_path,
|
|
status.repair_turn,
|
|
status.phase,
|
|
status.blocking_compiler_error_count,
|
|
status.raw_compiler_error_count,
|
|
status.model_visible_compiler_error_count,
|
|
status.static_type_check_error_count,
|
|
status.unresolved_targeted_test_failure_count,
|
|
status.named_targeted_failures,
|
|
status.unattributed_targeted_execution_failure_count,
|
|
status.diagnostic_movement,
|
|
status.patch_outcome,
|
|
status.unresolved_patch_issue_count,
|
|
status.decision_branch,
|
|
status.decision_reason,
|
|
)
|
|
|
|
def _render_compiler_errors(
|
|
self, errors: Sequence[CompileError], *, all_errors: bool = False,
|
|
) -> str:
|
|
"""Render current static or runtime semantic failures for model feedback."""
|
|
|
|
rendered: list[str] = []
|
|
# Oversized prompt expansion preserves every diagnostic; ordinary renders retain their prefix.
|
|
for error in errors if all_errors else errors[:self.error_permit]:
|
|
# Preserve Pyright's structured location evidence with its diagnostic.
|
|
if isinstance(error, TypeCheckError):
|
|
# Include only coordinates Pyright supplied, in canonical location order.
|
|
location = ":".join(
|
|
str(value)
|
|
for value in (error.path, error.line, error.column)
|
|
if value is not None
|
|
)
|
|
rendered.append(": ".join(filter(None, (location, error.message))))
|
|
continue
|
|
rendered.append(error.message)
|
|
return "\n".join(rendered).strip()
|
|
|
|
def _capture_mementos(
|
|
self,
|
|
editable: list[Path],
|
|
*,
|
|
stack: bool = True,
|
|
) -> list[Memento]:
|
|
"""Capture one nested ClassMeta snapshot for every editable managed file."""
|
|
|
|
if self.file_manager is None:
|
|
raise DirectCodeEditFailed("Compilation requires a bound FileManager.")
|
|
mementos: list[Memento] = []
|
|
captured: set[tuple[str, str]] = set()
|
|
try:
|
|
for path in editable:
|
|
metadata = self.file_manager.current_metadata(
|
|
path,
|
|
)
|
|
if metadata is None:
|
|
raise DirectCodeEditFailed(f"Compilation requires ClassMeta for {path}.")
|
|
identity = (metadata.content_type.value, metadata.location)
|
|
if identity in captured:
|
|
continue
|
|
captured.add(identity)
|
|
mementos.append(metadata.set_or_get_memento(self.file_manager, stack=stack))
|
|
except Exception:
|
|
for memento in reversed(mementos):
|
|
memento.rollback()
|
|
raise
|
|
return mementos
|
|
|
|
def _changed_class_metas(
|
|
self,
|
|
mementos: Sequence[Memento],
|
|
) -> list[ClassMeta]:
|
|
"""Return canonical metadata for inner mementos whose candidate changed source."""
|
|
|
|
if self.file_manager is None:
|
|
raise DirectCodeEditFailed("Compilation requires a bound FileManager.")
|
|
changed: list[ClassMeta] = []
|
|
seen: set[tuple[ContentType, str]] = set()
|
|
for memento in mementos:
|
|
if not memento.diff:
|
|
continue
|
|
class_meta = self.file_manager.current_metadata(memento.file_path)
|
|
if class_meta is None:
|
|
raise DirectCodeEditFailed(
|
|
f"Changed managed file is missing ClassMeta: {memento.file_path}."
|
|
)
|
|
identity = (class_meta.content_type, class_meta.location)
|
|
if identity in seen:
|
|
continue
|
|
seen.add(identity)
|
|
changed.append(class_meta)
|
|
return changed
|
|
|
|
def _is_no_patch(self, raw_output: str) -> bool:
|
|
"""Return whether one raw model response explicitly reports no further patch."""
|
|
|
|
for line in raw_output.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("SUMMARY:"):
|
|
continue
|
|
return stripped.startswith("NO_PATCH:")
|
|
return False
|
|
|
|
def _no_patch_reason(self, raw_output: str) -> str:
|
|
"""Extract direct-editor no-patch detail without forwarding its control token."""
|
|
|
|
# Read only the direct model-result channel that owns the NO_PATCH protocol.
|
|
for line in raw_output.splitlines():
|
|
stripped = line.strip()
|
|
# Skip non-protocol decoration while finding the direct control result.
|
|
if not stripped or stripped.startswith("SUMMARY:"):
|
|
continue
|
|
# Return ordinary diagnostic text after consuming the local control marker.
|
|
if stripped.startswith("NO_PATCH:"):
|
|
return stripped.partition(":")[2].strip() or "no concrete patch was supplied"
|
|
return "no concrete patch was supplied"
|
|
return "no concrete patch was supplied"
|
|
|
|
def _failure_summary(self, message: str) -> str:
|
|
"""Render one non-raising editor summary for failures that must not cross the editor boundary."""
|
|
|
|
return "\n".join(
|
|
[
|
|
"DIRECT SEARCH/REPLACE EDIT",
|
|
"Changed files:",
|
|
"",
|
|
"Patch summary:",
|
|
message,
|
|
]
|
|
).strip()
|