1283 lines
56 KiB
Python
1283 lines
56 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Semantic tests for the direct edit-worthiness gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from band_aider.judgement.python_ast_dependency_graph_maintainer import PythonAstDependencyGraphMaintainer
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.context import provenance
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.provenance import DeadEndError, Direction, Phase, Provenance
|
|
from band_aider.context.working_backlog import (
|
|
BackLog,
|
|
BacklogLane,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
OutboundBacklogItem,
|
|
TestingBacklog as BetaWork,
|
|
)
|
|
from band_aider.judgement.code_edit_or_leave_it import (
|
|
CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN,
|
|
CodeEditLeaveSummary,
|
|
CodeEditOrLeaveIt,
|
|
consume_code_edit_leave_summary,
|
|
is_code_edit_leave_summary,
|
|
)
|
|
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult
|
|
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from tests.band_aider_loop_impl import FakeClient, TestFileManager, _alpha_context, _alpha_gate, _file_manager
|
|
|
|
|
|
class _JudgementClient(CompletionClient):
|
|
"""Return stationary direct-gate decisions and preserve rendered prompts."""
|
|
|
|
def __init__(self, responses: str | list[str]) -> None:
|
|
self._responses = [responses] if isinstance(responses, str) else list(responses)
|
|
self.prompts: list[str] = []
|
|
self.cot_prompts: list[str] = []
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Expose this client as a root completion client."""
|
|
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Describe the fixed non-routable test provider configuration."""
|
|
|
|
return (OpenAIConfig(model="test-model", api_base="http://model.invalid"),)
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the next configured response without contacting a provider."""
|
|
|
|
del logger
|
|
# These scenarios need no dependency cards beyond their explicit repository evidence.
|
|
if "DEPENDENCY REFERENCE SELECTION" in prompt:
|
|
return '{"references": []}'
|
|
# Select the declared prior COT by identity without generating replacement text.
|
|
if "CONTEXTUAL COT SELECTION" in prompt:
|
|
self.cot_prompts.append(prompt)
|
|
return "KEEP: COT-0000"
|
|
# The fixed gate result already contains the later attribution evidence.
|
|
if "BEGIN FUTURE GUIDANCE APPLICABILITY" in prompt.splitlines()[0]:
|
|
return '{"applicable": false, "reason": "The completed local result already carries the attribution."}'
|
|
self.prompts.append(prompt)
|
|
response = self._responses.pop(0) if len(self._responses) > 1 else self._responses[0]
|
|
try:
|
|
payload = json.loads(response)
|
|
except json.JSONDecodeError:
|
|
return response
|
|
if not isinstance(payload, dict) or not isinstance(payload.get("needs_edit"), bool):
|
|
return response
|
|
payload.setdefault(
|
|
"evidence_type",
|
|
"static_or_signature" if payload["needs_edit"] else "none",
|
|
)
|
|
payload.setdefault("semantic_divergence", "none")
|
|
return json.dumps(payload)
|
|
|
|
|
|
class _RecordingEditor(CodeEditor):
|
|
"""Nominal editor that records only the request delegated by the gate."""
|
|
|
|
def __init__(self) -> None:
|
|
self.requests: list[AiderEditRequest] = []
|
|
self.file_manager: FileManager | None = None
|
|
self.new_files_callback: Callable[..., bool] | None = None
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Expose this editor as the concrete end of the decorator chain."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Record the unchanged request and return a stationary edit result."""
|
|
|
|
del logger
|
|
self.requests.append(request)
|
|
return CodeEditorResult([Path("changed.py")], "inner summary", [])
|
|
|
|
|
|
@dataclass
|
|
class _MemoryProject:
|
|
"""Own a real FileManager and all source evidence inside one memory repository."""
|
|
|
|
file_manager: TestFileManager
|
|
|
|
@classmethod
|
|
def create(cls) -> "_MemoryProject":
|
|
"""Create a fresh project whose repository boundary cannot touch the host."""
|
|
|
|
return cls(_file_manager(Path("/testbed")))
|
|
|
|
def source(self, location: str, content: str) -> Path:
|
|
"""Add a source file under the managed source root and refresh metadata."""
|
|
|
|
return self._write(self.file_manager.src_root / location, content)
|
|
|
|
def test(self, location: str, content: str) -> Path:
|
|
"""Add a test file under the managed test root and refresh metadata."""
|
|
|
|
return self._write(self.file_manager.test_root / location, content)
|
|
|
|
def diff(self, path: Path, text: str) -> None:
|
|
"""Attach deterministic diff evidence through the real FileManager subtype."""
|
|
|
|
self.file_manager.set_diff(self.file_manager.repo_location(path), text)
|
|
|
|
def _write(self, path: Path, content: str) -> Path:
|
|
"""Write evidence only through the in-memory executor before resyncing it."""
|
|
|
|
self.file_manager.repo_exec.write_text(path, content)
|
|
self.file_manager.resync()
|
|
return path
|
|
|
|
|
|
def _gate(
|
|
project: _MemoryProject,
|
|
editor: _RecordingEditor,
|
|
client: _JudgementClient,
|
|
) -> CodeEditOrLeaveIt:
|
|
"""Bind the production direct gate to the scenario's real project collaborators."""
|
|
|
|
editor.file_manager = project.file_manager
|
|
return CodeEditOrLeaveIt(
|
|
inner=editor,
|
|
client=client,
|
|
repo_root=project.file_manager.repo_root,
|
|
repo_exec=project.file_manager.repo_exec,
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("resume", (False, True))
|
|
def test_progressed_duplicate_requires_validated_llm_leave(
|
|
resume: bool,
|
|
caplog: pytest.LogCaptureFixture,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Sibling history permits a judgment; malformed or ungrounded answers must still be corrected."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
path = source.resolve_path(manager)
|
|
failure = FailingTest(test, ["> old()", " return 1"], "AttributeError: old")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
root = graph.start(
|
|
Phase.ALPHA,
|
|
InspectionBacklog(ClassMeta("src/bar.py", "", [], {}), "root"),
|
|
)
|
|
earlier = graph.track_provenance(
|
|
root,
|
|
Phase.ALPHA,
|
|
FailedQAItem(source, "repair old call", failed_test=failure),
|
|
)
|
|
graph.track_provenance(
|
|
earlier,
|
|
Phase.BETA,
|
|
BetaWork(test, "verify repair", downstream_classes=[source]),
|
|
)
|
|
item = FailedQAItem(source, "repeated repair", failed_test=failure)
|
|
graph.track_provenance(root, Phase.ALPHA, item)
|
|
active_graph = (
|
|
Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=True)
|
|
if resume
|
|
else graph
|
|
)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient([
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": []}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": [1]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": null}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": [1]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "file": "src/foo.py"}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["invented()"]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": ["invented()"]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": ["return 1"]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": ["old()"]}',
|
|
])
|
|
gate = _gate(project, inner, client)
|
|
original = manager.repo_exec.read_text(path)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="repair old call", editable_files=[path], readonly_files=[],
|
|
failure_error_message="", failure_stack_trace=[],
|
|
is_remediation=True, remediation_originator_test=test,
|
|
)
|
|
recheck = Mock(return_value=False)
|
|
monkeypatch.setattr("band_aider.judgement.code_edit_or_leave_it.RemediationCheck", Mock(return_value=recheck))
|
|
|
|
with caplog.at_level(logging.INFO), provenance.bind(active_graph), provenance.item_scope(Phase.ALPHA, item):
|
|
_edit_result = gate.edit(request, logger=logging.getLogger("tests.duplicate-remediation"))
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert inner.requests == []
|
|
recheck.assert_called_once_with()
|
|
assert changed == []
|
|
assert type(summary) is CodeEditLeaveSummary
|
|
assert len(client.prompts) == 9
|
|
assert manager.repo_exec.read_text(path) == original
|
|
assert "nonempty array" in client.prompts[1]
|
|
assert "nonempty array" in client.prompts[2]
|
|
assert "negative_reason must be an array" in client.prompts[3]
|
|
assert "negative_reason must be an array" in client.prompts[4]
|
|
assert "file is not a valid judgement response field" in client.prompts[5]
|
|
assert "not found as a complete line in live code" in client.prompts[6]
|
|
assert "not a complete line in the supplied traceback" in client.prompts[7]
|
|
assert "still present in live code" in client.prompts[8]
|
|
assert "not a complete line in the supplied traceback" not in client.prompts[8]
|
|
assert earlier in client.prompts[0]
|
|
assert '"evidence_type": "already_fixed"' in client.prompts[0]
|
|
assert "nonempty array of exact current editable code lines" in client.prompts[0]
|
|
assert "negative_reason" in client.prompts[0]
|
|
assert "Historical traceback evidence may be stale" in client.prompts[0]
|
|
assert "does not establish that the current source is fixed" in client.prompts[0]
|
|
assert "admitted already_fixed" in caplog.text
|
|
assert CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN in summary
|
|
assert "already_fixed" not in summary
|
|
assert "Already fixed" not in summary
|
|
assert "return 1" not in summary
|
|
assert "old()" not in summary
|
|
|
|
|
|
@pytest.mark.parametrize("resumed", (False, True))
|
|
@pytest.mark.parametrize("pending_output", (False, True))
|
|
def test_admitted_leave_consumes_alpha_without_another_postflight(
|
|
resumed: bool, pending_output: bool, monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Validated local judgment uses ordinary LEAVE on fresh and resumed Alpha work."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source, test = manager.require_src("src/foo.py"), manager.require_test("tests/test_foo.py")
|
|
failure = FailingTest(test, [], "AttributeError: old")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
earlier = graph.start(Phase.ALPHA, FailedQAItem(source, "repair", failed_test=failure))
|
|
graph.track_provenance(
|
|
earlier,
|
|
Phase.BETA,
|
|
BetaWork(test, "verified repair", downstream_classes=[source]),
|
|
)
|
|
item = FailedQAItem(source, "repeat", failed_test=failure)
|
|
graph.start(Phase.ALPHA, item)
|
|
raw = BackLog(completion={})
|
|
raw.queue_remediation(item)
|
|
context = _alpha_context(manager, raw, source, test)
|
|
outbound = BetaWork(test, "earlier testing assignment", downstream_classes=[source])
|
|
# Model output already owned by Alpha before it encounters this duplicate.
|
|
if pending_output:
|
|
context.outbound_queue.append(OutboundBacklogItem(
|
|
lane=BacklogLane.Testing, dedupe_key=test.location,
|
|
value=outbound, lineage={"earlier": outbound},
|
|
))
|
|
# Only recovered active work should trigger the existing resume verification.
|
|
if resumed:
|
|
with provenance.bind(graph):
|
|
assert context.move_inbound_from_backlog() is not None
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient([
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["invented()"]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"]}',
|
|
])
|
|
loop = AlphaLoop(
|
|
gate=_alpha_gate(context),
|
|
context=context,
|
|
aider_worker=_gate(project, inner, client),
|
|
provenance=graph,
|
|
)
|
|
guidance = Mock(return_value="Check whether this source still needs the reported repair.")
|
|
monkeypatch.setattr(context.backlog, "remediation_work_needed_for", guidance)
|
|
postflight = Mock(return_value=[])
|
|
monkeypatch.setattr(context.backlog, "create_outbound_for", postflight)
|
|
monkeypatch.setattr(
|
|
"band_aider.loop.alpha_loop.AlphaEditAcceptanceDecision",
|
|
Mock(side_effect=AssertionError("admitted LEAVE must skip acceptance")),
|
|
)
|
|
|
|
with provenance.bind(graph):
|
|
loop._drain_source_backlog()
|
|
|
|
assert inner.requests == []
|
|
# Existing Alpha staging revisits a fresh claim once when an outbound batch is already pending.
|
|
passes = 2 if pending_output and not resumed else 1
|
|
assert len(client.prompts) == passes + 1
|
|
assert guidance.call_count == passes
|
|
postflight.assert_not_called()
|
|
assert is_code_edit_leave_summary(loop._source_summary)
|
|
assert "already_fixed" not in loop._source_summary
|
|
assert not context.inbound_queue
|
|
assert context.active_inbound is None
|
|
assert not raw.remediation
|
|
assert list(raw.testing) == ([outbound] if pending_output else [])
|
|
|
|
|
|
def test_revived_accepted_remediation_can_leave_after_correction() -> None:
|
|
"""A fresh revival may cite its prior handoff, but still must ground its no-edit judgment."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
failure = FailingTest(test, ["old()"], "AttributeError: old")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
original = graph.start(Phase.ALPHA, FailedQAItem(source, "repair", failed_test=failure))
|
|
beta = graph.track_provenance(original, Phase.BETA, BetaWork(test, "verify repair"))
|
|
_, revived = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK, current_message_id=beta,
|
|
message="Reconsider the source repair.", justification="The later test work reached a dead end.",
|
|
))
|
|
recovered = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=True)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient([
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["invented()"]}',
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": ["old()"]}',
|
|
])
|
|
gate = _gate(project, inner, client)
|
|
|
|
with provenance.bind(recovered), provenance.item_scope(Phase.ALPHA, revived):
|
|
_edit_result = gate.edit(AiderEditRequest(
|
|
authoritative_prompt="repair", editable_files=[source.resolve_path(manager)], readonly_files=[],
|
|
))
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert revived.message_id != original
|
|
assert original in client.prompts[0]
|
|
assert "already_fixed is available" in client.prompts[0]
|
|
assert "not found as a complete line in live code" in client.prompts[1]
|
|
assert len(client.prompts) == 2
|
|
assert inner.requests == []
|
|
assert changed == []
|
|
assert is_code_edit_leave_summary(summary)
|
|
assert "already_fixed" not in summary
|
|
|
|
|
|
@pytest.mark.parametrize("unavailable", ("self", "terminal", "other_file", "other_error", "other_trace"))
|
|
@pytest.mark.parametrize("stripped", (False, True))
|
|
def test_unavailable_already_fixed_retries_without_releasing_remediation(unavailable: str, stripped: bool) -> None:
|
|
"""Live citations cannot compensate for ineligible history, even when request evidence was stripped."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/foo.py")
|
|
failure = FailingTest(manager.require_test("tests/test_foo.py"), ["old()"], "failed")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
prior_source = ClassMeta("src/bar.py", "", [], {}) if unavailable == "other_file" else source
|
|
prior_failure = failure
|
|
# Change exactly one failure component for mismatch scenarios.
|
|
if unavailable == "other_error":
|
|
prior_failure = FailingTest(failure.test, ["old()"], "different")
|
|
elif unavailable == "other_trace":
|
|
prior_failure = FailingTest(failure.test, ["different()"], "failed")
|
|
prior = FailedQAItem(prior_source, "old repair", failed_test=prior_failure)
|
|
prior_id = graph.start(Phase.ALPHA, prior)
|
|
item = FailedQAItem(source, "repair", failed_test=failure)
|
|
# A progressed representation cannot establish that a second request with its own ID is duplicate work.
|
|
if unavailable == "self":
|
|
item = prior
|
|
# Record a successful phase handoff except in the terminal counterexample.
|
|
if unavailable != "terminal":
|
|
graph.track_provenance(
|
|
prior_id,
|
|
Phase.BETA,
|
|
BetaWork(failure.test, "verified repair", downstream_classes=[prior_source]),
|
|
)
|
|
# Give every non-self current obligation its own provenance identity.
|
|
if unavailable != "self":
|
|
graph.start(Phase.ALPHA, item)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient([
|
|
'{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"]}',
|
|
'{"needs_edit": false, "evidence_type": "none", "reason": "No work."}',
|
|
'{"needs_edit": true, "reason": "The authoritative task requires a current source change."}',
|
|
])
|
|
gate = _gate(project, inner, client)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="repair source", editable_files=[source.resolve_path(manager)], readonly_files=[],
|
|
failure_error_message="" if stripped else "failed",
|
|
failure_stack_trace=[] if stripped else ["old()"],
|
|
)
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
gate.edit(request)
|
|
|
|
assert len(client.prompts) == 3
|
|
assert "already_fixed is unavailable" in client.prompts[0]
|
|
assert "repeating already_fixed cannot satisfy this prerequisite" in client.prompts[1]
|
|
assert "Remediation no-edit judgement requires file" in client.prompts[2]
|
|
assert inner.requests == [request]
|
|
|
|
|
|
@pytest.mark.parametrize("obsolete_call", (False, True))
|
|
@pytest.mark.parametrize("deleted_citation", (False, True))
|
|
def test_progressed_history_does_not_override_current_code_or_llm_judgment(obsolete_call: bool, deleted_citation: bool) -> None:
|
|
"""The same history and answer leave fixed code but require correction when the obsolete call remains."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
project.source("foo.py", "def foo():\n old()\n return 1\n")
|
|
manager.dependency_graph_maintainer = PythonAstDependencyGraphMaintainer(repo_exec=manager.repo_exec)
|
|
manager.capture_frozen_once()
|
|
path = project.source("foo.py", "def foo():\n" + (" old()\n" if obsolete_call else "") + " return 1\n")
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
failure = FailingTest(test, ["old()"], "AttributeError: old")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
earlier = graph.start(Phase.ALPHA, FailedQAItem(source, "repair old call", failed_test=failure))
|
|
graph.track_provenance(earlier, Phase.BETA, BetaWork(test, "verify", downstream_classes=[source]))
|
|
item = FailedQAItem(source, "repeat repair", failed_test=failure)
|
|
graph.start(Phase.ALPHA, item)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient([
|
|
('{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1", "<DELETED> old()\\n</DELETED>"]}'
|
|
if deleted_citation else '{"needs_edit": false, "evidence_type": "already_fixed", "reason": ["return 1"], "negative_reason": ["old()"]}'),
|
|
'{"needs_edit": true, "evidence_type": "static_or_signature", "reason": "The focused code still calls old()."}',
|
|
])
|
|
gate = _gate(project, inner, client)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="repair old call", editable_files=[path], readonly_files=[],
|
|
failure_error_message=failure.error_message, failure_stack_trace=list(failure.stack_trace),
|
|
)
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
_edit_result = gate.edit(request)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert len(client.prompts) == (2 if obsolete_call else 1)
|
|
assert inner.requests == ([request] if obsolete_call else [])
|
|
assert changed == ([Path("changed.py")] if obsolete_call else [])
|
|
assert is_code_edit_leave_summary(summary) is not obsolete_call
|
|
# Live evidence must return the judgment to the model before it can choose an edit.
|
|
if obsolete_call:
|
|
assert "still present in live code" in client.prompts[1]
|
|
|
|
|
|
def test_progressed_history_still_allows_an_immediate_edit_judgment() -> None:
|
|
"""A matching prior handoff cannot overrule the LLM's current edit judgment."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
failure = FailingTest(test, ["return 1"], "assert 1 == 2")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
earlier = graph.start(Phase.ALPHA, FailedQAItem(source, "repair", failed_test=failure))
|
|
graph.track_provenance(earlier, Phase.BETA, BetaWork(test, "verify", downstream_classes=[source]))
|
|
item = FailedQAItem(source, "repair again", failed_test=failure)
|
|
graph.start(Phase.ALPHA, item)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient('{"needs_edit": true, "reason": "The current function still returns the wrong value."}')
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="repair", editable_files=[source.resolve_path(manager)], readonly_files=[],
|
|
failure_error_message=failure.error_message, failure_stack_trace=list(failure.stack_trace),
|
|
)
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
_edit_result = _gate(project, inner, client).edit(request)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert len(client.prompts) == 1
|
|
assert inner.requests == [request]
|
|
assert changed == [Path("changed.py")]
|
|
assert summary == "inner summary"
|
|
|
|
|
|
def test_code_edit_or_leave_it_automatically_passes_blank_editable_file() -> None:
|
|
"""A blank intended-new source bypasses judgement and reaches the inner editor."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/blank.py", "")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient('{"needs_edit": false, "reason": "unused"}')
|
|
gate = _gate(project, inner, client)
|
|
|
|
_edit_result = gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Update the implementation.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
)
|
|
)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert changed == [Path("changed.py")]
|
|
assert summary == "inner summary"
|
|
assert client.prompts == []
|
|
assert inner.requests[0].editable_files == [editable]
|
|
|
|
|
|
def test_code_edit_or_leave_it_skips_inner_and_leaves_files_untouched() -> None:
|
|
"""A grounded no-edit decision leaves the focused source unchanged in memory."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
'{"needs_edit": false, "reason": "Current code already satisfies the request."}'
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
|
|
_edit_result = gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Ensure value stays one.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
)
|
|
)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert changed == []
|
|
assert "CODE EDIT OR LEAVE IT" in summary
|
|
assert CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN in summary
|
|
assert "Current code already satisfies the request." in summary
|
|
assert inner.requests == []
|
|
assert project.file_manager.repo_exec.read_text(editable) == "value = 1\n"
|
|
prompt = client.prompts[0]
|
|
assert "ORIGINAL COT: IMMUTABLE STABILIZATION TARGET" in prompt
|
|
assert "ROLE AND EDIT BOUNDARIES" in prompt
|
|
assert "Ensure value stays one." in prompt
|
|
assert "PINNED CONTEXT" in prompt
|
|
assert "PROVISIONAL INSPECTION HYPOTHESIS" in prompt
|
|
assert "FAILURE EVIDENCE" in prompt
|
|
assert "READONLY FILE EVIDENCE" in prompt
|
|
assert '"readonly_files"' not in prompt
|
|
assert "repo-relative/path.py" not in prompt
|
|
|
|
|
|
def test_code_edit_or_leave_it_receives_selected_prior_cot_as_one_turn_context() -> None:
|
|
"""The no-edit gate sees exact earlier authority without changing its edit request."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
'{"needs_edit": false, "reason": "The current source preserves the removed API contract."}'
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
prior = CotTaskRecord(
|
|
"Remove Memento.clear.",
|
|
"src/memento.py",
|
|
ClassMeta("src/memento.py", "remove clear", [], {}),
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Inspect the focused source.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
previous_cot_records=(prior,),
|
|
)
|
|
|
|
_edit_result = gate.edit(request)
|
|
changed, _summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert changed == []
|
|
assert "Remove Memento.clear." in client.prompts[0]
|
|
assert "CURRENT COT — PRIMARY AUTHORITY AND EDIT SCOPE" in client.prompts[0]
|
|
assert request.authoritative_prompt == "Inspect the focused source."
|
|
assert len(client.cot_prompts) == 1
|
|
|
|
|
|
def test_code_edit_or_leave_it_treats_its_literal_leave_token_in_source_as_inert() -> None:
|
|
"""A leave token in source evidence cannot alter the runtime edit decision."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source(
|
|
"band_aider/judgement/code_edit_or_leave_it.py",
|
|
"CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN = \"[CODE_EDIT_OR_LEAVE_IT:LEAVE_IT]\"\n"
|
|
"def current_behavior():\n return 'edit me'\n",
|
|
)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
'{"needs_edit": true, "evidence_type": "cot", "semantic_divergence": "none", '
|
|
'"reason": "The COT explicitly assigns the focused source."}'
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
|
|
_edit_result = gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Implement the focused COT-owned behavior.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
)
|
|
)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert changed == [Path("changed.py")]
|
|
assert summary == "inner summary"
|
|
assert inner.requests
|
|
assert CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN in client.prompts[0]
|
|
|
|
|
|
def test_code_edit_or_leave_it_prompt_keeps_provisional_guidance_below_authority() -> None:
|
|
"""Immutable COT and local role authority precede provisional evidence in the prompt."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("focused.py", "def focused():\n return 1\n")
|
|
client = _JudgementClient('{"needs_edit": true, "reason": "unused"}')
|
|
gate = _gate(
|
|
project,
|
|
_RecordingEditor(),
|
|
client,
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="LOCAL ROLE AUTHORITY",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
backlog_guidance=["PROVISIONAL GUIDANCE"],
|
|
failure_error_message="CURRENT FAILURE",
|
|
)
|
|
|
|
gate.edit(request)
|
|
prompt = client.prompts[0]
|
|
|
|
assert prompt.index("ORIGINAL COT: IMMUTABLE STABILIZATION TARGET") < prompt.index(
|
|
"ROLE AND EDIT BOUNDARIES"
|
|
)
|
|
assert prompt.index("ROLE AND EDIT BOUNDARIES") < prompt.index(
|
|
"PROVISIONAL INSPECTION HYPOTHESIS"
|
|
)
|
|
assert prompt.index("PROVISIONAL INSPECTION HYPOTHESIS") < prompt.index(
|
|
"FAILURE EVIDENCE"
|
|
)
|
|
assert "Prior inspection guidance is a hypothesis." in prompt
|
|
assert "Dependency proximity does not authorize mutation." in prompt
|
|
assert "may" in prompt
|
|
assert "might" in prompt
|
|
assert "could" in prompt
|
|
|
|
|
|
def test_direct_runtime_leave_summary_is_a_typed_local_control_result() -> None:
|
|
"""Only the gate's known summary type may carry runtime leave-control state."""
|
|
|
|
summary = CodeEditLeaveSummary("CODE EDIT OR LEAVE IT\n" + CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN)
|
|
|
|
assert isinstance(summary, str)
|
|
assert type(summary) is CodeEditLeaveSummary
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"evidence_channel",
|
|
[
|
|
"editable source",
|
|
"readonly source",
|
|
"Git diff",
|
|
"failure output",
|
|
"historical log",
|
|
"test fixture",
|
|
],
|
|
)
|
|
def test_literal_leave_token_in_nonruntime_evidence_is_inert(evidence_channel: str) -> None:
|
|
"""Only a typed direct-gate result can activate or consume the leave protocol."""
|
|
|
|
evidence = f"{evidence_channel} quotes {CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN} as data"
|
|
|
|
assert is_code_edit_leave_summary(evidence) is False
|
|
assert consume_code_edit_leave_summary(evidence) == evidence
|
|
|
|
|
|
@pytest.mark.parametrize("direct_remediation", [False, True])
|
|
def test_alpha_no_edit_remediation_revives_beta(
|
|
direct_remediation: bool,
|
|
caplog: pytest.LogCaptureFixture,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A rejected source blame retains ordinary evidence through admission and revival."""
|
|
|
|
monkeypatch.setattr(
|
|
ContextualFileRenderer, "cheap_read",
|
|
Mock(return_value="REFERENCE CARD WITHOUT IMPLEMENTATION"),
|
|
)
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("band_aider/context/memento.py", "class Memento: pass\n")
|
|
caller = project.source(
|
|
"band_aider/loop/alpha_loop.py",
|
|
"source_file.set_or_get_memento(...).clear()\n",
|
|
)
|
|
inner = _RecordingEditor()
|
|
leave = '{"needs_edit": false, "file": "src/band_aider/loop/alpha_loop.py", "reason": "The stale clear call belongs to alpha_loop.py."}'
|
|
admit = json.dumps({
|
|
"dead_end": True,
|
|
"scope_fault": "wrong_editable_file",
|
|
"message": "Reconsider the source blame; the stale clear call is outside Memento.",
|
|
"justification": "alpha_loop.py calls clear, but Memento no longer exposes it.",
|
|
})
|
|
client = _JudgementClient([leave, admit, leave, admit] if direct_remediation else [leave, admit])
|
|
gate = _gate(project, inner, client)
|
|
manager = project.file_manager
|
|
manager.client = FakeClient(acceptance_reasons=("class Memento: pass",))
|
|
source = manager.require_src(manager.repo_location(editable))
|
|
readonly = manager.require_src(manager.repo_location(caller))
|
|
test = manager.require_test("tests/test_foo.py")
|
|
failure = FailingTest(test, ["alpha_loop.py: clear()"], "AttributeError: Memento has no clear")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
alpha_id = graph.start(Phase.ALPHA, InspectionBacklog(source, "Inspect Memento."))
|
|
beta = BetaWork(test, "Repair the failed lifecycle test.", downstream_classes=[source])
|
|
beta.record_failure(failure, [readonly])
|
|
beta_id = graph.track_provenance(alpha_id, Phase.BETA, beta)
|
|
parent_id = beta_id
|
|
# Direct-test routing inside Alpha creates the consecutive Alpha starts seen in the run.
|
|
if direct_remediation:
|
|
parent_id = graph.track_provenance(beta_id, Phase.ALPHA, FailedQAItem(
|
|
source, "Repair the source.", failed_test=failure,
|
|
))
|
|
item = FailedQAItem(
|
|
source, "Repair Memento.", failure_context_sources=[readonly], failed_test=failure,
|
|
)
|
|
item_id = graph.track_provenance(parent_id, Phase.ALPHA, item)
|
|
backlog = BackLog(completion={})
|
|
backlog.queue_remediation(item)
|
|
context = _alpha_context(manager, backlog, source, test)
|
|
loop = AlphaLoop(gate=_alpha_gate(context), context=context, aider_worker=gate, provenance=graph)
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
loop.run()
|
|
|
|
revived = backlog.peek_testing()
|
|
assert revived is not None
|
|
assert revived.message_id != beta_id
|
|
assert graph.messages[revived.message_id].parent_id == graph.messages[beta_id].parent_id
|
|
assert revived.failure_error_message == failure.error_message
|
|
restarts = list(filter(lambda edge: edge.direction is not None, graph.edges))
|
|
assert revived.future_guidance[0].origin_message_id == (restarts[0].to_id if direct_remediation else item_id)
|
|
assert "Reconsider the source blame; the stale clear call is outside Memento." in revived.future_guidance[0].message
|
|
assert "Attributed file: src/band_aider/loop/alpha_loop.py" in revived.future_guidance[0].message
|
|
assert revived.future_guidance[0].justification == (
|
|
"alpha_loop.py calls clear, but Memento no longer exposes it."
|
|
)
|
|
assert graph.edges[-1].direction is Direction.ROLLBACK
|
|
assert len(restarts) == (2 if direct_remediation else 1)
|
|
assert restarts[0].to_id != parent_id
|
|
assert graph.messages[restarts[0].to_id].parent_id == graph.messages[parent_id].parent_id
|
|
assert graph.edges[-1].to_id == revived.message_id
|
|
assert backlog.peek_remediation() is None
|
|
assert context.active_inbound is None
|
|
assert inner.requests == []
|
|
assert manager.repo_exec.read_text(editable) == "class Memento: pass\n"
|
|
assert "Alpha rollback revived beta phase work." in caplog.text
|
|
assert len(client.prompts) == (4 if direct_remediation else 2)
|
|
assert "Readonly files are evidence only" in client.prompts[0]
|
|
completed_prompt = client.prompts[1].split("==== BEGIN COMPLETED EDIT PROMPT ====\n", 1)[1].split(
|
|
"\n==== END COMPLETED EDIT PROMPT ====", 1
|
|
)[0]
|
|
assert '{"needs_edit": false' not in completed_prompt
|
|
assert '"needs_edit": false, "evidence_type": "none"' not in completed_prompt
|
|
assert '---REDACTED---' in completed_prompt
|
|
assert "Readonly files are evidence only" in completed_prompt
|
|
assert "class Memento: pass" in completed_prompt
|
|
assert "source_file.set_or_get_memento(...).clear()" in completed_prompt
|
|
assert "REFERENCE CARD WITHOUT IMPLEMENTATION" not in completed_prompt
|
|
assert '"needs_edit": false' in client.prompts[1]
|
|
assert '{"dead_end": true|false, "scope_fault":' in client.prompts[1]
|
|
|
|
|
|
@pytest.mark.parametrize("retained_beta", [False, True])
|
|
def test_no_edit_failure_without_admitted_restart_keeps_working(retained_beta: bool) -> None:
|
|
"""Absent history or declined admission cannot turn an unresolved failure into LEAVE."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("focused.py", "value = 1\n")
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/focused.py")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
item = FailedQAItem(source, "Fix the failing value assertion.")
|
|
responses = ['{"needs_edit": false, "file": "tests/test_foo.py", "reason": "The asserted contract belongs to the supplied test."}']
|
|
# Offer an actual restart only in the counterfactual where prior Beta history exists.
|
|
if retained_beta:
|
|
beta_id = graph.start(Phase.BETA, BetaWork(manager.require_test("tests/test_foo.py"), "Test value."))
|
|
graph.track_provenance(beta_id, Phase.ALPHA, item)
|
|
responses.append('{"dead_end": false, "scope_fault": "none", "message": "", "justification": ""}')
|
|
else:
|
|
graph.start(Phase.ALPHA, item)
|
|
responses.append('{"needs_edit": true, "reason": "The focused value is one, but the contract requires two."}')
|
|
client = _JudgementClient(responses)
|
|
inner = _RecordingEditor()
|
|
gate = _gate(project, inner, client)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Make value equal two.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
failure_raw_output="AssertionError: expected 2, got 1",
|
|
)
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
_edit_result = gate.edit(request)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert changed == [Path("changed.py")]
|
|
assert summary == "inner summary"
|
|
assert inner.requests == [request]
|
|
assert "No restart was admitted" in client.prompts[-1]
|
|
assert "AssertionError: expected 2, got 1" in client.prompts[-1]
|
|
assert not any(edge.direction is not None for edge in graph.edges)
|
|
|
|
|
|
def test_no_edit_reattribution_restarts_the_immediate_parent() -> None:
|
|
"""Remediation attribution returns one step with evidence for another local decision."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("redundant.py", "class Redundant: pass\n")
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/redundant.py")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
alpha_id = graph.start(Phase.ALPHA, InspectionBacklog(source, "Plan the source boundary."))
|
|
beta_id = graph.track_provenance(alpha_id, Phase.BETA, BetaWork(manager.require_test("tests/test_foo.py"), "Test source."))
|
|
item = FailedQAItem(source, "Repair source.")
|
|
graph.track_provenance(beta_id, Phase.ALPHA, item)
|
|
inner = _RecordingEditor()
|
|
gate = _gate(project, inner, _JudgementClient([
|
|
'{"needs_edit": false, "file": "tests/test_foo.py", "reason": "The test requires an obsolete redundant source boundary."}',
|
|
'{"dead_end": true, "scope_fault": "wrong_editable_file", "message": "Revisit the source plan.", "justification": "The existing source already owns this contract."}',
|
|
]))
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
with pytest.raises(DeadEndError) as raised:
|
|
gate.edit(AiderEditRequest(
|
|
authoritative_prompt="Repair the failed contract.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
failure_error_message="Duplicate source boundary",
|
|
))
|
|
|
|
phase, revived = graph.revive(raised.value)
|
|
assert raised.value.direction is Direction.ROLLBACK
|
|
assert phase is Phase.BETA
|
|
assert revived.message_id != beta_id
|
|
assert graph.messages[revived.message_id].parent_id == graph.messages[beta_id].parent_id
|
|
assert inner.requests == []
|
|
|
|
|
|
def test_no_edit_dead_end_provider_failure_preserves_alpha_claim(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Unavailable routing leaves remediation recoverable instead of reporting a clean no-op."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
beta_id = graph.start(Phase.BETA, BetaWork(test, "Test foo."))
|
|
item = FailedQAItem(source, "Fix foo.", failed_test=FailingTest(test, [], "AssertionError"))
|
|
graph.track_provenance(beta_id, Phase.ALPHA, item)
|
|
backlog = BackLog(completion={})
|
|
backlog.queue_remediation(item)
|
|
context = _alpha_context(manager, backlog, source, test)
|
|
inner = _RecordingEditor()
|
|
gate = _gate(project, inner, _JudgementClient('{"needs_edit": false, "file": "tests/test_foo.py", "reason": "The test asserts the obsolete contract."}'))
|
|
loop = AlphaLoop(gate=_alpha_gate(context), context=context, aider_worker=gate, provenance=graph)
|
|
|
|
def unavailable(client: CompletionClient, prompt: str, **kwargs: object) -> object:
|
|
"""Represent the provider becoming unavailable at dead-end admission."""
|
|
|
|
assert "DIRECT EDITOR DEAD-END ADMISSION" in prompt
|
|
raise ConnectionError("endpoint unreachable")
|
|
|
|
monkeypatch.setattr("band_aider.judgement.dead_end_decision.json_completion.load_json_completion", unavailable)
|
|
with pytest.raises(ConnectionError, match="endpoint unreachable"):
|
|
loop.run()
|
|
|
|
assert context.active_inbound is not None
|
|
assert context.active_inbound.value.message_id == item.message_id
|
|
assert backlog.peek_testing() is None
|
|
assert inner.requests == []
|
|
assert not any(edge.direction is not None for edge in graph.edges)
|
|
|
|
|
|
def test_code_edit_or_leave_it_delegates_the_original_request_for_an_edit() -> None:
|
|
"""An edit decision delegates the original complete request without scope changes."""
|
|
|
|
project = _MemoryProject.create()
|
|
second = project.source("pkg/second.py", "# second.py\n")
|
|
context = project.source("pkg/context.py", "# context.py\n")
|
|
noise = project.source("pkg/noise.py", "# noise.py\n")
|
|
project.diff(second, "diff --git a/second b/second\n-value = 1\n+value = 2")
|
|
project.diff(context, "diff --git a/context b/context\n+changed context")
|
|
project.diff(noise, "diff --git a/noise b/noise\n+changed noise")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
'{"needs_edit": true, "reason": "Only the second module needs the implementation."}'
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="This downstream prompt must not appear in the gate.",
|
|
editable_files=[second],
|
|
readonly_files=[context, noise],
|
|
)
|
|
|
|
_edit_result = gate.edit(request)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
assert changed == [Path("changed.py")]
|
|
assert summary == "inner summary"
|
|
delegated = inner.requests[0]
|
|
assert delegated is request
|
|
assert delegated.editable_files == [second]
|
|
assert delegated.readonly_files == [context, noise]
|
|
prompt = client.prompts[0]
|
|
assert "This downstream prompt must not appear in the gate." in prompt
|
|
assert "READONLY FILE EVIDENCE" in prompt
|
|
assert "changed context" in prompt
|
|
assert "changed noise" in prompt
|
|
assert "GIT DIFF" in prompt
|
|
|
|
|
|
def test_code_edit_or_leave_it_includes_work_needed_and_failed_test_evidence() -> None:
|
|
"""The gate prompt renders current work guidance and failed-test source evidence."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("foo.py", "def foo():\n return 1\n")
|
|
context = project.source("helper.py", "def helper():\n return 2\n")
|
|
failed_test = project.test("test_foo.py", "def test_foo():\n assert foo() == 2\n")
|
|
project.diff(editable, "diff --git a/foo b/foo\n- return 0\n+ return 1")
|
|
project.diff(context, "diff --git a/helper b/helper\n- return 1\n+ return 2")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
'{"needs_edit": true, "reason": "failure evidence shows more source work is still needed."}'
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
|
|
gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Repair foo so the failing test passes.",
|
|
editable_files=[editable],
|
|
readonly_files=[context, failed_test],
|
|
backlog_guidance=[
|
|
"src/foo.py: The work required involves modifying foo to return the tested value."
|
|
],
|
|
failure_error_message="AssertionError: expected 2 but got 1",
|
|
failure_stack_trace=["tests/test_foo.py:2: AssertionError: expected 2 but got 1"],
|
|
failure_raw_output="tests/test_foo.py:2: AssertionError: expected 2 but got 1\n",
|
|
)
|
|
)
|
|
|
|
prompt = client.prompts[0]
|
|
assert "Repair foo so the failing test passes." in prompt
|
|
assert "PROVISIONAL INSPECTION HYPOTHESIS" in prompt
|
|
assert "The work required involves modifying foo to return the tested value." in prompt
|
|
assert "FAILURE EVIDENCE" in prompt
|
|
assert "AssertionError: expected 2 but got 1" in prompt
|
|
assert "READONLY FILE EVIDENCE" in prompt
|
|
assert "FILE: tests/test_foo.py" in prompt
|
|
assert "assert foo() == 2" in prompt
|
|
assert "FILE: src/helper.py" in prompt
|
|
assert "FOCUSED EDITABLE FILES" in prompt
|
|
assert "FILE: src/foo.py" in prompt
|
|
|
|
|
|
def test_code_edit_or_leave_it_uses_renderer_and_keeps_failed_test_content() -> None:
|
|
"""Readonly test content remains present while editable diff stays selector-only."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("foo.py", "def foo():\n return 1\n")
|
|
failed_test = project.test("test_foo.py", "def test_foo():\n assert foo() == 2\n")
|
|
project.diff(editable, "editable diff must stay selector-only")
|
|
project.diff(failed_test, "failed-test upstream diff")
|
|
client = _JudgementClient('{"needs_edit": true, "reason": "done"}')
|
|
gate = _gate(
|
|
project,
|
|
_RecordingEditor(),
|
|
client,
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Repair foo.",
|
|
editable_files=[editable],
|
|
readonly_files=[failed_test],
|
|
failure_error_message="AssertionError",
|
|
)
|
|
|
|
gate.edit(request)
|
|
prompt = client.prompts[0]
|
|
|
|
readonly_section = prompt.split("READONLY FILE EVIDENCE", 1)[1].split(
|
|
"FOCUSED EDITABLE FILES", 1
|
|
)[0]
|
|
assert "FILE: tests/test_foo.py\n" in readonly_section
|
|
assert "def test_foo():\n assert foo() == 2\n" in readonly_section
|
|
assert "UPSTREAM GIT DIFF\nfailed-test upstream diff" in prompt
|
|
assert "editable diff must stay selector-only" not in prompt
|
|
|
|
|
|
def test_code_edit_or_leave_it_retries_until_invalid_selection_is_corrected() -> None:
|
|
"""An invalid response is rejected before the stationary valid response delegates work."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
upstream = project.source("pkg/upstream.py", "value = 2\n")
|
|
project.diff(upstream, "diff --git a/upstream b/upstream\n-value = 1\n+value = 2")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
[
|
|
'{"needs_edit": true, "editable_files": [{"path": "pkg/missing.py"}], "reason": "missing"}',
|
|
'{"needs_edit": true, "reason": "valid"}',
|
|
]
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
|
|
gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Update the implementation.",
|
|
editable_files=[editable],
|
|
readonly_files=[upstream],
|
|
)
|
|
)
|
|
|
|
assert len(client.prompts) == 2
|
|
delegated = inner.requests[0]
|
|
assert delegated.editable_files == [editable]
|
|
assert delegated.readonly_files == [upstream]
|
|
assert "editable_files is not a valid judgement response field." in client.prompts[1]
|
|
assert '"editable_files": [' in client.prompts[1]
|
|
assert '"path": "pkg/missing.py"' in client.prompts[1]
|
|
assert "REJECTED PREVIOUS RESPONSE" in client.prompts[1]
|
|
|
|
|
|
def test_initial_examples_explain_and_satisfy_every_classification() -> None:
|
|
"""The first request teaches complete combinations accepted by the edit gate."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
client = _JudgementClient('{"needs_edit": false, "reason": "Already correct."}')
|
|
gate = _gate(project, _RecordingEditor(), client)
|
|
gate.edit(AiderEditRequest(
|
|
authoritative_prompt="Keep the current value.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
))
|
|
prompt = client.prompts[0]
|
|
examples = prompt.split("VALID RESPONSE EXAMPLES\n", 1)[1].split("\n\n", 1)[0]
|
|
combinations = set()
|
|
# Each published example must survive the same admission used before real editing.
|
|
for line in examples.splitlines():
|
|
decision = gate._decision_from(json.loads(line))
|
|
combinations.add((decision.needs_edit, decision.evidence_type, decision.semantic_divergence))
|
|
assert decision.reason
|
|
assert combinations == {
|
|
(False, "none", "none"),
|
|
(True, "cot", "none"),
|
|
(True, "static_or_signature", "none"),
|
|
(True, "test_or_compile_failure", "none"),
|
|
(True, "semantic_divergence", "remove_use"),
|
|
(True, "semantic_divergence", "alter_use"),
|
|
(True, "semantic_divergence", "add_behavior"),
|
|
}
|
|
assert "static_or_signature with remove_use is invalid" in prompt
|
|
assert "It does not mean the edit has no behavioral effect." in prompt
|
|
|
|
|
|
def test_retry_replaces_old_rejection_with_malformed_model_text() -> None:
|
|
"""A JSON decoding failure retains the actual last response, not an earlier decision."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient([
|
|
'{"needs_edit": true, "evidence_type": "cot", "semantic_divergence": "remove_use", "reason": "OLDER_REJECTION"}',
|
|
"BROKEN_JSON",
|
|
"BROKEN_JSON",
|
|
"BROKEN_JSON",
|
|
'{"needs_edit": true, "reason": "Corrected."}',
|
|
])
|
|
gate = _gate(project, inner, client)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Update the current value.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
)
|
|
|
|
gate.edit(request)
|
|
|
|
retry = client.prompts[-1]
|
|
assert "BROKEN_JSON" in retry
|
|
assert "OLDER_REJECTION" not in retry
|
|
assert inner.requests == [request]
|
|
|
|
|
|
class _InterruptedJudgement(_JudgementClient):
|
|
"""Use ``_InterruptedJudgement(replies)`` to fail one call between rejected and accepted replies."""
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Expose a transport failure after a prior response to test evidence attribution."""
|
|
|
|
# Only the second judgement attempt fails; dependency classification stays available.
|
|
if "DEPENDENCY REFERENCE SELECTION" not in prompt and len(self.prompts) == 1:
|
|
self.prompts.append(prompt)
|
|
raise RuntimeError("Provider unavailable")
|
|
return super().complete(prompt, logger)
|
|
|
|
|
|
def test_failed_call_does_not_reuse_an_older_rejected_response() -> None:
|
|
"""An attempt without a response cannot attribute the earlier payload to its failure."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
client = _InterruptedJudgement([
|
|
'{"needs_edit": true, "evidence_type": "cot", "semantic_divergence": "remove_use", "reason": "OLDER_REJECTION"}',
|
|
'{"needs_edit": true, "reason": "Corrected."}',
|
|
])
|
|
inner = _RecordingEditor()
|
|
gate = _gate(project, inner, client)
|
|
gate.edit(AiderEditRequest(
|
|
authoritative_prompt="Update the current value.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
))
|
|
|
|
assert "Provider unavailable" in client.prompts[-1]
|
|
assert "No model response was received for this attempt." in client.prompts[-1]
|
|
assert "OLDER_REJECTION" not in client.prompts[-1]
|
|
assert len(inner.requests) == 1
|
|
|
|
|
|
def test_code_edit_or_leave_it_allows_scoped_readonly_citation_without_direct_graph_edge() -> None:
|
|
"""The declared readonly scope remains admissible without an inferred direct graph edge."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
direct = project.source("pkg/direct.py", "value = 1\n")
|
|
unrelated = project.source("pkg/unrelated.py", "value = 1\n")
|
|
project.diff(direct, "diff --git a/direct b/direct\n-value = 1\n+value = 2")
|
|
project.diff(unrelated, "diff --git a/unrelated b/unrelated\n-value = 1\n+value = 2")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient('{"needs_edit": true, "reason": "frontier scoped"}')
|
|
gate = _gate(project, inner, client)
|
|
|
|
gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Update the implementation.",
|
|
editable_files=[editable],
|
|
readonly_files=[direct, unrelated],
|
|
)
|
|
)
|
|
|
|
assert len(client.prompts) == 1
|
|
assert inner.requests[0].editable_files == [editable]
|
|
|
|
|
|
def test_code_edit_or_leave_it_leaves_comment_only_semantics_to_the_judgement_prompt() -> None:
|
|
"""Comment-only diff policy is prompt guidance, not brittle test-side keyword logic."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
upstream = project.source("pkg/upstream.py", "# new note\n")
|
|
project.diff(upstream, "diff --git a/upstream b/upstream\n-# old note\n+# new note")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient('{"needs_edit": true, "reason": "comment"}')
|
|
gate = _gate(project, inner, client)
|
|
|
|
gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Update the implementation.",
|
|
editable_files=[editable],
|
|
readonly_files=[upstream],
|
|
)
|
|
)
|
|
|
|
assert len(client.prompts) == 1
|
|
assert "Comment-only upstream diffs do not require downstream changes" in client.prompts[0]
|
|
assert inner.requests[0].editable_files == [editable]
|
|
|
|
|
|
def test_code_edit_or_leave_it_fails_open_when_needs_edit_is_missing() -> None:
|
|
"""A malformed no-op response cannot accidentally suppress a later valid decision."""
|
|
|
|
project = _MemoryProject.create()
|
|
editable = project.source("pkg/mod.py", "value = 1\n")
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient(
|
|
[
|
|
"{}",
|
|
'{"needs_edit": false, "reason": "No upstream diff requires this file."}',
|
|
]
|
|
)
|
|
gate = _gate(project, inner, client)
|
|
|
|
gate.edit(
|
|
AiderEditRequest(
|
|
authoritative_prompt="Update the implementation.",
|
|
editable_files=[editable],
|
|
readonly_files=[],
|
|
)
|
|
)
|
|
|
|
assert len(client.prompts) == 2
|
|
assert inner.requests == []
|
|
|
|
|
|
def test_code_edit_or_leave_it_forwards_runtime_bindings_to_inner() -> None:
|
|
"""Decorator rebinding keeps its real project and callback collaborators synchronized."""
|
|
|
|
project = _MemoryProject.create()
|
|
inner = _RecordingEditor()
|
|
gate = _gate(project, inner, _JudgementClient("{}"))
|
|
|
|
def reject_new_files(**files: object) -> bool:
|
|
"""Return a stationary callback answer without mutating the project."""
|
|
|
|
del files
|
|
return False
|
|
|
|
gate.file_manager = project.file_manager
|
|
gate.new_files_callback = reject_new_files
|
|
|
|
assert gate.file_manager is project.file_manager
|
|
assert gate.new_files_callback is reject_new_files
|
|
assert inner.file_manager is project.file_manager
|
|
assert inner.new_files_callback is reject_new_files
|
|
|
|
|
|
@pytest.mark.parametrize("remediation", (False, True))
|
|
def test_fresh_test_exit_requires_explicit_remediation(monkeypatch: pytest.MonkeyPatch, remediation: bool) -> None:
|
|
"""Populated failure fields alone never activate fresh verification in ordinary work."""
|
|
|
|
project = _MemoryProject.create()
|
|
manager = project.file_manager
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="work", editable_files=[source.resolve_path(manager)], readonly_files=[test.resolve_path(manager)],
|
|
is_remediation=remediation, remediation_originator_test=test,
|
|
failure_error_message="old error", failure_stack_trace=["tests/test_foo.py:2"],
|
|
)
|
|
check = Mock(return_value=True)
|
|
constructor = Mock(return_value=check)
|
|
monkeypatch.setattr("band_aider.judgement.code_edit_or_leave_it.RemediationCheck", constructor)
|
|
inner = _RecordingEditor()
|
|
client = _JudgementClient('{"needs_edit": true, "reason": "repair current code"}')
|
|
gate = _gate(project, inner, client)
|
|
|
|
_edit_result = gate.edit(request)
|
|
changed, summary = _edit_result.changed_paths, _edit_result.summary
|
|
|
|
# The business state controls test verification, regardless of populated optional evidence.
|
|
if remediation:
|
|
assert changed == []
|
|
assert isinstance(summary, CodeEditLeaveSummary)
|
|
assert client.prompts == []
|
|
assert inner.requests == []
|
|
check.assert_called_once_with()
|
|
else:
|
|
constructor.assert_not_called()
|
|
assert inner.requests == [request]
|
|
assert client.prompts
|