1093 lines
47 KiB
Python
1093 lines
47 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Input contracts for stabilization decisions using real memory-backed rendering."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.stabilization_escalation import StabilizationEscalation
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context import provenance
|
|
from band_aider.context.provenance import FutureGuidance, Phase, Provenance
|
|
from band_aider.context.validation_utils import (
|
|
SemanticValidationError,
|
|
ValidationLocation,
|
|
ValidationResult,
|
|
)
|
|
from band_aider.context.file_blame_check import FileBlameDecision
|
|
from band_aider.context.targeted_failure_owner_check import (
|
|
TargetedAlphaOrBetaRoutingDecision,
|
|
)
|
|
from band_aider.context.targeted_test_collection_failure_owner_check import (
|
|
TargetedCollectionAlphaOrBetaRoutingDecision,
|
|
)
|
|
from band_aider.context.working_backlog import (
|
|
ChangeType,
|
|
InspectionBacklog,
|
|
TestingBacklog,
|
|
)
|
|
from band_aider.judgement.alpha_blank_source_correction_check import (
|
|
AlphaBlankSourceCorrectionDecision,
|
|
)
|
|
from band_aider.judgement.alpha_edit_acceptance_check import (
|
|
AlphaEditAcceptanceDecision,
|
|
)
|
|
from band_aider.judgement.beta_contract_guidance_check import (
|
|
BetaContractGuidanceDecision,
|
|
)
|
|
from band_aider.judgement.bootstrap_acceptance_check import BootstrapAcceptanceCheck
|
|
from band_aider.judgement.edit_request import AiderEditRequest, AiderEditRequestMask, AiderFailureMode
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.inspection_work_needed_decision import InspectionWorkNeededDecision
|
|
from band_aider.judgement.file_parser import FileParser
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder, PromptFormatter
|
|
from band_aider.judgement.source_preflight_decision import AlphaSourcePreflightDecision, BootstrapSourcePreflightDecision
|
|
from band_aider.judgement.source_remediation_guidance_check import (
|
|
SourceRemediationGuidanceCheck,
|
|
)
|
|
from band_aider.judgement.stabilization_decision import (
|
|
FutureGuidanceApplicability,
|
|
FutureGuidanceRelocalization,
|
|
StabilizationDecision,
|
|
)
|
|
from tests.support.completion_clients import (
|
|
ScriptedCompletionClient,
|
|
ScriptedOpenAICompatibleClient,
|
|
StaticCompletionClient,
|
|
StaticOpenAICompatibleClient,
|
|
)
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
def _file_manager(
|
|
*,
|
|
sources: tuple[str, ...] = (),
|
|
tests: tuple[str, ...] = (),
|
|
prompt: str = "FILE_MANAGER_PROMPT_SENTINEL",
|
|
) -> FileManager:
|
|
"""Build one actual FileManager over explicit source and test text in memory."""
|
|
|
|
repo_root = Path("/testbed")
|
|
repo_exec = MemoryRepoExec(repo_root)
|
|
for location in sources:
|
|
repo_exec.write_text(repo_root / location, f"# {location}\npass\n")
|
|
for location in tests:
|
|
repo_exec.write_text(
|
|
repo_root / location,
|
|
f"# {location}\ndef test_fixture() -> None:\n assert True\n",
|
|
)
|
|
prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
parser_client = StaticOpenAICompatibleClient(json.dumps({
|
|
"location": "",
|
|
"summary": "memory fixture",
|
|
"relevance": ["fixture relevance"],
|
|
"irrelevance": [],
|
|
"relevant_symbols": {},
|
|
}))
|
|
return FileManager(
|
|
parser=FileParser(parser_client, prompt_formatter_builder, repo_exec),
|
|
client=parser_client,
|
|
repo_root=repo_root,
|
|
src_root=repo_root / "src",
|
|
test_root=repo_root / "tests",
|
|
actor="test",
|
|
prompt_formatter_builder=prompt_formatter_builder,
|
|
repo_exec=repo_exec,
|
|
prompt=prompt,
|
|
)
|
|
|
|
|
|
def _source(manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return one discovered source with an explicit non-null assertion boundary."""
|
|
|
|
source = manager.current_src(location)
|
|
assert source is not None
|
|
return source
|
|
|
|
|
|
def _test(manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return one discovered test with an explicit non-null assertion boundary."""
|
|
|
|
test = manager.current_test(location)
|
|
assert test is not None
|
|
return test
|
|
|
|
|
|
class _ShiftedBuilder(MarkdownPromptFormatterBuilder):
|
|
"""Expose literal-depth assumptions with ``manager.prompt_formatter_builder = _ShiftedBuilder()``."""
|
|
|
|
def generate(
|
|
self,
|
|
name: str,
|
|
*,
|
|
depth: int = 4,
|
|
escape: bool = False,
|
|
neutralize_json: bool = False,
|
|
) -> PromptFormatter:
|
|
"""Preserve the formatter contract with a different initial section depth."""
|
|
|
|
return super().generate(
|
|
name,
|
|
depth=depth,
|
|
escape=escape,
|
|
neutralize_json=neutralize_json,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("responses", (
|
|
(' \r\n```json\n[]\n```\t\n',),
|
|
(' \r\n```json\n[}\n```\t\n', '```json\n[]\n```'),
|
|
(' \r\n```json\n{"requested": []}\n```\t\n', '```json\n[]\n```'),
|
|
))
|
|
def test_fenced_preflight_preserves_retry_evidence(responses: tuple[str, ...]) -> None:
|
|
"""Fence cleanup admits valid preflight and preserves raw evidence for real rejections."""
|
|
|
|
manager = _file_manager(sources=("src/candidate.py",))
|
|
client = ScriptedOpenAICompatibleClient(responses)
|
|
decision = AlphaSourcePreflightDecision(
|
|
client=client, file_manager=manager, source=_source(manager, "src/candidate.py"),
|
|
request=AiderEditRequest("Inspect candidate.", [Path("src/candidate.py")], []),
|
|
work_needed="Inspect candidate.",
|
|
)
|
|
|
|
assert decision() is True
|
|
assert len(client.prompts) == len(responses)
|
|
assert decision.escalation.last_response == responses[-1]
|
|
# Only malformed JSON and invalid shapes warrant a correction prompt.
|
|
if len(responses) > 1:
|
|
assert (
|
|
"==== BEGIN REJECTED PREVIOUS RESPONSE ====\n"
|
|
+ responses[0]
|
|
+ "\n==== END REJECTED PREVIOUS RESPONSE ===="
|
|
) in client.prompts[1]
|
|
assert "RESPONSE VALIDATION ERROR" in client.prompts[1]
|
|
|
|
|
|
def test_preflight_inherits_depth_and_preserves_multiline_evidence() -> None:
|
|
"""Preflight nests current code and protects embedded section boundaries."""
|
|
|
|
manager = _file_manager(sources=("src/candidate.py",))
|
|
manager.prompt_formatter_builder = _ShiftedBuilder()
|
|
source = _source(manager, "src/candidate.py")
|
|
content = "value = 1\n== END ALPHA SOURCE PREFLIGHT ==\nvalue = 2\n"
|
|
manager.repo_exec.write_text(manager.source_path(source.location), content)
|
|
client = ScriptedCompletionClient(('[]',))
|
|
decision = AlphaSourcePreflightDecision(
|
|
client=client,
|
|
file_manager=manager,
|
|
source=source,
|
|
request=AiderEditRequest("Inspect candidate.", [Path(source.location)], []),
|
|
work_needed="First instruction\nSecond instruction",
|
|
)
|
|
|
|
assert decision() is True
|
|
assert client.prompts == [decision.prompt]
|
|
assert decision.prompt.startswith("======== BEGIN ALPHA SOURCE PREFLIGHT ========\n")
|
|
assert "========== BEGIN CURRENT FOCUSED SOURCE ==========\n" in decision.prompt
|
|
assert "============ BEGIN WORK NEEDED ============\nFirst instruction\nSecond instruction\n" in decision.prompt
|
|
assert "value = 1\n\\== END ALPHA SOURCE PREFLIGHT ==\nvalue = 2\n" in decision.prompt
|
|
assert manager.repo_exec.read_text(manager.source_path(source.location)) == content
|
|
|
|
|
|
def test_acceptance_inherits_depth_for_repeated_source_mementos() -> None:
|
|
"""Each exact edit baseline remains attached to its source through native nesting."""
|
|
|
|
manager = _file_manager(sources=("src/first.py", "src/second.py"))
|
|
manager.prompt_formatter_builder = _ShiftedBuilder()
|
|
first = _source(manager, "src/first.py")
|
|
second = _source(manager, "src/second.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Keep both source obligations.",
|
|
editable_files=[manager.source_path(first.location), manager.source_path(second.location)],
|
|
readonly_files=[],
|
|
)
|
|
client = ScriptedCompletionClient(('{"outcome": "pass", "reason": ["pass"], "instruction": ""}',))
|
|
decision = AlphaEditAcceptanceDecision(
|
|
client=client,
|
|
file_manager=manager,
|
|
request=request,
|
|
editable_sources=[first, second],
|
|
readonly_sources=[],
|
|
readonly_tests=[],
|
|
previous_summary="No edit required.",
|
|
)
|
|
|
|
assert decision() is True
|
|
assert decision.prompt.startswith("======== BEGIN ALPHA EDIT ACCEPTANCE ========\n")
|
|
assert "========== BEGIN EDITABLE SOURCE MEMENTOS ==========\n" in decision.prompt
|
|
assert decision.prompt.count("============ BEGIN SOURCE MEMENTO ============\n") == 2
|
|
first_record, second_record = decision.prompt.split("============ BEGIN SOURCE MEMENTO ============\n")[1:]
|
|
assert "============== BEGIN LOCATION ==============\nsrc/first.py\n" in first_record
|
|
assert "============== BEGIN PRE ==============\n# src/first.py\npass\n" in first_record
|
|
assert "src/second.py" not in first_record
|
|
assert "============== BEGIN LOCATION ==============\nsrc/second.py\n" in second_record
|
|
assert "============== BEGIN PRE ==============\n# src/second.py\npass\n" in second_record
|
|
|
|
|
|
def test_future_guidance_inherits_depth_and_neutralizes_nested_response_shape() -> None:
|
|
"""Applicability and localization retain later observations as nested evidence."""
|
|
|
|
manager = _file_manager()
|
|
manager.prompt_formatter_builder = _ShiftedBuilder()
|
|
current_prompt = (
|
|
"== BEGIN EARLIER DECISION ==\n"
|
|
'Original task\nReturn {"outcome":"supported"}.\n'
|
|
"== END EARLIER DECISION =="
|
|
)
|
|
request = AiderEditRequest("Current role", [], [], backlog_guidance=["Current work"])
|
|
guidance = FutureGuidance("later", "earlier", "First observation\nSecond observation", "The current work stalled.")
|
|
client = ScriptedCompletionClient((
|
|
'{"applicable": true, "reason": "This is the same local obligation."}',
|
|
'{"localized_prompt": "Check the current operation."}',
|
|
))
|
|
applicability = FutureGuidanceApplicability(
|
|
client=client, file_manager=manager, request=request, future_guidance=guidance,
|
|
current_prompt=current_prompt, logger=logging.getLogger(__name__),
|
|
)
|
|
localization = FutureGuidanceRelocalization(
|
|
client=client, file_manager=manager, request=request, future_guidance=guidance,
|
|
current_prompt=current_prompt, logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
assert applicability() is True
|
|
assert localization() is True
|
|
assert applicability.prompt.startswith("======== BEGIN FUTURE GUIDANCE APPLICABILITY ========\n")
|
|
assert localization.prompt.startswith("======== BEGIN FUTURE GUIDANCE RELOCALIZATION ========\n")
|
|
assert "========== BEGIN FUTURE GUIDANCE ==========\n" in applicability.prompt
|
|
assert "========== BEGIN APPLICABLE FUTURE GUIDANCE ==========\n" in localization.prompt
|
|
assert all("============ BEGIN MESSAGE ============\nFirst observation\nSecond observation\n" in prompt for prompt in client.prompts)
|
|
assert all(map(
|
|
lambda prompt: "\\== BEGIN EARLIER DECISION ==\nOriginal task" in prompt,
|
|
client.prompts,
|
|
))
|
|
assert all(map(lambda prompt: '{"outcome":"supported"}' not in prompt, client.prompts))
|
|
assert all(map(lambda prompt: 'Original task\n---REDACTED---\n' in prompt, client.prompts))
|
|
assert all(map(lambda prompt: '"outcome"' not in prompt, client.prompts))
|
|
assert json.dumps(applicability.valid_response_references[0], separators=(",", ":")) in client.prompts[0]
|
|
assert json.dumps(localization.valid_response_references[0], separators=(",", ":")) in client.prompts[1]
|
|
assert current_prompt == (
|
|
"== BEGIN EARLIER DECISION ==\n"
|
|
'Original task\nReturn {"outcome":"supported"}.\n'
|
|
"== END EARLIER DECISION =="
|
|
)
|
|
assert request.backlog_guidance == ["Current work"]
|
|
|
|
|
|
class _ReducingDecision(StabilizationDecision):
|
|
"""Exercise the base retry loop with explicit child-owned response state."""
|
|
|
|
def __init__(self, client: StaticCompletionClient | ScriptedCompletionClient | None = None) -> None:
|
|
"""Create a real decision base with an unused stationary completion client."""
|
|
|
|
manager = _file_manager()
|
|
super().__init__(
|
|
client=client or StaticCompletionClient('{"outcome": "supported"}'),
|
|
file_manager=manager,
|
|
request=AiderEditRequest("", [], []),
|
|
)
|
|
self.calls: list[tuple[object | None, ValidationResult | None]] = []
|
|
self.invalid = ValidationResult(errors=(SemanticValidationError(
|
|
location=ValidationLocation(["response", "outcome"]),
|
|
error="must be supported.",
|
|
),))
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""Declare the one response shape needed by the reduction fixture."""
|
|
|
|
return [{"outcome": "supported"}]
|
|
|
|
def _decide(
|
|
self,
|
|
response: object | None = None,
|
|
validation: ValidationResult | None = None,
|
|
) -> tuple[object | None, ValidationResult]:
|
|
"""Return one rejection followed by one accepted decision."""
|
|
|
|
self.calls.append((response, validation))
|
|
if validation is None:
|
|
return {"outcome": "unsupported"}, self.invalid
|
|
self.changes.append((self, True))
|
|
return {"outcome": "supported"}, ValidationResult()
|
|
|
|
|
|
def test_stabilization_localizes_applicable_future_guidance_before_child_retries(
|
|
caplog,
|
|
) -> None:
|
|
"""An applicable later observation changes only this decision and request."""
|
|
|
|
client = ScriptedCompletionClient((
|
|
'{"applicable": true, "reason": "The current decision can avoid the dead end."}',
|
|
json.dumps({"localized_prompt": 'Verify the overlooked selected operation before deciding. {"outcome":"nested"}'}),
|
|
))
|
|
decision = _ReducingDecision(client)
|
|
decision.prompt = 'ORIGINAL DECISION PROMPT\nReturn {"outcome":"supported"}.'
|
|
decision.request.backlog_guidance = ["Original work instruction."]
|
|
decision.request.future_guidance = (FutureGuidance(
|
|
origin_message_id="later",
|
|
addressed_message_id="earlier",
|
|
message="Reconsider the selected guidance.",
|
|
justification="Later work reached a dead end.",
|
|
),)
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
assert decision() is True
|
|
assert decision.calls == [
|
|
(None, None),
|
|
({"outcome": "unsupported"}, decision.invalid),
|
|
]
|
|
assert decision.request.backlog_guidance == [
|
|
"Original work instruction.",
|
|
'Verify the overlooked selected operation before deciding. {"outcome":"nested"}',
|
|
]
|
|
assert "LOCALIZED BACKLOG GUIDANCE" in decision.prompt
|
|
assert "Verify the overlooked selected operation before deciding." not in decision.prompt
|
|
assert '{"outcome":"nested"}' not in decision.prompt
|
|
assert '"outcome":"nested"' not in decision.prompt
|
|
assert "---REDACTED---" in decision.prompt
|
|
assert '{"outcome":"supported"}' in decision.prompt
|
|
assert "Reconsider the selected guidance." not in decision.prompt
|
|
assert "applicable=True" in caplog.text
|
|
assert len(client.prompts) == 2
|
|
assert "FUTURE GUIDANCE APPLICABILITY" in client.prompts[0]
|
|
assert "FUTURE GUIDANCE RELOCALIZATION" in client.prompts[1]
|
|
assert all(map(lambda prompt: '{"outcome":"supported"}' not in prompt, client.prompts))
|
|
assert all(map(lambda prompt: 'ORIGINAL DECISION PROMPT\n---REDACTED---' in prompt, client.prompts))
|
|
assert all(map(lambda prompt: '"outcome":"supported"' not in prompt, client.prompts))
|
|
assert not hasattr(decision, "response")
|
|
assert not hasattr(decision, "validation")
|
|
|
|
|
|
def test_stabilization_sequences_future_guidance_through_neutralized_current_prompt() -> None:
|
|
"""Each admitted localization joins the prompt while its response shape is neutralized."""
|
|
|
|
client = ScriptedCompletionClient((
|
|
'{"applicable": true, "reason": "The first observation applies."}',
|
|
'{"localized_prompt": "First localized instruction."}',
|
|
'{"applicable": true, "reason": "The second observation applies after the first."}',
|
|
'{"localized_prompt": "Second localized instruction."}',
|
|
))
|
|
decision = _ReducingDecision(client)
|
|
decision.prompt = 'PREDICATE CONTRACT {"outcome":"supported"}'
|
|
decision.request.authoritative_prompt = "Repair the selected local work."
|
|
decision.request.backlog_guidance = ["Original work instruction."]
|
|
decision.request.future_guidance = (
|
|
FutureGuidance("later-1", "earlier", "First observation.", "First dead end."),
|
|
FutureGuidance("later-2", "earlier", "Second observation.", "Second dead end."),
|
|
)
|
|
|
|
assert decision() is True
|
|
assert decision.request.backlog_guidance == [
|
|
"Original work instruction.",
|
|
"First localized instruction.",
|
|
"Second localized instruction.",
|
|
]
|
|
assert len(client.prompts) == 4
|
|
assert all(map(lambda prompt: "First localized instruction." not in prompt, client.prompts[:2]))
|
|
assert all(map(lambda prompt: "First localized instruction." in prompt, client.prompts[2:]))
|
|
assert all(map(lambda prompt: '{"outcome":"supported"}' not in prompt, client.prompts))
|
|
assert all(map(lambda prompt: '"outcome":"supported"' not in prompt, client.prompts))
|
|
assert all(map(lambda prompt: 'PREDICATE CONTRACT' not in prompt, client.prompts))
|
|
assert all(map(lambda prompt: '---REDACTED---' in prompt, client.prompts))
|
|
assert 'PREDICATE CONTRACT {"outcome":"supported"}' in decision.prompt
|
|
assert decision.prompt.count("BEGIN LOCALIZED BACKLOG GUIDANCE") == 2
|
|
|
|
|
|
def test_stabilization_logs_inapplicable_future_guidance_without_mutation(caplog) -> None:
|
|
"""A later observation outside this decision's authority leaves current work unchanged."""
|
|
|
|
client = ScriptedCompletionClient((
|
|
'{"applicable": false, "reason": "This decision cannot change the selected owner."}',
|
|
))
|
|
decision = _ReducingDecision(client)
|
|
decision.prompt = "ORIGINAL DECISION PROMPT"
|
|
decision.request.backlog_guidance = ["Original work instruction."]
|
|
decision.request.future_guidance = (FutureGuidance(
|
|
origin_message_id="later",
|
|
addressed_message_id="earlier",
|
|
message="Reconsider the selected guidance.",
|
|
justification="Later work reached a dead end.",
|
|
),)
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
assert decision() is True
|
|
|
|
assert decision.request.backlog_guidance == ["Original work instruction."]
|
|
assert decision.prompt == "ORIGINAL DECISION PROMPT"
|
|
assert "applicable=False" in caplog.text
|
|
assert len(client.prompts) == 1
|
|
|
|
|
|
def test_alpha_edit_acceptance_retains_goal_and_evidence_without_editor_instructions() -> None:
|
|
"""Acceptance asks about the goal and resulting code without nesting an editing contract."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/alpha_readonly.py",),
|
|
tests=("tests/test_alpha.py",),
|
|
prompt="ALPHA_ORIGINAL_COT_SENTINEL",
|
|
)
|
|
readonly_source = _source(manager, "src/alpha_readonly.py")
|
|
readonly_test = _test(manager, "tests/test_alpha.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="ALPHA_AUTHORITATIVE_TASK_SENTINEL",
|
|
editable_files=[],
|
|
readonly_files=[Path(readonly_source.location), Path(readonly_test.location)],
|
|
pinned_context=["ALPHA_PINNED_CONTEXT_SENTINEL"],
|
|
backlog_guidance=["ALPHA_WORK_GUIDANCE_SENTINEL"],
|
|
failure_error_message="ALPHA_FAILURE_MESSAGE_SENTINEL",
|
|
failure_stack_trace=["ALPHA_FAILURE_STACK_SENTINEL"],
|
|
failure_raw_output="ALPHA_FAILURE_RAW_OUTPUT_SENTINEL",
|
|
allowed_additional_roots=[Path("src/allowed.py")],
|
|
)
|
|
decision = AlphaEditAcceptanceDecision(
|
|
client=StaticCompletionClient('{"outcome": "pass", "reason": ["pass"], "instruction": ""}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
editable_sources=[],
|
|
readonly_sources=[readonly_source],
|
|
readonly_tests=[readonly_test],
|
|
previous_summary="ALPHA_PREVIOUS_EDITOR_SUMMARY_SENTINEL",
|
|
)
|
|
|
|
assert isinstance(decision, StabilizationDecision)
|
|
assert decision.request is request
|
|
assert all(token in decision.prompt for token in (
|
|
"ALPHA_ORIGINAL_COT_SENTINEL",
|
|
"ALPHA_FAILURE_MESSAGE_SENTINEL",
|
|
"ALPHA_FAILURE_STACK_SENTINEL",
|
|
readonly_source.location,
|
|
readonly_test.location,
|
|
))
|
|
assert not any(map(lambda token: token in decision.prompt, (
|
|
"ALPHA_AUTHORITATIVE_TASK_SENTINEL", "ALPHA_FAILURE_RAW_OUTPUT_SENTINEL",
|
|
"ALPHA_PREVIOUS_EDITOR_SUMMARY_SENTINEL",
|
|
)))
|
|
|
|
|
|
def test_alpha_blank_source_correction_decision_input_is_superset() -> None:
|
|
"""Blank-source correction retains the task and bypasses model work without blanks."""
|
|
|
|
manager = _file_manager(sources=("src/blank_owner.py",))
|
|
editable = _source(manager, "src/blank_owner.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="BLANK_REQUEST_ROLE_BOUNDARY_SENTINEL",
|
|
editable_files=[Path(editable.location)],
|
|
readonly_files=[],
|
|
)
|
|
decision = AlphaBlankSourceCorrectionDecision(
|
|
client=StaticOpenAICompatibleClient('{"instruction": "unused", "reason": "unused", "blank_files": []}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
work_needed_by_location={editable.location: "BLANK_WORK_NEEDED_SENTINEL"},
|
|
editable_sources=[editable],
|
|
readonly_sources=[],
|
|
readonly_tests=[],
|
|
blank_sources=[],
|
|
previous_summary="BLANK_PREVIOUS_EDITOR_SUMMARY_SENTINEL",
|
|
alpha_task="BLANK_ALPHA_TASK_SENTINEL",
|
|
)
|
|
|
|
assert decision() is True
|
|
assert decision.additional == {"instruction": "", "blank_files": []}
|
|
assert "BLANK_ALPHA_TASK_SENTINEL" in decision.prompt
|
|
assert "BLANK_WORK_NEEDED_SENTINEL" in decision.prompt
|
|
|
|
|
|
def test_inspection_work_needed_decision_input_is_superset() -> None:
|
|
"""Inspection prompt keeps focused, related, and immutable work evidence distinct."""
|
|
|
|
manager = _file_manager(sources=(
|
|
"src/inspection_focus.py",
|
|
"src/inspection_upstream.py",
|
|
"src/inspection_downstream.py",
|
|
), prompt="INSPECTION_ORIGINAL_COT_SENTINEL")
|
|
focused = _source(manager, "src/inspection_focus.py")
|
|
upstream = _source(manager, "src/inspection_upstream.py")
|
|
downstream = _source(manager, "src/inspection_downstream.py")
|
|
item = InspectionBacklog(
|
|
focused_spotlight=focused,
|
|
summary="INSPECTION_SUMMARY_SENTINEL",
|
|
upstream_classes=[upstream],
|
|
downstream_classes=[downstream],
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[Path(focused.location)],
|
|
readonly_files=[Path(upstream.location), Path(downstream.location)],
|
|
)
|
|
decision = InspectionWorkNeededDecision(
|
|
client=StaticCompletionClient('{"work_required": false, "evidence_type": "none", "semantic_divergence": "none", "reason": ["pass"], "work_needed": ""}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
inspection_prompt="INSPECTION_CURRENT_TASK_SENTINEL",
|
|
focused_source=focused,
|
|
readonly_sources=[upstream, downstream, upstream],
|
|
inspection_items=[item],
|
|
)
|
|
|
|
assert isinstance(decision, StabilizationDecision)
|
|
assert decision.request is request
|
|
assert all(token in decision.prompt for token in (
|
|
"INSPECTION_ORIGINAL_COT_SENTINEL",
|
|
"INSPECTION_CURRENT_TASK_SENTINEL",
|
|
"INSPECTION_SUMMARY_SENTINEL",
|
|
focused.location,
|
|
upstream.location,
|
|
downstream.location,
|
|
))
|
|
assert decision.prompt.index("ORIGINAL COT: IMMUTABLE STABILIZATION TARGET") < decision.prompt.index(
|
|
"LOCAL INSPECTION SCOPE"
|
|
)
|
|
|
|
|
|
def test_source_remediation_guidance_requests_a_grounded_stepwise_instruction() -> None:
|
|
"""Source remediation prompt asks for grounded ordered work from real source evidence."""
|
|
|
|
manager = _file_manager(sources=("src/remediation.py", "src/remediation_context.py"))
|
|
source = _source(manager, "src/remediation.py")
|
|
readonly = _source(manager, "src/remediation_context.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="REMEDIATION_AUTHORITATIVE_TASK_SENTINEL",
|
|
editable_files=[Path(source.location)],
|
|
readonly_files=[Path(readonly.location)],
|
|
failure_error_message="REMEDIATION_FAILURE_MESSAGE_SENTINEL",
|
|
failure_stack_trace=["REMEDIATION_FAILURE_STACK_SENTINEL"],
|
|
)
|
|
guidance = SourceRemediationGuidanceCheck(
|
|
client=StaticOpenAICompatibleClient('{"work_needed": "unused", "reason": ["pass"]}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
original_cot="REMEDIATION_ORIGINAL_COT_SENTINEL",
|
|
source=source,
|
|
failure_evidence="REMEDIATION_FAILURE_SENTINEL",
|
|
precedent_diffs="REMEDIATION_PRECEDENT_SENTINEL",
|
|
readonly_sources=[readonly],
|
|
)
|
|
|
|
assert isinstance(guidance, StabilizationDecision)
|
|
assert guidance.request is request
|
|
assert all(token in guidance.prompt for token in (
|
|
"concrete current-file divergence",
|
|
"compact ordered instruction",
|
|
"1. Identify the exact current operation",
|
|
"2. State the smallest source change",
|
|
"3. State the focused verification",
|
|
source.location,
|
|
readonly.location,
|
|
))
|
|
|
|
|
|
def test_bootstrap_acceptance_retries_semantic_failure_with_validation_feedback() -> None:
|
|
"""Bootstrap retry adds validation evidence without changing the first prompt."""
|
|
|
|
manager = _file_manager(sources=("src/bootstrap.py",))
|
|
source = _source(manager, "src/bootstrap.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="BOOTSTRAP_REQUEST_SENTINEL",
|
|
editable_files=[Path(source.location)],
|
|
readonly_files=[],
|
|
)
|
|
client = ScriptedCompletionClient((
|
|
json.dumps({
|
|
"outcome": "pass",
|
|
"reason": ["pass"],
|
|
"instruction": "Keep editing bootstrap source.",
|
|
}),
|
|
json.dumps({
|
|
"outcome": "pass",
|
|
"reason": ["pass"],
|
|
"instruction": "",
|
|
}),
|
|
))
|
|
decision = BootstrapAcceptanceCheck(
|
|
client=client,
|
|
file_manager=manager,
|
|
edit_request=request,
|
|
editable_sources=[source],
|
|
readonly_sources=[],
|
|
readonly_tests=[],
|
|
previous_summary="BOOTSTRAP_SUMMARY_SENTINEL",
|
|
)
|
|
first_prompt = decision.prompt
|
|
|
|
assert decision() is True
|
|
assert client.prompts[0] == first_prompt
|
|
assert client.prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert "pass must leave instruction empty" in client.prompts[1]
|
|
|
|
|
|
def test_stabilization_decision_does_not_intercept_the_dead_end_decision() -> None:
|
|
"""A normal decision keeps ownership of its own prompt and response loop."""
|
|
|
|
manager = _file_manager(sources=("src/bootstrap.py",))
|
|
source = _source(manager, "src/bootstrap.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="BOOTSTRAP_REQUEST_SENTINEL",
|
|
editable_files=[Path(source.location)],
|
|
readonly_files=[],
|
|
)
|
|
client = ScriptedCompletionClient((json.dumps({
|
|
"outcome": "pass",
|
|
"reason": ["pass"],
|
|
"instruction": "",
|
|
}),))
|
|
decision = BootstrapAcceptanceCheck(
|
|
client=client,
|
|
file_manager=manager,
|
|
edit_request=request,
|
|
editable_sources=[source],
|
|
readonly_sources=[],
|
|
readonly_tests=[],
|
|
previous_summary="BOOTSTRAP_SUMMARY_SENTINEL",
|
|
)
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
item = InspectionBacklog(focused_spotlight=source, summary="bootstrap source")
|
|
graph.start(Phase.ALPHA, item)
|
|
|
|
with provenance.bind(graph):
|
|
with provenance.item_scope(Phase.ALPHA, item):
|
|
assert decision() is True
|
|
|
|
assert "DEAD-END DECISION" not in client.prompts[0]
|
|
assert "BOOTSTRAP EDIT ACCEPTANCE" in client.prompts[0]
|
|
|
|
|
|
def test_beta_contract_guidance_decision_input_is_superset() -> None:
|
|
"""Beta refinement retains current test and readonly source evidence in one prompt."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/contract_one.py", "src/contract_two.py"),
|
|
tests=("tests/test_contract.py",),
|
|
)
|
|
focused_test = _test(manager, "tests/test_contract.py")
|
|
source_one = _source(manager, "src/contract_one.py")
|
|
source_two = _source(manager, "src/contract_two.py")
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[Path(focused_test.location)],
|
|
readonly_files=[Path(source_one.location), Path(source_two.location)],
|
|
failure_error_message="BETA_FAILURE_MESSAGE_SENTINEL",
|
|
failure_stack_trace=["BETA_FAILURE_STACK_SENTINEL"],
|
|
failure_raw_output="BETA_FAILURE_RAW_SENTINEL",
|
|
)
|
|
decision = BetaContractGuidanceDecision(
|
|
client=StaticOpenAICompatibleClient('{"beta_instruction": "unused"}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
focused_test=focused_test,
|
|
original_cot="BETA_ORIGINAL_COT_SENTINEL",
|
|
accepted_source_diff="BETA_ACCEPTED_SOURCE_DIFF_SENTINEL",
|
|
assignment_type="BETA_ASSIGNMENT_PROVENANCE_SENTINEL",
|
|
test_status="BETA_FOCUSED_TEST_STATUS_SENTINEL",
|
|
previous_beta_guidance="BETA_PREVIOUS_GUIDANCE_SENTINEL",
|
|
readonly_sources=[source_one, source_two, source_one],
|
|
logger=logging.getLogger("test.beta.backlog"),
|
|
)
|
|
|
|
assert isinstance(decision, StabilizationDecision)
|
|
assert decision.request is request
|
|
assert all(token in decision.prompt for token in (
|
|
"BETA_ORIGINAL_COT_SENTINEL",
|
|
"BETA_ACCEPTED_SOURCE_DIFF_SENTINEL",
|
|
"BETA_ASSIGNMENT_PROVENANCE_SENTINEL",
|
|
"BETA_FOCUSED_TEST_STATUS_SENTINEL",
|
|
"BETA_PREVIOUS_GUIDANCE_SENTINEL",
|
|
"BETA_FAILURE_MESSAGE_SENTINEL",
|
|
"BETA_FAILURE_STACK_SENTINEL",
|
|
focused_test.location,
|
|
source_one.location,
|
|
source_two.location,
|
|
))
|
|
assert "BETA_FAILURE_RAW_SENTINEL" not in decision.prompt
|
|
|
|
|
|
def _targeted_story(
|
|
manager: FileManager,
|
|
*,
|
|
focused_test: ClassMeta,
|
|
sources: list[ClassMeta],
|
|
failure_phase: str,
|
|
) -> TargetedFailureStory:
|
|
"""Create one real targeted failure story over explicit memory metadata."""
|
|
|
|
testing_item = TestingBacklog(
|
|
focused_spotlight=focused_test,
|
|
summary="ROUTING_TESTING_ITEM_SENTINEL",
|
|
upstream_classes=[sources[1]],
|
|
downstream_classes=[sources[0]],
|
|
original_cot="ROUTING_ORIGINAL_COT_SENTINEL",
|
|
change_type=ChangeType.Update,
|
|
failure_context_sources=sources[2:],
|
|
)
|
|
failing = FailingTest(
|
|
test=focused_test,
|
|
error_message="ROUTING_FAILURE_MESSAGE_SENTINEL",
|
|
stack_trace=[focused_test.location, *map(lambda source: source.location, sources), "ROUTING_FAILURE_STACK_SENTINEL"],
|
|
raw_output="ROUTING_FAILURE_RAW_OUTPUT_SENTINEL",
|
|
)
|
|
return TargetedFailureStory(
|
|
testing_item=testing_item,
|
|
raw_output=failing.raw_output,
|
|
work_needed="ROUTING_WORK_NEEDED_SENTINEL",
|
|
candidate_sources=sources,
|
|
failing=failing,
|
|
failure_phase=failure_phase,
|
|
accepted_alpha_summary="ROUTING_ACCEPTED_ALPHA_SUMMARY_SENTINEL",
|
|
structural_evidence={"structural": "ROUTING_STRUCTURAL_EVIDENCE_SENTINEL"},
|
|
semantic_evidence={"semantic": "ROUTING_SEMANTIC_EVIDENCE_SENTINEL"},
|
|
previous_beta_guidance="ROUTING_PREVIOUS_BETA_GUIDANCE_SENTINEL",
|
|
beta_edit_diff="ROUTING_BETA_EDIT_DIFF_SENTINEL",
|
|
beta_editor_summary="ROUTING_BETA_EDITOR_SUMMARY_SENTINEL",
|
|
)
|
|
|
|
|
|
def test_preflight_and_guidance_helpers_keep_independent_file_ladders() -> None:
|
|
"""One remediation may assess different files without sharing any of its three decision ladders."""
|
|
|
|
manager = _file_manager(sources=("src/owner.py", "src/a/item.py", "src/b/item.py"))
|
|
owner = _source(manager, "src/owner.py")
|
|
first = _source(manager, "src/a/item.py")
|
|
other = _source(manager, "src/b/item.py")
|
|
item = InspectionBacklog(focused_spotlight=owner, summary="repair")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=False)
|
|
graph.start(Phase.ALPHA, item)
|
|
answers = (
|
|
'{"applicable":true,"reason":"This candidate must be judged using the current lifecycle."}',
|
|
'{"localized_prompt":"Judge only this candidate against the current lifecycle."}',
|
|
'[]',
|
|
)
|
|
local = ScriptedCompletionClient(answers * 13)
|
|
cloud = ScriptedCompletionClient(answers)
|
|
runtime = StabilizationEscalation(
|
|
(local, cloud), check_name="runtime", prompt_formatter_builder=manager.prompt_formatter_builder,
|
|
)
|
|
guidance = FutureGuidance("later", item.message_id, "Honor the current lifecycle.", "The old evidence is stale.")
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
# Spend A's local allowance across accepted decisions, rather than manufacturing a different node.
|
|
for _ in range(12):
|
|
decision = AlphaSourcePreflightDecision(
|
|
client=runtime, file_manager=manager, source=first,
|
|
request=AiderEditRequest(
|
|
"Inspect candidate.", [Path(first.location)], [], future_guidance=(guidance,),
|
|
),
|
|
work_needed="Inspect candidate.",
|
|
)
|
|
assert decision() is True
|
|
other_decision = AlphaSourcePreflightDecision(
|
|
client=runtime, file_manager=manager, source=other,
|
|
request=AiderEditRequest(
|
|
"Inspect candidate.", [Path(other.location)], [], future_guidance=(guidance,),
|
|
),
|
|
work_needed="Inspect candidate.",
|
|
)
|
|
assert other_decision() is True
|
|
assert len(local.prompts) == 39
|
|
assert cloud.prompts == []
|
|
|
|
resumed = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=True)
|
|
manager.repo_exec.write_text(manager.source_path(first.location), "current_value = 42\n")
|
|
with provenance.bind(resumed), provenance.item_scope(Phase.ALPHA, item):
|
|
resumed_decision = AlphaSourcePreflightDecision(
|
|
client=runtime, file_manager=manager, source=first,
|
|
request=AiderEditRequest(
|
|
"Inspect candidate.", [Path(first.location)], [], future_guidance=(guidance,),
|
|
),
|
|
work_needed="Inspect current candidate.",
|
|
)
|
|
assert resumed_decision() is True
|
|
assert len(cloud.prompts) == 3
|
|
assert "FUTURE GUIDANCE APPLICABILITY" in cloud.prompts[0]
|
|
assert "FUTURE GUIDANCE RELOCALIZATION" in cloud.prompts[1]
|
|
assert "current_value = 42" in cloud.prompts[2]
|
|
assert len(local.prompts) == 39
|
|
|
|
|
|
def test_postflight_routing_uses_test_file_not_parent_source_for_progression() -> None:
|
|
"""Routing a second test under one Alpha source item starts with its own local allowance."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/owner.py", "src/related.py"),
|
|
tests=("tests/a/test_item.py", "tests/b/test_item.py"),
|
|
)
|
|
owner = _source(manager, "src/owner.py")
|
|
related = _source(manager, "src/related.py")
|
|
first = _test(manager, "tests/a/test_item.py")
|
|
other = _test(manager, "tests/b/test_item.py")
|
|
item = InspectionBacklog(focused_spotlight=owner, summary="repair")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=False)
|
|
graph.start(Phase.ALPHA, item)
|
|
answer = '{"outcome":"test","reason":"The assertion is stale.","additional":{"test_work_needed":"Update the assertion."}}'
|
|
local = ScriptedCompletionClient((answer,) * 13)
|
|
cloud = ScriptedCompletionClient((answer,))
|
|
runtime = StabilizationEscalation(
|
|
(local, cloud), check_name="runtime", prompt_formatter_builder=manager.prompt_formatter_builder,
|
|
)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
# Each repeated decision has the same failed test and must retain its spent attempts.
|
|
for focused_test in (first,) * 12 + (other, first):
|
|
decision = TargetedAlphaOrBetaRoutingDecision(
|
|
client=runtime, file_manager=manager,
|
|
request=AiderEditRequest("Route failure", [manager.test_path(focused_test.location)], []),
|
|
story=_targeted_story(manager, focused_test=focused_test, sources=[owner, related], failure_phase="after_alpha"),
|
|
)
|
|
decision()
|
|
assert decision.outcome == "test"
|
|
assert len(local.prompts) == 13
|
|
assert len(cloud.prompts) == 1
|
|
assert other.location in local.prompts[-1]
|
|
assert first.location in cloud.prompts[0]
|
|
|
|
|
|
def test_targeted_alpha_or_beta_routing_decision_input_is_superset() -> None:
|
|
"""Ordinary routing retains collective current source, test, and failure evidence."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/routing_one.py", "src/routing_two.py", "src/routing_three.py"),
|
|
tests=("tests/test_routing.py",),
|
|
)
|
|
focused_test = _test(manager, "tests/test_routing.py")
|
|
sources = [
|
|
_source(manager, "src/routing_one.py"),
|
|
_source(manager, "src/routing_two.py"),
|
|
_source(manager, "src/routing_three.py"),
|
|
]
|
|
story = _targeted_story(
|
|
manager,
|
|
focused_test=focused_test,
|
|
sources=sources,
|
|
failure_phase="ROUTING_FAILURE_PHASE_SENTINEL",
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[Path(focused_test.location)],
|
|
readonly_files=[Path(source.location) for source in sources],
|
|
)
|
|
decision = TargetedAlphaOrBetaRoutingDecision(
|
|
client=StaticCompletionClient('{"outcome": "source", "reason": "grounded", "additional": {"test_work_needed": ""}}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
story=story,
|
|
logger=logging.getLogger("test.beta.routing"),
|
|
)
|
|
|
|
assert isinstance(decision, StabilizationDecision)
|
|
assert decision.request is request
|
|
assert all(token in decision.prompt for token in (
|
|
"ROUTING_ORIGINAL_COT_SENTINEL",
|
|
"ROUTING_ACCEPTED_ALPHA_SUMMARY_SENTINEL",
|
|
"ROUTING_TESTING_ITEM_SENTINEL",
|
|
"ROUTING_FAILURE_RAW_OUTPUT_SENTINEL",
|
|
focused_test.location,
|
|
*(source.location for source in sources),
|
|
))
|
|
|
|
|
|
def test_targeted_collection_routing_decision_input_is_superset() -> None:
|
|
"""Collection routing retains its singular test boundary and source context."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/collection_one.py", "src/collection_two.py", "src/collection_three.py"),
|
|
tests=("tests/test_collection.py",),
|
|
)
|
|
focused_test = _test(manager, "tests/test_collection.py")
|
|
sources = [
|
|
_source(manager, "src/collection_one.py"),
|
|
_source(manager, "src/collection_two.py"),
|
|
_source(manager, "src/collection_three.py"),
|
|
]
|
|
story = _targeted_story(
|
|
manager,
|
|
focused_test=focused_test,
|
|
sources=sources,
|
|
failure_phase="collection",
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[Path(focused_test.location)],
|
|
readonly_files=[Path(source.location) for source in sources],
|
|
)
|
|
decision = TargetedCollectionAlphaOrBetaRoutingDecision(
|
|
client=StaticCompletionClient('{"assignments": []}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
story=story,
|
|
logger=logging.getLogger("test.beta.collection"),
|
|
)
|
|
|
|
assert isinstance(decision, StabilizationDecision)
|
|
assert decision.request is request
|
|
assert "collection" in decision.prompt
|
|
assert focused_test.location in decision.prompt
|
|
assert all(source.location in decision.prompt for source in sources)
|
|
assert "ROUTING_FAILURE_MESSAGE_SENTINEL" in decision.prompt
|
|
assert "ROUTING_FAILURE_STACK_SENTINEL" in decision.prompt
|
|
|
|
|
|
def test_file_blame_decision_input_is_superset() -> None:
|
|
"""File blame retains each failed test and every candidate source in one prompt."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/blame_one.py", "src/blame_two.py", "src/blame_three.py"),
|
|
tests=("tests/test_blame_one.py", "tests/test_blame_two.py"),
|
|
)
|
|
sources = [
|
|
_source(manager, "src/blame_one.py"),
|
|
_source(manager, "src/blame_two.py"),
|
|
_source(manager, "src/blame_three.py"),
|
|
]
|
|
first_test = _test(manager, "tests/test_blame_one.py")
|
|
second_test = _test(manager, "tests/test_blame_two.py")
|
|
first_failure = FailingTest(
|
|
test=first_test,
|
|
error_message="BLAME_FAILURE_MESSAGE_ONE_SENTINEL",
|
|
stack_trace=[*map(lambda source: source.location, sources), "BLAME_FAILURE_STACK_ONE_SENTINEL"],
|
|
raw_output="BLAME_FAILURE_RAW_ONE_SENTINEL",
|
|
)
|
|
second_failure = FailingTest(
|
|
test=second_test,
|
|
error_message="BLAME_FAILURE_MESSAGE_TWO_SENTINEL",
|
|
stack_trace=["BLAME_FAILURE_STACK_TWO_SENTINEL"],
|
|
raw_output="BLAME_FAILURE_RAW_TWO_SENTINEL",
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[],
|
|
readonly_files=[Path(source.location) for source in sources],
|
|
)
|
|
decision = FileBlameDecision(
|
|
client=StaticOpenAICompatibleClient('{"outcome": "none", "reason": "no owner", "additional": {"source_locations": []}}'),
|
|
file_manager=manager,
|
|
request=request,
|
|
current_cot="BLAME_CURRENT_TASK_SENTINEL",
|
|
failure_summary="BLAME_FAILURE_SUMMARY_SENTINEL",
|
|
failed_tests={
|
|
first_test.location: first_failure,
|
|
second_test.location: second_failure,
|
|
},
|
|
candidates=[sources[0], sources[1], sources[0], sources[2]],
|
|
)
|
|
|
|
assert isinstance(decision, StabilizationDecision)
|
|
assert decision.request is request
|
|
assert all(token in decision.prompt for token in (
|
|
"BLAME_CURRENT_TASK_SENTINEL",
|
|
"BLAME_FAILURE_SUMMARY_SENTINEL",
|
|
first_test.location,
|
|
second_test.location,
|
|
"BLAME_FAILURE_RAW_ONE_SENTINEL",
|
|
"BLAME_FAILURE_RAW_TWO_SENTINEL",
|
|
*(source.location for source in sources),
|
|
))
|
|
assert decision.prompt.count(sources[0].location) >= 1
|
|
|
|
|
|
@pytest.mark.parametrize("decision_type", (
|
|
AlphaBlankSourceCorrectionDecision,
|
|
AlphaSourcePreflightDecision,
|
|
BootstrapSourcePreflightDecision,
|
|
BetaContractGuidanceDecision,
|
|
TargetedAlphaOrBetaRoutingDecision,
|
|
TargetedCollectionAlphaOrBetaRoutingDecision,
|
|
InspectionWorkNeededDecision,
|
|
))
|
|
def test_local_retry_section_preserves_raw_completion(decision_type: type[StabilizationDecision]) -> None:
|
|
"""Every formerly lossy handler and inspection show the unchanged rejected completion."""
|
|
|
|
manager = _file_manager(
|
|
sources=("src/focus.py", "src/upstream.py", "src/context.py"),
|
|
tests=("tests/test_focus.py",),
|
|
)
|
|
source = _source(manager, "src/focus.py")
|
|
focused_test = _test(manager, "tests/test_focus.py")
|
|
sources = [source, _source(manager, "src/upstream.py"), _source(manager, "src/context.py")]
|
|
request = AiderEditRequest("Inspect the focused source.", [Path(source.location)], [])
|
|
rejected = ' \r\n```json\n{ "include": false,\n "broken": }\n```\n== BEGIN MODEL TEXT ==\n unchanged \t\n== END MODEL TEXT ==\n'
|
|
client = ScriptedOpenAICompatibleClient((rejected, '{ "second": "response" }', '{}'))
|
|
story = _targeted_story(manager, focused_test=focused_test, sources=sources, failure_phase="collection")
|
|
# Construct only the requested real decision; each case owns isolated evidence and responses.
|
|
factories = {
|
|
AlphaBlankSourceCorrectionDecision: lambda: AlphaBlankSourceCorrectionDecision(
|
|
client=client, file_manager=manager, request=request,
|
|
work_needed_by_location={source.location: "Restore the source."},
|
|
editable_sources=[source], readonly_sources=[], readonly_tests=[],
|
|
blank_sources=[source], previous_summary="",
|
|
),
|
|
AlphaSourcePreflightDecision: lambda: AlphaSourcePreflightDecision(
|
|
client=client, file_manager=manager, request=request, source=source,
|
|
work_needed="Inspect the focused source.",
|
|
),
|
|
BootstrapSourcePreflightDecision: lambda: BootstrapSourcePreflightDecision(
|
|
client=client, file_manager=manager, request=request,
|
|
original_request="Implement the source.", implementation_prompt="Implement the source.",
|
|
design_plan={}, editable_sources=[source], readonly_sources=[],
|
|
work_needed_by_location={source.location: "Implement the source."},
|
|
),
|
|
BetaContractGuidanceDecision: lambda: BetaContractGuidanceDecision(
|
|
client=client, file_manager=manager, request=request, focused_test=focused_test,
|
|
original_cot="Preserve the contract.", accepted_source_diff="",
|
|
assignment_type="adapt", test_status="failed", readonly_sources=[source],
|
|
),
|
|
TargetedAlphaOrBetaRoutingDecision: lambda: TargetedAlphaOrBetaRoutingDecision(
|
|
client=client, file_manager=manager, request=request, story=story,
|
|
),
|
|
TargetedCollectionAlphaOrBetaRoutingDecision: lambda: TargetedCollectionAlphaOrBetaRoutingDecision(
|
|
client=client, file_manager=manager, request=request, story=story,
|
|
),
|
|
InspectionWorkNeededDecision: lambda: InspectionWorkNeededDecision(
|
|
client=client, file_manager=manager, request=request,
|
|
inspection_prompt="Inspect the focused source.", focused_source=source,
|
|
readonly_sources=[], inspection_items=[],
|
|
),
|
|
}
|
|
decision = factories[decision_type]()
|
|
|
|
previous, validation = decision._decide()
|
|
assert not validation.valid
|
|
previous, validation = decision._decide(previous, validation)
|
|
|
|
assert not validation.valid
|
|
assert (
|
|
"==== BEGIN REJECTED PREVIOUS RESPONSE ====\n"
|
|
+ rejected
|
|
+ "\n==== END REJECTED PREVIOUS RESPONSE ===="
|
|
) in client.prompts[1]
|
|
assert "Expecting value" in client.prompts[1]
|
|
assert "---REDACTED---" not in client.prompts[1]
|
|
|
|
decision._decide(previous, validation)
|
|
|
|
assert (
|
|
'==== BEGIN REJECTED PREVIOUS RESPONSE ====\n{ "second": "response" }\n'
|
|
'==== END REJECTED PREVIOUS RESPONSE ===='
|
|
) in client.prompts[2]
|
|
assert "Expecting value" not in client.prompts[2]
|
|
assert "broken" not in client.prompts[2]
|
|
assert "MODEL TEXT" not in client.prompts[2]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", tuple(AiderFailureMode))
|
|
def test_request_projection_preserves_originating_test(mode: AiderFailureMode) -> None:
|
|
"""Changing visible failure detail cannot change the compiler's execution target."""
|
|
|
|
manager = _file_manager(sources=("src/value.py",), tests=("tests/origin.py", "tests/support.py"))
|
|
origin = _test(manager, "tests/origin.py")
|
|
request = AiderEditRequest(
|
|
"repair", [Path("src/value.py")], [Path("tests/support.py"), Path("tests/origin.py")],
|
|
is_remediation=True, remediation_originator_test=origin, failure_error_message="historical failure",
|
|
)
|
|
|
|
projected = AiderEditRequestMask(request, mode).projected()
|
|
|
|
assert projected.is_remediation
|
|
assert projected.remediation_originator_test is origin
|
|
assert request.is_remediation
|
|
assert request.remediation_originator_test is origin
|