147 lines
5.1 KiB
Python
147 lines
5.1 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from types import TracebackType
|
|
from typing import Callable, TYPE_CHECKING
|
|
|
|
from band_aider.judgement.cached_file_patcher import CachedFilePatcher
|
|
from band_aider.judgement.code_edit_or_leave_it import is_code_edit_leave_summary
|
|
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult
|
|
|
|
if TYPE_CHECKING:
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.file_manager import FileManager
|
|
|
|
|
|
class CodeEditorCacheSession:
|
|
"""CodeEditorCacheSession activates cache behavior for one bounded edit scope."""
|
|
|
|
def __init__(self, editor: "CachedCodeEditor", topic: str) -> None:
|
|
"""Store the editor and caller-owned topic for this RAII scope."""
|
|
|
|
self._editor = editor
|
|
self.topic = topic
|
|
|
|
def __enter__(self) -> "CodeEditorCacheSession":
|
|
"""Activate the topic cache for this scope."""
|
|
|
|
self._editor._activate_cache(self.topic)
|
|
return self
|
|
|
|
def infer_from(self, files: list["ClassMeta"]) -> "CodeEditorCacheSession":
|
|
"""Set explicit files whose diffs are authoritative for cache training."""
|
|
|
|
self._editor.file_patcher.infer_from(files)
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc: BaseException | None,
|
|
traceback: TracebackType | None,
|
|
) -> None:
|
|
"""Deactivate cache behavior without discarding learned state."""
|
|
|
|
self._editor._deactivate_cache()
|
|
|
|
|
|
class CachedCodeEditor(CodeEditor):
|
|
"""CachedCodeEditor decorates a CodeEditor with opt-in topic-scoped cache behavior."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
inner: CodeEditor,
|
|
file_patcher: CachedFilePatcher,
|
|
) -> None:
|
|
"""Initialize the editor decorator with the wrapped editor and cached patcher."""
|
|
|
|
self._inner = inner
|
|
self.file_patcher = file_patcher
|
|
self._active_topic: str | None = None
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Return the wrapped editor."""
|
|
|
|
return self._inner
|
|
|
|
def enable_cache(self, topic: str) -> CodeEditorCacheSession:
|
|
"""Return a context manager that enables cache behavior for one caller-owned topic."""
|
|
|
|
return CodeEditorCacheSession(self, topic)
|
|
|
|
def edit(
|
|
self,
|
|
request: "AiderEditRequest",
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Apply cached inference when available, otherwise delegate to the wrapped editor."""
|
|
|
|
if self._active_topic is not None:
|
|
cached = self.file_patcher.apply_cached(request=request)
|
|
if cached.applied:
|
|
if logger is not None:
|
|
logger.info("CachedCodeEditor applied cached edit for %s.", self._active_topic)
|
|
request.editor_records = [*request.editor_records, *cached.records]
|
|
return CodeEditorResult(cached.changed_paths, cached.summary, list(request.editor_records))
|
|
|
|
result = self._inner.edit(request, logger=logger)
|
|
# Keep the direct gate's runtime LEAVE result out of cache history before its loop consumer handles it.
|
|
if self._active_topic is not None and not is_code_edit_leave_summary(result.summary):
|
|
self.file_patcher.record_code_editor_result(
|
|
request=request,
|
|
changed_paths=result.changed_paths,
|
|
summary=result.summary,
|
|
)
|
|
return result
|
|
|
|
def _activate_cache(self, topic: str) -> None:
|
|
"""Activate the cache layer for one topic."""
|
|
|
|
self._active_topic = topic
|
|
self.file_patcher.activate(
|
|
topic,
|
|
file_manager=getattr(self._inner, "file_manager", None),
|
|
)
|
|
|
|
def _deactivate_cache(self) -> None:
|
|
"""Return the cache layer to dormant pass-through mode."""
|
|
|
|
self._active_topic = None
|
|
self.file_patcher.deactivate()
|
|
|
|
def __getattr__(self, name: str) -> object:
|
|
"""Forward unknown attributes to the wrapped editor."""
|
|
|
|
return getattr(self._inner, name)
|
|
|
|
@property
|
|
def file_manager(self) -> "FileManager | None":
|
|
"""Return the wrapped editor's current file manager when available."""
|
|
|
|
return getattr(self._inner, "file_manager", None)
|
|
|
|
@file_manager.setter
|
|
def file_manager(self, value: "FileManager | None") -> None:
|
|
"""Forward explicit file-manager rebinding to the wrapped editor."""
|
|
|
|
setattr(self._inner, "file_manager", value)
|
|
|
|
@property
|
|
def new_files_callback(self) -> Callable[..., bool] | None:
|
|
"""Return the wrapped editor's new-file callback when available."""
|
|
|
|
return getattr(self._inner, "new_files_callback", None)
|
|
|
|
@new_files_callback.setter
|
|
def new_files_callback(self, value: Callable[..., bool] | None) -> None:
|
|
"""Forward explicit new-file callback rebinding to the wrapped editor."""
|
|
|
|
setattr(self._inner, "new_files_callback", value)
|