271 lines
12 KiB
Python
271 lines
12 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Response citations gate three decisions without changing their shared summary contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from band_aider.context.validation_utils import MechanicalValidationError
|
|
from band_aider.judgement.bootstrap_acceptance_check import BootstrapAcceptanceCheck
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.inspection_work_needed_decision import InspectionWorkNeededDecision
|
|
from band_aider.judgement.source_remediation_guidance_check import SourceRemediationGuidanceCheck
|
|
from band_aider.judgement.stabilization_decision import StabilizationDecision
|
|
from band_aider.judgement.python_ast_dependency_graph_maintainer import PythonAstDependencyGraphMaintainer
|
|
from tests.band_aider_loop_impl import TestFileManager as MemoryFileManager, _file_manager
|
|
from tests.support.completion_clients import ScriptedOpenAICompatibleClient
|
|
|
|
|
|
VERDICTS = ("bootstrap_pass", "bootstrap_fail", "inspection_work", "inspection_no_work", "remediation")
|
|
|
|
|
|
def _response(verdict: str, reason: object) -> dict[str, object]:
|
|
"""Describe explicit model responses for each admissible workflow outcome."""
|
|
|
|
return {
|
|
"bootstrap_pass": {"outcome": "pass", "instruction": "", "reason": reason},
|
|
"bootstrap_fail": {"outcome": "fail", "instruction": "Return 2 from foo; verify it.", "reason": reason},
|
|
"inspection_work": {
|
|
"work_required": True, "evidence_type": "cot", "semantic_divergence": "none",
|
|
"work_needed": "Return 2 from foo; verify it.", "reason": reason,
|
|
},
|
|
"inspection_no_work": {
|
|
"work_required": False, "evidence_type": "none", "semantic_divergence": "none",
|
|
"work_needed": "", "reason": reason,
|
|
},
|
|
"remediation": {"work_needed": "Restore foo returning 2; verify it.", "reason": reason},
|
|
}[verdict]
|
|
|
|
|
|
def _decision(
|
|
verdict: str, manager: MemoryFileManager, client: ScriptedOpenAICompatibleClient,
|
|
) -> StabilizationDecision:
|
|
"""Bind each real decision to the same small selected-source and readonly scope."""
|
|
|
|
source = manager.require_src("src/foo.py")
|
|
readonly = manager.require_test("tests/test_foo.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Return 2 from foo.", editable_files=[Path(source.location)],
|
|
readonly_files=[Path(readonly.location)], backlog_guidance=["Preserve the selected file scope."],
|
|
)
|
|
# Bootstrap judges the complete declared editable scope for either verdict.
|
|
if verdict.startswith("bootstrap"):
|
|
return BootstrapAcceptanceCheck(
|
|
client=client, file_manager=manager, edit_request=request, editable_sources=[source],
|
|
readonly_sources=[], readonly_tests=[readonly], previous_summary="Updated foo.",
|
|
baseline=manager.pristine_sources,
|
|
)
|
|
# Inspection judges work necessity for one focused source.
|
|
if verdict.startswith("inspection"):
|
|
return InspectionWorkNeededDecision(
|
|
client=client, file_manager=manager, request=request, inspection_prompt="Return 2 from foo.",
|
|
focused_source=source, readonly_sources=[readonly], inspection_items=[],
|
|
)
|
|
return SourceRemediationGuidanceCheck(
|
|
client=client, file_manager=manager, request=request, original_cot="Return 2 from foo.",
|
|
source=source, failure_evidence="foo returned 1", precedent_diffs="", readonly_sources=[readonly],
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
@pytest.mark.parametrize("blank", [False, True])
|
|
def test_deleted_citations_validate_before_guidance_or_verdict(verdict: str, blank: bool) -> None:
|
|
"""Every consuming decision verifies historical text, including the blank-file exception."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
path = manager.source_path("src/foo.py")
|
|
manager.dependency_graph_maintainer = PythonAstDependencyGraphMaintainer(repo_exec=manager.repo_exec)
|
|
manager.capture_frozen_once()
|
|
manager.repo_exec.write_text(path, "" if blank else "def foo():\n return 2\n")
|
|
invalid = _response(verdict, ["<DELETED> return invented()\n</DELETED>"])
|
|
valid = _response(verdict, ["<DELETED> return 1\n</DELETED>"])
|
|
decision = _decision(verdict, manager, ScriptedOpenAICompatibleClient((json.dumps(invalid), json.dumps(valid))))
|
|
|
|
response, validation = decision._decide()
|
|
assert not validation.valid
|
|
assert decision.outcome == ""
|
|
_, accepted = decision._decide(response, validation)
|
|
assert accepted.valid
|
|
assert decision.outcome
|
|
assert "<DELETED>" in decision.prompt
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
@pytest.mark.parametrize("reason", [
|
|
"return 1", None, [], [""], [7], ["return"], ["return 1\nreturn 2"],
|
|
["return 1", "invented()"], ["assert foo() == 1"],
|
|
])
|
|
def test_invalid_response_cannot_publish_a_verdict_or_work(verdict: str, reason: object) -> None:
|
|
"""One invalid member, malformed array, or readonly citation blocks the whole response."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
payload = _response(verdict, reason)
|
|
client = ScriptedOpenAICompatibleClient((json.dumps(payload),))
|
|
decision = _decision(verdict, manager, client)
|
|
|
|
response, validation = decision._decide()
|
|
|
|
assert response == payload
|
|
assert not validation.valid
|
|
assert "reason" in validation.stringify()
|
|
assert decision.outcome == ""
|
|
assert decision.reason == ""
|
|
assert decision.additional == {}
|
|
assert decision.changes == []
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
@pytest.mark.parametrize("reason", [["return 1"], ["def foo():", "return 1"]])
|
|
def test_response_array_is_checked_before_existing_summary_formatting(verdict: str, reason: list[str]) -> None:
|
|
"""Both verdicts remain reachable and the response retains all its citation members."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
payload = _response(verdict, reason)
|
|
decision = _decision(verdict, manager, ScriptedOpenAICompatibleClient((json.dumps(payload),)))
|
|
|
|
response, validation = decision._decide()
|
|
|
|
assert validation.valid
|
|
assert response == payload
|
|
assert decision.reason == "\n".join(reason)
|
|
assert decision.changes == [(decision, verdict not in {"bootstrap_fail", "inspection_no_work"})]
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
def test_missing_reason_returns_to_parent_correction(verdict: str) -> None:
|
|
"""Missing response evidence cannot become a default acceptance or an empty instruction."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
invalid = _response(verdict, ["return 1"])
|
|
del invalid["reason"]
|
|
client = ScriptedOpenAICompatibleClient((json.dumps(invalid), json.dumps(_response(verdict, ["return 1"]))))
|
|
decision = _decision(verdict, manager, client)
|
|
|
|
assert decision() is (verdict not in {"bootstrap_fail", "inspection_no_work"})
|
|
assert len(client.prompts) == 2
|
|
assert "REJECTED PREVIOUS RESPONSE" in client.prompts[1]
|
|
assert "reason" in client.prompts[1]
|
|
assert len(decision.changes) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
def test_citation_correction_retains_authority_within_local_allowance(verdict: str) -> None:
|
|
"""Local correction retains original authority and requests a complete replacement array."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
rejected = json.dumps(_response(verdict, ["return 1", "invented() "]))
|
|
accepted = json.dumps(_response(verdict, ["def foo():", "return 1"]))
|
|
client = ScriptedOpenAICompatibleClient((rejected,) * 2 + (accepted,))
|
|
decision = _decision(verdict, manager, client)
|
|
|
|
assert decision() is (verdict not in {"bootstrap_fail", "inspection_no_work"})
|
|
assert len(client.prompts) == 3
|
|
assert len(decision.changes) == 1
|
|
assert "all exact current" in client.prompts[0]
|
|
# Every correction preserves authority and replaces only the latest invalid response.
|
|
for prompt in client.prompts[1:]:
|
|
assert "Return 2 from foo." in prompt
|
|
assert prompt.count("BEGIN REJECTED PREVIOUS RESPONSE") == 1
|
|
assert "reason[1]" in prompt
|
|
assert "invented()" in prompt
|
|
assert "complete replacement reason array" in prompt
|
|
assert "Every item must pass." in prompt
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
def test_malformed_json_uses_counted_parent_attempts(verdict: str) -> None:
|
|
"""Malformed JSON is corrected by the parent without hidden adapter calls."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
malformed = '{"reason": ['
|
|
accepted = json.dumps(_response(verdict, ["return 1"]))
|
|
client = ScriptedOpenAICompatibleClient((malformed, malformed, accepted))
|
|
decision = _decision(verdict, manager, client)
|
|
|
|
assert decision() is (verdict not in {"bootstrap_fail", "inspection_no_work"})
|
|
assert len(client.prompts) == 3
|
|
assert "JSON RESPONSE REPAIR" not in client.prompts[1]
|
|
assert "REJECTED PREVIOUS RESPONSE" in client.prompts[-1]
|
|
assert (
|
|
"==== BEGIN REJECTED PREVIOUS RESPONSE ====\n"
|
|
+ malformed
|
|
+ "\n==== END REJECTED PREVIOUS RESPONSE ===="
|
|
) in client.prompts[-1]
|
|
assert "Return 2 from foo." in client.prompts[-1]
|
|
assert len(decision.changes) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", VERDICTS)
|
|
def test_same_citation_changes_validity_when_live_source_changes(verdict: str) -> None:
|
|
"""Counterfactual: identical response fails before its cited line exists and passes afterwards."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
payload = _response(verdict, ["return 2"])
|
|
client = ScriptedOpenAICompatibleClient((json.dumps(payload), json.dumps(payload)))
|
|
decision = _decision(verdict, manager, client)
|
|
|
|
rejected, validation = decision._decide()
|
|
assert not validation.valid
|
|
assert decision.changes == []
|
|
manager.repo_exec.write_text(manager.src_root / "foo.py", "def foo():\n return 2\n")
|
|
response, validation = decision._decide(rejected, validation)
|
|
|
|
assert validation.valid
|
|
assert response == rejected == payload
|
|
assert len(decision.changes) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", ["inspection_no_work", "remediation"])
|
|
@pytest.mark.parametrize("content", ["", " \n\t\n"])
|
|
def test_successfully_read_blank_source_allows_explanation_array(verdict: str, content: str) -> None:
|
|
"""Blank inspection can decline work; blank remediation still publishes repair guidance."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
manager.repo_exec.write_text(manager.src_root / "foo.py", content)
|
|
payload = _response(verdict, ["The selected file is blank."])
|
|
decision = _decision(verdict, manager, ScriptedOpenAICompatibleClient((json.dumps(payload),)))
|
|
|
|
assert decision() is (verdict == "remediation")
|
|
assert decision.additional["work_needed"] == payload["work_needed"]
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", ["inspection_no_work", "remediation"])
|
|
def test_blank_source_still_requires_every_explanation_to_be_nonblank(verdict: str) -> None:
|
|
"""The blank-file exception never discards an invalid member from its response array."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
manager.repo_exec.write_text(manager.src_root / "foo.py", "")
|
|
payload = _response(verdict, ["The selected file is blank.", " "])
|
|
decision = _decision(verdict, manager, ScriptedOpenAICompatibleClient((json.dumps(payload),)))
|
|
|
|
_, validation = decision._decide()
|
|
|
|
assert not validation.valid
|
|
assert "reason[1]" in validation.stringify()
|
|
assert decision.changes == []
|
|
|
|
|
|
@pytest.mark.parametrize("verdict", ["inspection_no_work", "remediation"])
|
|
@pytest.mark.parametrize("error", [FileNotFoundError("missing source"), PermissionError("unreadable source")])
|
|
def test_failed_source_read_cannot_claim_blank_file_exception(verdict: str, error: Exception) -> None:
|
|
"""A missing or unreadable selected file cannot admit prose as blank-file evidence."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
payload = _response(verdict, ["The selected file is blank."])
|
|
decision = _decision(verdict, manager, ScriptedOpenAICompatibleClient((json.dumps(payload),)))
|
|
|
|
with patch.object(manager.repo_exec, "read_text", side_effect=error):
|
|
response, validation = decision._decide()
|
|
|
|
assert response == payload
|
|
assert not validation.valid
|
|
assert any(isinstance(item, MechanicalValidationError) for item in validation.errors)
|
|
assert "blank-file exception does not apply" in validation.stringify()
|
|
assert decision.changes == []
|