band-aider-pub/band_aider/judgement/single_task_closure_check.py
2026-10-17 11:57:37 -07:00

111 lines
4.2 KiB
Python

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only
from __future__ import annotations
import json
from dataclasses import dataclass
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.judgement.vibe_check import VibeCheck
from band_aider.judgement.prompt_formatter import PromptFormatterBuilder, format_prompt
@dataclass
class SingleTaskClosureCheck(VibeCheck):
"""SingleTaskClosureCheck decides whether one COT task is satisfied after a relevance queue drain."""
client: OpenAICompatibleClient
prompt_formatter_builder: PromptFormatterBuilder
reason: str = ""
remaining_work: str = ""
def __call__(
self,
task: str,
files: list[ClassMeta],
net_effects: dict[str, str],
) -> bool:
"""Return True when one COT task appears satisfied and False when another iteration is needed."""
data = load_json_completion(
self.client,
self._prompt(task, files, net_effects),
prompt_formatter_builder=self.prompt_formatter_builder,
check_name=type(self).__name__,
references=[{'satisfied': True, 'reason': '', 'remaining_work': ''}],
)
satisfied = bool(data.get("satisfied"))
self.reason = str(data.get("reason") or "")
self.remaining_work = str(data.get("remaining_work") or "")
self.changes.append((self, satisfied))
return satisfied
def _file_picture(self, candidate: ClassMeta) -> dict[str, object]:
"""Return a completion-evidence picture for one file."""
return {
"location": candidate.location,
"summary": candidate.summary,
"relevance": candidate.relevance,
"relevant_symbols": candidate.relevant_symbols,
}
def _prompt(
self,
task: str,
files: list[ClassMeta],
net_effects: dict[str, str],
) -> str:
"""Build the single-task closure prompt from task, changed files, and net effects."""
expected = {
"satisfied": True,
"reason": "one sentence explaining whether this COT task is satisfied",
"remaining_work": "what still has to be done if this task is not satisfied",
}
body = (
"Decide whether this stabilization goal is satisfied after this pass.\n"
"The original request is the immutable overarching goal; the current task, net effects, and file evidence are subordinate evidence.\n"
"Use the task, file locations, latest summaries, relevance notes, "
"and net file effects as evidence.\n\n"
"Return satisfied false if more relevant files should be discovered or another "
"global iteration is needed for this task.\n"
"Return satisfied true only when the evidence supports closing this task.\n\n"
"Return strict JSON only with this shape:\n"
f"{json.dumps(expected, indent=2)}\n\n"
"ORIGINAL COT: IMMUTABLE STABILIZATION TARGET\n"
f"{format_prompt(self.prompt_formatter_builder, 'ORIGINAL STABILIZATION GOAL', self.prompt, depth=2)}\n\n"
f"Current COT task:\n{task}\n\n"
f"Net file effects:\n{json.dumps(net_effects, indent=2)}\n\n"
"File evidence:\n"
f"{json.dumps([self._file_picture(file) for file in files], indent=2)}"
)
return format_prompt(
self.prompt_formatter_builder,
"TASK CLOSURE",
body,
escape=False,
)
@dataclass
class SingleTaskClosureCheckFactory:
"""SingleTaskClosureCheckFactory creates single-task closure checks for one original request."""
client: OpenAICompatibleClient
prompt_formatter_builder: PromptFormatterBuilder
def create(self, original_prompt: str) -> SingleTaskClosureCheck:
"""Create a single-task closure check for one original request."""
return SingleTaskClosureCheck(
changes=[],
prompt=original_prompt,
client=self.client,
prompt_formatter_builder=self.prompt_formatter_builder,
)