band-aider-pub/band_aider/judgement/cached_patch_generalization_check.py

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only

from __future__ import annotations

import json
from typing import Any

from band_aider.ai_client.completion_client import CompletionClient
from band_aider.ai_client.json_completion import load_json_completion
from band_aider.judgement.vibe_check import ReasonableVibeCheck
from band_aider.judgement.prompt_formatter import PromptFormatterBuilder, format_prompt


class CachedPatchGeneralizationCheck(ReasonableVibeCheck):
    """Judge whether recent patch samples are safely reusable as mechanical SEARCH/REPLACE blocks."""

    prompt_template = (
        "You are judging whether recent code-edit patches can be reused as cached mechanical "
        "SEARCH/REPLACE blocks.\n"
        "This judgement must be language agnostic.\n"
        "Return strict JSON only.\n"
        "Return cacheable only when every change in the samples is purely mechanical: name changes, "
        "module/path/reference changes, import target changes, or narrow literal/reference replacement.\n"
        "Return uncacheable if any sample includes boolean logic, control flow, workflow changes, "
        "state growth, state-shape changes, retry/timing/sequencing behavior, new behavior, or a mix "
        "of mechanical and stateful/control-flow changes.\n"
        "Each sample includes formal file evidence. changed_files are files changed by the successful "
        "patch. infer_files are caller-supplied authoritative evidence; when infer_files is non-empty, "
        "treat fallback_file_diffs as absent. fallback_file_diffs are editable/readonly file evidence "
        "used only when infer_files is empty. Each file evidence item has a role, file_context, "
        "change_kind, movement.from, movement.to, git_diff, and current_content. Use change_kind and "
        "movement to recognize created, deleted, moved, renamed, updated, unchanged, and unknown file "
        "lifecycle changes without assuming any programming language.\n"
        "File creation, deletion, movement, or rename evidence is cacheable only when the associated "
        "patches are still purely mechanical reference/module/path/name updates. If file lifecycle "
        "evidence is mixed with behavioral, stateful, or control-flow edits, return uncacheable.\n"
        "When cacheable, synthesize the narrowest SEARCH text and narrowest REPLACE text that can be "
        "applied safely to future editable files. Multiple blocks are allowed. Do not include prose.\n"
        'Return exactly this JSON shape:\n'
        '{{\n'
        '  "outcome": "cacheable|uncacheable",\n'
        '  "reason": "brief reason",\n'
        '  "additional": {{\n'
        '    "blocks": [{{"search": "text", "replace": "text"}}],\n'
        '    "contains_stateful_or_control_flow_change": false,\n'
        '    "contains_mixed_mechanical_and_stateful_change": false\n'
        "  }}\n"
        "}}\n\n"
        "The topic is an opaque cache key only. Do not infer file names, modules, hashes, or source "
        "meaning from the topic.\n\n"
        "Opaque topic:\n"
        "{topic}\n\n"
        "Last successful patch samples:\n"
        "{samples}"
    )

    def __init__(
        self,
        *,
        client: CompletionClient,
        prompt_formatter_builder: PromptFormatterBuilder,
        topic: str,
        samples: list[dict[str, object]],
    ) -> None:
        """Initialize one cacheability judgement prompt."""

        self._client = client
        self._prompt_formatter_builder = prompt_formatter_builder
        super().__init__(
            prompt=format_prompt(
                self._prompt_formatter_builder,
                "PATCH GENERALIZATION",
                self.prompt_template.format(
                    topic=topic,
                    samples=json.dumps(samples, indent=2),
                ),
            )
        )

    def __call__(self, *args: object, **kwargs: object) -> bool:
        """Correct malformed cache judgments before admitting cacheable or uncacheable."""

        try:
            data = load_json_completion(
                self._client,
                self.prompt,
                prompt_formatter_builder=self._prompt_formatter_builder,
                check_name=type(self).__name__,
                references=[{'outcome': '', 'reason': '', 'additional': {'blocks': [{'search': '', 'replace': ''}], 'contains_stateful_or_control_flow_change': False, 'contains_mixed_mechanical_and_stateful_change': False}}],
                validate=self.validate_response,
            )
        except Exception as error:
            data = {
                "outcome": "uncacheable",
                "reason": f"Patch generalization failed: {type(error).__name__}: {error}",
                "additional": {},
            }

        outcome = str(data.get("outcome") or "").strip().lower()
        reason = str(data.get("reason") or "").strip()
        additional = data.get("additional")
        if not isinstance(additional, dict):
            additional = {}

        blocks = self._normalized_blocks(additional.get("blocks"))
        stateful = additional.get("contains_stateful_or_control_flow_change") is not False
        mixed = additional.get("contains_mixed_mechanical_and_stateful_change") is not False
        if outcome != "cacheable" or stateful or mixed or not blocks:
            self.outcome = "uncacheable"
            self.reason = reason or "Recent patches were not purely mechanical."
            self.additional = {
                "blocks": [],
                "contains_stateful_or_control_flow_change": stateful,
                "contains_mixed_mechanical_and_stateful_change": mixed,
            }
            self.changes.append((self, False))
            return False

        self.outcome = "cacheable"
        self.reason = reason or "Recent patches were purely mechanical."
        self.additional = {
            "blocks": blocks,
            "contains_stateful_or_control_flow_change": False,
            "contains_mixed_mechanical_and_stateful_change": False,
        }
        self.changes.append((self, True))
        return True

    def validate_response(self, data: dict) -> None:
        """Require cacheability and the declared patch evidence to agree."""
        # Admit only the two declared cache decisions.
        if data["outcome"] not in {"cacheable", "uncacheable"}:
            raise ValueError("outcome must be cacheable or uncacheable.")
        additional = data["additional"]
        # A cacheable answer cannot contain stateful changes or unusable blocks.
        if data["outcome"] == "cacheable" and (
            additional["contains_stateful_or_control_flow_change"]
            or additional["contains_mixed_mechanical_and_stateful_change"]
            or not additional["blocks"]
        ):
            raise ValueError("cacheable requires nonempty blocks and both stateful/mixed flags false.")
        # An empty search cannot identify a reusable patch target.
        if any(not block["search"] for block in additional["blocks"]):
            raise ValueError("Every patch block requires nonempty search text; replace may be empty.")

    def _normalized_blocks(self, value: Any) -> list[dict[str, str]]:
        """Return validated SEARCH/REPLACE blocks from one JSON value."""

        if not isinstance(value, list):
            return []

        blocks: list[dict[str, str]] = []
        for item in value:
            if not isinstance(item, dict):
                continue
            search = item.get("search")
            replace = item.get("replace")
            if not isinstance(search, str) or not search:
                continue
            if not isinstance(replace, str):
                continue
            blocks.append(
                {
                    "search": search,
                    "replace": replace,
                }
            )
        return blocks