141 lines
7.1 KiB
Python
141 lines
7.1 KiB
Python
"""Malformed model output must not become a default or prematurely typed decision."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client import json_completion
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.cot_task_generator import CotTaskGenerator
|
|
from band_aider.judgement.current_cot_completed_check import CurrentCotCompletedCheck
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.global_relevant_check import GlobalRelevantCheck
|
|
from band_aider.judgement.intent_check import IntentCheck
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.prompt_localization import PromptLocalization
|
|
from tests.band_aider_loop_impl import _file_manager
|
|
from tests.support.completion_clients import ScriptedCompletionClient, ScriptedOpenAICompatibleClient
|
|
|
|
|
|
def test_completion_does_not_coerce_a_string_boolean_or_stop_after_model_circuits() -> None:
|
|
"""Twenty rejected boolean strings precede the only decision recorded by the owner."""
|
|
rejected = tuple(json.dumps({"completed": "false", "reason": f"rejected-{index}"}) for index in range(20))
|
|
client = ScriptedOpenAICompatibleClient((*rejected, '{"completed": false, "reason": "Work remains."}'))
|
|
check = CurrentCotCompletedCheck(changes=[], prompt="Repair lifecycle.", client=client,
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
|
|
assert check("Repair lifecycle.", []) is False
|
|
assert check.changes == [(check, False)]
|
|
assert len(client.prompts) == 21
|
|
assert client.prompts[-1].startswith(client.prompts[0])
|
|
assert rejected[-1] in client.prompts[-1]
|
|
assert rejected[-2] not in client.prompts[-1]
|
|
assert "boolean" in client.prompts[-1]
|
|
|
|
|
|
def test_intent_corrects_shape_enum_and_coherence_before_routing() -> None:
|
|
"""Valid JSON alone cannot authorize feature bootstrap."""
|
|
raw = ' \r\n== BEGIN MODEL TEXT ==\n { nope }\n== END MODEL TEXT ==\n\t'
|
|
client = ScriptedCompletionClient((
|
|
raw, '[]', '{"intent":"unknown","new_feature":false,"reason":"Wrong enum."}',
|
|
'{"intent":"existing_files","new_feature":true,"reason":"Contradiction."}',
|
|
'{"intent":"existing_files","new_feature":false,"reason":"Repair existing code."}',
|
|
))
|
|
check = IntentCheck(changes=[], prompt="Fix the existing caller.", client=client,
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
|
|
assert check() is False
|
|
assert len(client.prompts) == 5
|
|
assert raw in client.prompts[1]
|
|
assert raw not in client.prompts[2]
|
|
assert "must agree with intent" in client.prompts[-1]
|
|
assert "unknown" not in client.prompts[-1]
|
|
assert check.changes == [(check, False)]
|
|
|
|
|
|
def test_relevance_only_mutates_metadata_after_typed_admission() -> None:
|
|
"""A malformed positive cannot add evidence before the corrected negative response."""
|
|
client = ScriptedOpenAICompatibleClient((
|
|
'{"pull":"false","reason":"SHOULD NOT BE RECORDED","relevant_symbols":{"x":42}}',
|
|
'{"pull":false,"reason":"","relevant_symbols":{}}',
|
|
))
|
|
candidate = ClassMeta("src/x.py", "Current metadata", [], {})
|
|
check = GlobalRelevantCheck(changes=[], prompt="Repair y.", client=client,
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
|
|
assert check(candidate) is False
|
|
assert candidate.relevance == []
|
|
assert candidate.relevant_symbols == {}
|
|
assert len(client.prompts) == 2
|
|
|
|
|
|
def test_cot_generation_corrects_beyond_ten_without_changing_file_authority() -> None:
|
|
"""Repeated invented file identities cannot exhaust generation or enter typed records."""
|
|
candidate = ClassMeta("src/service.py", "Service", [], {})
|
|
wrong = '{"cot_tasks":[{"location":"src/invented.py","task":"Repair service."}]}'
|
|
right = '{"cot_tasks":[{"location":"src/service.py","task":"Repair service."}]}'
|
|
client = ScriptedCompletionClient((*([wrong] * 15), right))
|
|
generator = CotTaskGenerator(client, prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
|
|
records = generator.generate("Repair service.", [candidate])
|
|
|
|
assert len(client.prompts) == 16
|
|
assert len(records) == 1
|
|
assert records[0].root is candidate
|
|
assert records[0].location == candidate.location
|
|
assert "location is not in the current candidate page" in client.prompts[-1]
|
|
assert client.prompts[-1].count(wrong) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [ConnectionError("transport unavailable"), KeyboardInterrupt()])
|
|
def test_infrastructure_failure_or_cancellation_does_not_become_a_format_retry(failure: BaseException) -> None:
|
|
"""Only response validation can continue the correction loop."""
|
|
class FailingClient(ScriptedCompletionClient):
|
|
"""Return one malformed answer and then interrupt the correcting invocation."""
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Keep the failure on the provider boundary rather than the parser."""
|
|
if self.prompts:
|
|
self.prompts.append(prompt)
|
|
raise failure
|
|
return super().complete(prompt, logger)
|
|
|
|
client = FailingClient(('{"completed":"false"}',))
|
|
|
|
with pytest.raises(type(failure)):
|
|
json_completion.load_json_completion(client, "Return completed as a boolean.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(), references=[{"completed": False}])
|
|
|
|
assert len(client.prompts) == 2
|
|
|
|
|
|
@pytest.mark.prompt_localization
|
|
@pytest.mark.parametrize("stage", range(5))
|
|
def test_each_localization_judgement_corrects_more_than_twelve_bad_answers(stage: int) -> None:
|
|
"""Selector, producer, feedback and both reviewers retain their independent retry owners."""
|
|
manager = _file_manager(Path("/testbed"))
|
|
manager.prompt = "Keep construction inert.\nPreserve resolution."
|
|
required = {"relevant cot": [1, 2], "relevant code": [], "relevant stack": [],
|
|
"relevant backlog guidance": [], "relevant future guidance": []}
|
|
incomplete = [{"prompt": "Keep construction inert.", "relevant cot": [1]}]
|
|
complete = [{"prompt": "Keep construction inert. Preserve resolution.", "relevant cot": [1, 2]}]
|
|
feedback = {"summary": "The instruction is missing the independent resolution obligation."}
|
|
passed = {"outcome": "pass", "reason": "Complete and aligned.", "instruction": ""}
|
|
responses = list(map(json.dumps, (required, incomplete, feedback, complete, passed, passed)))
|
|
position = (0, 1, 2, 4, 5)[stage]
|
|
malformed = tuple(f"malformed-stage-{stage}-attempt-{index}" for index in range(15))
|
|
responses[position:position] = malformed
|
|
client = ScriptedCompletionClient(tuple(responses))
|
|
|
|
result = PromptLocalization(client=client, file_manager=manager,
|
|
request=AiderEditRequest(manager.prompt, [], []), owner="Inspection").localize(manager.prompt)
|
|
|
|
assert result == complete[0]["prompt"]
|
|
assert len(client.prompts) == 21
|
|
correction = client.prompts[position + 15]
|
|
assert malformed[-1] in correction
|
|
assert malformed[-2] not in correction
|