440 lines
16 KiB
Python
440 lines
16 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Focused contracts for grounded Alpha and Beta judgement responses."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from band_aider.context.validation_utils import (
|
|
SemanticValidationError,
|
|
ValidationError,
|
|
)
|
|
from band_aider.judgement.inspection_work_needed_decision import InspectionWorkNeededDecision
|
|
from band_aider.judgement.beta_contract_guidance_check import BetaContractGuidanceDecision
|
|
from band_aider.judgement.code_edit_or_leave_it import CodeEditOrLeaveIt
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.source_remediation_guidance_check import SourceRemediationGuidanceCheck
|
|
from tests.support.completion_clients import ScriptedOpenAICompatibleClient
|
|
from tests.band_aider_loop_impl import _file_manager
|
|
|
|
|
|
def _inspection_decision(response: object) -> InspectionWorkNeededDecision:
|
|
"""Bind inspection to a real focused module with explicit import and function evidence."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
focused = manager.require_src("src/foo.py")
|
|
manager.repo_exec.write_text(focused.resolve_path(manager), "from changed import Service\n\ndef foo():\n return 1\n")
|
|
return InspectionWorkNeededDecision(
|
|
client=ScriptedOpenAICompatibleClient((json.dumps(response),)),
|
|
file_manager=manager,
|
|
request=AiderEditRequest(authoritative_prompt="Inspect foo.", editable_files=[Path(focused.location)], readonly_files=[]),
|
|
inspection_prompt="Inspect foo.", focused_source=focused, readonly_sources=[], inspection_items=[],
|
|
)
|
|
|
|
|
|
def _beta_guidance_decision(
|
|
test_status: str,
|
|
response: object | None = None,
|
|
) -> BetaContractGuidanceDecision:
|
|
"""Return an isolated Beta decision at its validation boundary."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
focused = manager.require_test("tests/test_foo.py")
|
|
manager.repo_exec.write_text(focused.resolve_path(manager), "def test_foo():\n assert True\n")
|
|
responses = () if response is None else (json.dumps(response),)
|
|
return BetaContractGuidanceDecision(
|
|
client=ScriptedOpenAICompatibleClient(responses), file_manager=manager,
|
|
request=AiderEditRequest("Refine the selected test.", [Path(focused.location)], []),
|
|
focused_test=focused, original_cot="Preserve the COT.", accepted_source_diff="",
|
|
assignment_type="adapt", test_status=test_status,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("evidence_type", "semantic_divergence", "work_needed"),
|
|
[
|
|
(
|
|
"cot",
|
|
"none",
|
|
"1. Replace the COT-owned old_api call in src/root.py.\n"
|
|
"2. Verify the replacement satisfies the COT contract.",
|
|
),
|
|
(
|
|
"static_or_signature",
|
|
"none",
|
|
"1. Replace src/caller.py's removed old_api() call.\n"
|
|
"2. Verify the current signature is valid.",
|
|
),
|
|
(
|
|
"test_or_compile_failure",
|
|
"none",
|
|
"1. Repair the compiler-localized incompatible caller.\n"
|
|
"2. Re-run the cited check.",
|
|
),
|
|
(
|
|
"semantic_divergence",
|
|
"remove_use",
|
|
"1. Remove src/caller.py's obsolete lifecycle invocation.\n"
|
|
"2. Verify no old lifecycle use remains.",
|
|
),
|
|
(
|
|
"semantic_divergence",
|
|
"alter_use",
|
|
"1. Change src/caller.py to use managed entry sequencing.\n"
|
|
"2. Verify capture occurs at the managed lifecycle boundary.",
|
|
),
|
|
(
|
|
"semantic_divergence",
|
|
"add_behavior",
|
|
"1. Add the required rollback validation in src/caller.py.\n"
|
|
"2. Verify normal context exit preserves the new contract.",
|
|
),
|
|
],
|
|
)
|
|
def test_inspection_accepts_each_grounded_source_work_class(
|
|
evidence_type: str,
|
|
semantic_divergence: str,
|
|
work_needed: str,
|
|
) -> None:
|
|
"""Inspection accepts only a concrete source-owned divergence classification."""
|
|
|
|
payload = {
|
|
"work_required": True,
|
|
"evidence_type": evidence_type,
|
|
"semantic_divergence": semantic_divergence,
|
|
"reason": ["def foo():", "return 1"],
|
|
"work_needed": work_needed,
|
|
}
|
|
decision = _inspection_decision(payload)
|
|
_, validation = decision._decide()
|
|
|
|
assert validation.valid
|
|
assert decision.additional["work_required"] is True
|
|
assert decision.additional["evidence_type"] == evidence_type
|
|
assert decision.additional["semantic_divergence"] == semantic_divergence
|
|
assert decision.work_needed == work_needed
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{
|
|
"work_required": False,
|
|
"evidence_type": "none",
|
|
"semantic_divergence": "none",
|
|
"reason": ["from changed import Service"],
|
|
"work_needed": "",
|
|
},
|
|
{
|
|
"work_required": False,
|
|
"evidence_type": "none",
|
|
"semantic_divergence": "none",
|
|
"reason": ["def foo():", "return 1"],
|
|
"work_needed": "",
|
|
},
|
|
],
|
|
)
|
|
def test_inspection_closes_import_only_and_graph_only_items_without_work(
|
|
payload: dict[str, object],
|
|
) -> None:
|
|
"""Dependency proximity establishes inspection relevance, never mutation authority."""
|
|
|
|
decision = _inspection_decision(payload)
|
|
_, validation = decision._decide()
|
|
|
|
assert validation.valid
|
|
assert decision.additional["work_required"] is False
|
|
assert decision.additional["evidence_type"] == "none"
|
|
assert decision.additional["semantic_divergence"] == "none"
|
|
assert decision.work_needed == ""
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{
|
|
"work_required": False,
|
|
"evidence_type": "cot",
|
|
"semantic_divergence": "none",
|
|
"reason": ["return 1"],
|
|
"work_needed": "",
|
|
},
|
|
{
|
|
"work_required": True,
|
|
"evidence_type": "none",
|
|
"semantic_divergence": "none",
|
|
"reason": ["return 1"],
|
|
"work_needed": "invent work",
|
|
},
|
|
{
|
|
"work_required": True,
|
|
"evidence_type": "semantic_divergence",
|
|
"semantic_divergence": "none",
|
|
"reason": ["return 1"],
|
|
"work_needed": "invent work",
|
|
},
|
|
],
|
|
)
|
|
def test_inspection_rejects_ungrounded_or_internally_inconsistent_work(
|
|
payload: dict[str, object],
|
|
) -> None:
|
|
"""The no-work result is mandatory unless the positive evidence is coherent."""
|
|
|
|
decision = _inspection_decision(payload)
|
|
_, validation = decision._decide()
|
|
|
|
assert not validation.valid
|
|
assert all(isinstance(error, SemanticValidationError) for error in validation.errors)
|
|
assert decision.changes == []
|
|
|
|
|
|
def test_inspection_distinguishes_shape_error_metadata_from_semantic_errors() -> None:
|
|
"""A structurally incomplete response remains an ordinary validation error."""
|
|
|
|
decision = _inspection_decision({})
|
|
_, validation = decision._decide()
|
|
|
|
assert not validation.valid
|
|
assert all(type(error) is ValidationError for error in validation.errors)
|
|
assert decision.changes == []
|
|
|
|
|
|
def test_inspection_retries_invalid_response_until_a_valid_no_work_result() -> None:
|
|
"""Malformed inspection JSON stays local until the same judgement produces a legitimate cause."""
|
|
|
|
client = ScriptedOpenAICompatibleClient((
|
|
json.dumps({
|
|
"work_required": False,
|
|
"evidence_type": "cot",
|
|
"semantic_divergence": "none",
|
|
"reason": ["def foo():", "return 1"],
|
|
"work_needed": "",
|
|
}),
|
|
json.dumps({
|
|
"work_required": False,
|
|
"evidence_type": "none",
|
|
"semantic_divergence": "none",
|
|
"reason": ["def foo():", "return 1"],
|
|
"work_needed": "",
|
|
}),
|
|
))
|
|
manager = _file_manager(Path("/testbed"))
|
|
focused = manager.require_src("src/foo.py")
|
|
decision = InspectionWorkNeededDecision(
|
|
client=client,
|
|
file_manager=manager,
|
|
request=AiderEditRequest(
|
|
authoritative_prompt="Inspect foo without inventing work.",
|
|
editable_files=[Path(focused.location)],
|
|
readonly_files=[],
|
|
),
|
|
inspection_prompt="Inspect foo without inventing work.",
|
|
focused_source=focused,
|
|
readonly_sources=[],
|
|
inspection_items=[],
|
|
)
|
|
first_prompt = decision.prompt
|
|
|
|
assert decision() is False
|
|
prompts = client.prompts
|
|
assert len(prompts) == 2
|
|
assert prompts[0] == first_prompt
|
|
assert prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert (
|
|
"evidence_type: must be none when work_required is false."
|
|
in prompts[1]
|
|
)
|
|
assert decision.work_needed == ""
|
|
assert decision.additional["work_needed"] == ""
|
|
assert decision.reason == "def foo():\nreturn 1"
|
|
|
|
|
|
def test_source_remediation_retries_invalid_response_until_work_is_grounded() -> None:
|
|
"""Malformed remediation output cannot escape the owning judgement or end Alpha work."""
|
|
|
|
client = ScriptedOpenAICompatibleClient((
|
|
json.dumps({"work_needed": "", "reason": ["return 1"]}),
|
|
json.dumps({
|
|
"work_needed": "1. Repair the localized stale caller.\n2. Verify the targeted failure.",
|
|
"reason": ["return 1"],
|
|
}),
|
|
))
|
|
manager = _file_manager(Path("/testbed"))
|
|
focused = manager.require_src("src/foo.py")
|
|
decision = SourceRemediationGuidanceCheck(
|
|
client=client, file_manager=manager,
|
|
request=AiderEditRequest(authoritative_prompt="Repair foo.", editable_files=[Path(focused.location)], readonly_files=[]),
|
|
original_cot="Repair foo.", source=focused,
|
|
failure_evidence="Expected foo to return 2.", precedent_diffs="",
|
|
)
|
|
first_prompt = decision.prompt
|
|
|
|
assert decision() is True
|
|
prompts = client.prompts
|
|
assert len(prompts) == 2
|
|
assert prompts[0] == first_prompt
|
|
assert prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert "work_needed: must be a nonempty string." in prompts[1]
|
|
assert decision.additional["work_needed"].startswith("1. Repair")
|
|
|
|
|
|
def test_code_gate_requires_grounded_evidence_before_authorizing_an_edit() -> None:
|
|
"""A provisional claim cannot become mutation authority without local divergence evidence."""
|
|
|
|
decision = object.__new__(CodeEditOrLeaveIt)
|
|
|
|
parsed = decision._decision_from(
|
|
{
|
|
"needs_edit": False,
|
|
"evidence_type": "none",
|
|
"semantic_divergence": "none",
|
|
"reason": "The alleged old_api call is absent from the focused source.",
|
|
}
|
|
)
|
|
|
|
assert parsed.needs_edit is False
|
|
assert parsed.evidence_type == "none"
|
|
with pytest.raises(ValueError):
|
|
decision._decision_from(
|
|
{
|
|
"needs_edit": True,
|
|
"evidence_type": "none",
|
|
"semantic_divergence": "none",
|
|
"reason": "Prior guidance said it might be needed.",
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("test_status", "similarity", "action"),
|
|
[
|
|
("FAILED", 0.59, "delete"),
|
|
("FAILED", 0.60, "adapt"),
|
|
("FAILED", 0.91, "adapt"),
|
|
("REGRESSION STAR", 0.00, "add"),
|
|
],
|
|
)
|
|
def test_beta_contract_action_uses_the_semantic_similarity_threshold(
|
|
test_status: str,
|
|
similarity: float,
|
|
action: str,
|
|
) -> None:
|
|
"""Failed tests below 0.60 delete; retained regression intent adapts; missing coverage adds."""
|
|
|
|
payload = {
|
|
"asserted_contract": "The existing test asserts the old public lifecycle.",
|
|
"desired_contract": "The COT requires the managed lifecycle.",
|
|
"divergence": "The old public lifecycle is intentionally displaced.",
|
|
"contract_similarity": similarity,
|
|
"action": action,
|
|
"beta_instruction": "1. Update only the selected test function.\n2. Verify the desired contract.",
|
|
}
|
|
decision = _beta_guidance_decision(test_status, payload)
|
|
_, validation = decision._decide()
|
|
|
|
assert validation.valid
|
|
assert decision.additional["contract_similarity"] == similarity
|
|
assert decision.additional["action"] == action
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{
|
|
"asserted_contract": "old",
|
|
"desired_contract": "new",
|
|
"divergence": "different",
|
|
"contract_similarity": -0.01,
|
|
"action": "delete",
|
|
"beta_instruction": "remove it",
|
|
},
|
|
{
|
|
"asserted_contract": "old",
|
|
"desired_contract": "new",
|
|
"divergence": "different",
|
|
"contract_similarity": 0.61,
|
|
"action": "delete",
|
|
"beta_instruction": "remove it",
|
|
},
|
|
{
|
|
"asserted_contract": "old",
|
|
"desired_contract": "new",
|
|
"divergence": "different",
|
|
"contract_similarity": 1.01,
|
|
"action": "adapt",
|
|
"beta_instruction": "adapt it",
|
|
},
|
|
],
|
|
)
|
|
def test_beta_contract_guidance_rejects_invalid_similarity_or_failed_test_action(
|
|
payload: dict[str, object],
|
|
) -> None:
|
|
"""Malformed scores and action-threshold mismatches cannot become Beta instructions."""
|
|
|
|
decision = _beta_guidance_decision("FAILED", payload)
|
|
_, validation = decision._decide()
|
|
|
|
assert not validation.valid
|
|
assert all(isinstance(error, SemanticValidationError) for error in validation.errors)
|
|
assert decision.changes == []
|
|
|
|
|
|
def test_beta_contract_retry_handles_unmarshallable_numeric_semantics_without_throwing() -> None:
|
|
"""A shape-compatible huge integer becomes correction feedback, never an overflow exit."""
|
|
|
|
rejected = {
|
|
"asserted_contract": "old",
|
|
"desired_contract": "new",
|
|
"divergence": "different",
|
|
"contract_similarity": 10 ** 400,
|
|
"action": "delete",
|
|
"beta_instruction": "Remove the obsolete failed test.",
|
|
}
|
|
valid = {
|
|
**rejected,
|
|
"contract_similarity": 0.20,
|
|
}
|
|
client = ScriptedOpenAICompatibleClient(tuple(map(json.dumps, (rejected, valid))))
|
|
manager = _file_manager(Path("/testbed"))
|
|
focused = manager.require_test("tests/test_foo.py")
|
|
manager.repo_exec.write_text(focused.resolve_path(manager), "def test_foo():\n assert True\n")
|
|
decision = BetaContractGuidanceDecision(
|
|
client=client, file_manager=manager,
|
|
request=AiderEditRequest("Refine the selected test.", [Path(focused.location)], []),
|
|
focused_test=focused, original_cot="Preserve the COT.", accepted_source_diff="",
|
|
assignment_type="adapt", test_status="FAILED",
|
|
)
|
|
|
|
assert decision() is True
|
|
prompts = client.prompts
|
|
assert prompts[0] == decision.prompt
|
|
assert prompts[1].startswith(f"{decision.prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert "contract_similarity: must be between 0.0 and 1.0" in prompts[1]
|
|
|
|
|
|
def test_inspection_prompt_requires_work_evidence_and_compatible_guidance() -> None:
|
|
"""The actual inspection request receives work evidence rules and subordinate guidance requirements."""
|
|
|
|
decision = _inspection_decision({
|
|
"work_required": False, "evidence_type": "none", "semantic_divergence": "none",
|
|
"reason": ["def foo():"], "work_needed": "",
|
|
})
|
|
prompt = decision.prompt
|
|
|
|
assert "explicit COT ownership" in prompt
|
|
assert "concrete static/signature incompatibility" in prompt
|
|
assert "concrete compiler/test failure localization" in prompt
|
|
assert "concrete semantic divergence" in prompt
|
|
assert "Dependency proximity authorizes inspection only" in prompt
|
|
assert "Guidance requirements must be compatible with the authoritative task" in prompt
|
|
assert "diagnoses in guidance require verification" in prompt
|
|
assert "Satisfying the authoritative task is necessary but insufficient" in prompt
|
|
assert "REMOVE_USE" in prompt
|
|
assert "ALTER_USE" in prompt
|
|
assert "ADD_BEHAVIOR" in prompt
|