1287 lines
60 KiB
Python
1287 lines
60 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import TextIO
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.ai_client.completion_client import CompletionClient
|
|
from band_aider.ai_client.stabilization_escalation import StabilizationEscalation
|
|
from band_aider.judgement.failure_source_candidate_selection_strategy import FailureCandidateScope
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.validation_receipts import ValidationReceipt, ValidationReceipts
|
|
from band_aider.context.provenance import FutureGuidance
|
|
from band_aider.context.file_blame_check import (
|
|
FILE_BLAME_CANDIDATE_PRECEDENCE_RESPONSE_SHAPE,
|
|
FileBlameCandidatePrecedenceDecision,
|
|
FileBlameDecision,
|
|
)
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
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.judgement.pytest_runner_strategy import PytestRunnerStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult
|
|
from band_aider.judgement.repo_exec import HostRepoExec
|
|
from band_aider.strategies import failure_source_candidate_selection_strategy_factory
|
|
from tests.support.completion_clients import StaticOpenAICompatibleClient
|
|
from tests.support.memory_path import MemoryPath
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
from band_aider.judgement.source_discovery_strategy import CallImport, CallTraceNode
|
|
from band_aider.judgement.python_ast_dependency_graph_maintainer import PythonAstDependencyGraphMaintainer
|
|
from band_aider.judgement.code_based_reasoning import CodeBasedReasoning
|
|
from band_aider.judgement.source_remediation_guidance_check import SourceRemediationGuidanceCheck
|
|
from band_aider.context.targeted_failure_owner_check import TargetedAlphaOrBetaRoutingDecision
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
from band_aider.context.working_backlog import ChangeType
|
|
from band_aider.context.working_backlog import InspectionBacklog, TestingBacklog as BetaWork
|
|
from band_aider.context import provenance
|
|
from band_aider.context.provenance import Phase, Provenance
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck
|
|
|
|
|
|
class _QueuedClient(OpenAICompatibleClient):
|
|
"""Return queued stationary JSON responses while preserving prompt evidence."""
|
|
|
|
def __init__(self, *responses: object) -> None:
|
|
"""Store every configured response in the exact decision order."""
|
|
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
self.responses = list(responses)
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record one prompt and return the next static response without network access."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
response = self.responses.pop(0)
|
|
return response if isinstance(response, str) else json.dumps(response)
|
|
|
|
|
|
class _BlameParser(CodeParser):
|
|
"""Provide nominal metadata to the FileManager without host filesystem reads."""
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return an empty metadata record for one lexical path."""
|
|
|
|
del prompt
|
|
return ClassMeta(file_name, "", [], {}, content_type=content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return metadata without consuming a host-backed source stream."""
|
|
|
|
del stream
|
|
return self.parse_file(file_name, prompt, content_type)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Return metadata records in the supplied lexical order."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _RemovedTargetClient(_QueuedClient):
|
|
"""Keep helper relevance selection separate from the scenario's ownership verdicts.
|
|
|
|
Example: ``_RemovedTargetClient(route, blame, guidance)`` supplies judgments
|
|
while choosing no optional dependency cards.
|
|
"""
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Answer optional rendering independently of the queued semantic decisions."""
|
|
|
|
if "REFERENCE CARD SELECTION" in prompt:
|
|
return '{"references": []}'
|
|
return super().complete(prompt, logger)
|
|
|
|
|
|
class _BlameFileManager(FileManager):
|
|
"""Bind file-blame scenarios to a fresh stationary in-memory repository."""
|
|
|
|
def __init__(self, root: MemoryPath) -> None:
|
|
"""Construct a nominal FileManager that owns all source access for a scenario."""
|
|
|
|
super().__init__(
|
|
parser=_BlameParser(),
|
|
client=StaticOpenAICompatibleClient('{"cdc": ""}'),
|
|
repo_root=root,
|
|
src_root=root,
|
|
test_root=root,
|
|
actor="test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
repo_exec=root.repository,
|
|
prompt="File-blame fixture.",
|
|
)
|
|
|
|
def resync(self) -> None:
|
|
"""Retain only explicitly seeded candidate source evidence in each scenario."""
|
|
|
|
self.src_classes = []
|
|
self.test_classes = []
|
|
self.src_index = {}
|
|
self.test_index = {}
|
|
self.src_dirty.clear()
|
|
self.test_dirty.clear()
|
|
|
|
|
|
class _HostBlameFileManager(FileManager):
|
|
"""Expose an actual temporary project to the runner and blame renderers.
|
|
|
|
Example: ``_HostBlameFileManager(root, client)`` binds one whole-tier
|
|
traceback scenario to a concrete host repository adapter.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
root: Path,
|
|
client: OpenAICompatibleClient,
|
|
) -> None:
|
|
"""Construct a manager whose inventories are explicitly seeded by the scenario."""
|
|
|
|
super().__init__(
|
|
parser=_BlameParser(),
|
|
client=client,
|
|
repo_root=root,
|
|
src_root=root / "src",
|
|
test_root=root / "tests",
|
|
actor="test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
repo_exec=HostRepoExec(root),
|
|
prompt="Raise the failure at its originating source.",
|
|
)
|
|
|
|
def resync(self) -> None:
|
|
"""Retain only the source and test metadata explicitly seeded by the scenario."""
|
|
|
|
self.src_classes = []
|
|
self.test_classes = []
|
|
self.src_index = {}
|
|
self.test_index = {}
|
|
self.src_dirty.clear()
|
|
self.test_dirty.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def memory_root() -> MemoryPath:
|
|
"""Return a fresh lexical root backed by an in-memory repository."""
|
|
|
|
return MemoryPath.from_repo(MemoryRepoExec())
|
|
|
|
|
|
def _source(location: str, content_type: ContentType = ContentType.Source) -> ClassMeta:
|
|
"""Return one source or test record for collective file-blame evidence."""
|
|
|
|
return ClassMeta(location, f"summary {location}", [], {}, content_type=content_type)
|
|
|
|
|
|
def _request(
|
|
root: MemoryPath,
|
|
candidates: list[ClassMeta],
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
future_guidance: tuple[FutureGuidance, ...] = (),
|
|
validation_receipts: ValidationReceipts | None = None,
|
|
) -> AiderEditRequest:
|
|
"""Return the readonly candidate scope for one collective blame decision."""
|
|
|
|
return AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[],
|
|
readonly_files=[root / candidate.location for candidate in candidates],
|
|
previous_cot_records=previous_cot_records,
|
|
future_guidance=future_guidance,
|
|
validation_receipts=validation_receipts or ValidationReceipts(),
|
|
)
|
|
|
|
|
|
def _decision(
|
|
root: MemoryPath,
|
|
client: CompletionClient,
|
|
candidates: list[ClassMeta],
|
|
failures: list[FailingTest],
|
|
*,
|
|
current_cot: str = "Repair the failure.",
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
future_guidance: tuple[FutureGuidance, ...] = (),
|
|
) -> FileBlameDecision:
|
|
"""Build a direct file-blame decision over stationary in-memory source evidence."""
|
|
|
|
file_manager = _BlameFileManager(root)
|
|
file_manager.src_classes = candidates
|
|
file_manager.src_index = dict(map(lambda entry: (entry[1].location, entry[0]), enumerate(candidates)))
|
|
for candidate in candidates:
|
|
file_manager.repo_exec.write_text(
|
|
root / candidate.location,
|
|
f"def owner_{candidate.location.replace('/', '_').replace('.', '_')}():\n return True\n",
|
|
)
|
|
return FileBlameDecision(
|
|
client=client,
|
|
file_manager=file_manager,
|
|
request=_request(
|
|
root,
|
|
candidates,
|
|
previous_cot_records,
|
|
future_guidance,
|
|
),
|
|
current_cot=current_cot,
|
|
failure_summary="The targeted Alpha test failed.",
|
|
failed_tests={failure.test.location: failure for failure in failures},
|
|
candidates=candidates,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("confirmed_owner", ("source", "test"))
|
|
def test_file_blame_reviews_grounded_contradiction_at_highest_tier(
|
|
memory_root: MemoryPath, confirmed_owner: str,
|
|
) -> None:
|
|
"""Contradiction review retains supporting evidence and grounds ownership in recorded file roles."""
|
|
|
|
# Directory names cannot replace the supplied inventory's declared ownership.
|
|
source = _source("tests/editor.py")
|
|
test = _source("src/test_editor.py", ContentType.Test)
|
|
support = _source("src/support_contract.py", ContentType.Test)
|
|
support_body = "def contract():\n return 'current support evidence'\n"
|
|
expected = 3 if confirmed_owner == "test" else 2
|
|
returned = "2" if confirmed_owner == "test" else "True"
|
|
test_line = f" assert focus() == {expected}"
|
|
memory_root.repository.write_text(memory_root / source.location, f"def focus():\n return {returned}\n")
|
|
memory_root.repository.write_text(memory_root / test.location, f"from tests.editor import focus\n\ndef test_edit():\n{test_line}\n")
|
|
memory_root.repository.write_text(memory_root / support.location, support_body)
|
|
failure = FailingTest(test=test, error_message=f"return {returned} violates the expectation",
|
|
stack_trace=[f"{source.location}:2: return {returned}"])
|
|
finding = {"location": test.location,
|
|
"reason": [test_line],
|
|
"explanation": "This test expectation is obsolete under the required contract."}
|
|
labelled = {**finding, "outcome": "source"}
|
|
ungrounded = {**finding, "reason": [" assert invented() == 3"]}
|
|
unadmitted = {**finding, "location": "tests/unrelated.py"}
|
|
confirmed = (finding if confirmed_owner == "test" else {
|
|
"location": source.location, "reason": [" return True"],
|
|
"explanation": "The contract-consistent test exposes the wrong production result.",
|
|
})
|
|
local, middle = _QueuedClient(labelled, ungrounded, finding), _QueuedClient()
|
|
strongest = _QueuedClient(unadmitted, confirmed)
|
|
escalation = StabilizationEscalation(
|
|
(local, middle, strongest), check_name="FileBlame",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
routing = ReasonableVibeCheck(outcome="source", reason="Production returns the wrong result.",
|
|
additional={"test_work_needed": ""})
|
|
manager = _BlameFileManager(memory_root)
|
|
manager.src_classes, manager.test_classes = [source], [test, support]
|
|
manager.src_index, manager.test_index = {source.location: 0}, {test.location: 0, support.location: 1}
|
|
decision = FileBlameDecision(
|
|
client=escalation, file_manager=manager, request=_request(memory_root, [source]),
|
|
current_cot="focus must return 2.", failure_summary="The focused test failed.",
|
|
candidates=[source], failed_tests={test.location: failure}, routing=routing,
|
|
backlog=BetaWork(test, "review this failure", failure_context_sources=[support]),
|
|
)
|
|
|
|
assert decision() is True
|
|
|
|
assert len(local.prompts) == 3
|
|
assert middle.prompts == []
|
|
assert len(strongest.prompts) == 2
|
|
assert all(support_body in prompt for prompt in (*local.prompts, *strongest.prompts))
|
|
assert "outcome" in local.prompts[1]
|
|
assert "OWNERSHIP CONTRADICTION REVIEW" not in local.prompts[2]
|
|
assert "OWNERSHIP CONTRADICTION REVIEW" in strongest.prompts[0]
|
|
assert json.dumps(finding) in strongest.prompts[0]
|
|
assert routing.reason in strongest.prompts[0]
|
|
assert "focus must return 2." in strongest.prompts[0]
|
|
assert test_line in strongest.prompts[0]
|
|
assert "Do not restart the original routing question" in strongest.prompts[0]
|
|
assert "tests/unrelated.py" in strongest.prompts[1]
|
|
assert decision.outcome == ("blame" if confirmed_owner == "source" else "test")
|
|
assert decision.additional.get("source_locations") == ([source.location] if confirmed_owner == "source" else None)
|
|
assert decision.additional.get("test_location") == (test.location if confirmed_owner == "test" else None)
|
|
[reference] = decision.valid_response_references
|
|
assert isinstance(reference, dict)
|
|
assert set(reference) == {"location", "reason", "explanation"}
|
|
|
|
|
|
@pytest.mark.parametrize("traced", [False, True])
|
|
def test_imported_call_attribution_uses_live_code_without_fabricating_trace_overlap(
|
|
memory_root: MemoryPath, traced: bool,
|
|
) -> None:
|
|
"""Adding a runtime frame does not invalidate an imported-call citation."""
|
|
source = _source("src/editor.py")
|
|
test = _source("tests/test_editor.py", ContentType.Test)
|
|
manager = _BlameFileManager(memory_root)
|
|
manager.src_classes = [source]
|
|
manager.test_classes = [test]
|
|
manager.repo_exec.write_text(memory_root / source.location, "def edit():\n return 17\n")
|
|
manager.repo_exec.write_text(memory_root / test.location, "from src.editor import edit\ndef test_edit():\n edit()\n assert False\n")
|
|
trace = ["tests/test_editor.py:4: AssertionError"]
|
|
# Both admission paths must preserve the same valid source citation.
|
|
if traced:
|
|
trace.append("src/editor.py:2: def edit():")
|
|
failure = FailingTest(test=test, error_message="assert False", stack_trace=trace, raw_output="original")
|
|
before = failure.to_record()
|
|
responses = [{"location": source.location, "reason": [" return 17"], "explanation": "This implementation violates the required return contract."}]
|
|
client = _QueuedClient(*responses)
|
|
check = FileBlameDecision(client=client, file_manager=manager, request=_request(memory_root, [source]),
|
|
current_cot="Return a valid edit result.", failure_summary="caught error",
|
|
failed_tests={test.location: failure}, candidates=[source, test])
|
|
|
|
check()
|
|
|
|
assert check.additional["source_locations"] == [source.location]
|
|
assert len(client.prompts) == 1
|
|
assert "STATIC IMPORTED CALLS" in client.prompts[0]
|
|
assert failure.to_record() == before
|
|
assert check.additional["failure_code_fragment"] == ""
|
|
|
|
|
|
def test_imported_constructor_requires_real_source_citations(memory_root: MemoryPath) -> None:
|
|
"""An earlier constructor can be cited despite only a later method appearing in the trace."""
|
|
source = _source("src/acceptance.py")
|
|
test = _source("tests/test_acceptance.py", ContentType.Test)
|
|
manager = _BlameFileManager(memory_root)
|
|
manager.src_classes, manager.test_classes = [source], [test]
|
|
manager.repo_exec.write_text(memory_root / source.location,
|
|
"from contextlib import ExitStack\n"
|
|
"class Acceptance:\n"
|
|
" def __init__(self, source):\n"
|
|
" with ExitStack() as memento_stack:\n"
|
|
" memento_stack.enter_context(source.snapshot())\n"
|
|
" def decide(self):\n"
|
|
" return False\n")
|
|
manager.repo_exec.write_text(memory_root / test.location,
|
|
"from src.acceptance import Acceptance\n"
|
|
"def test_acceptance():\n"
|
|
" check = Acceptance(None)\n"
|
|
" assert check.decide()\n")
|
|
failure = FailingTest(test=test, error_message="assert False",
|
|
stack_trace=["tests/test_acceptance.py:4: AssertionError",
|
|
"src/acceptance.py:7: in decide", " return False"], raw_output="original")
|
|
before = failure.to_record()
|
|
client = _QueuedClient(
|
|
{"location": source.location, "reason": ["with MissingStack() as memento_stack:"],
|
|
"explanation": "The constructor restores the snapshot before the decision."},
|
|
{"location": source.location, "reason": ["with ExitStack() as memento_stack:"],
|
|
"explanation": "The constructor restores the snapshot before the decision."})
|
|
check = FileBlameDecision(client=client, file_manager=manager, request=_request(memory_root, [source]),
|
|
current_cot="Preserve the edit until acceptance decides.", failure_summary="acceptance lost the edit",
|
|
failed_tests={test.location: failure}, candidates=[source, test])
|
|
|
|
check()
|
|
|
|
assert check.additional["source_locations"] == [source.location]
|
|
assert len(client.prompts) == 2
|
|
assert "MissingStack" in client.prompts[1]
|
|
assert "was not found as a complete line in live code" in client.prompts[1]
|
|
assert "largest shared fragment" not in client.prompts[1]
|
|
assert check.additional["failure_code_fragment"] == ""
|
|
assert failure.to_record() == before
|
|
|
|
|
|
def test_trace_only_citation_still_requires_failure_overlap(memory_root: MemoryPath) -> None:
|
|
"""Without imported-call evidence, a real source line still needs runtime overlap."""
|
|
source = _source("src/owner.py")
|
|
test = _source("tests/test_owner.py", ContentType.Test)
|
|
failure = FailingTest(test=test, error_message="Broken contract",
|
|
stack_trace=["src/owner.py:2", " return True"], raw_output="original")
|
|
client = _QueuedClient(
|
|
{"location": source.location, "reason": ["def owner_src_owner_py():"],
|
|
"explanation": "The implementation violates the required return contract."},
|
|
{"location": source.location, "reason": ["return True"],
|
|
"explanation": "The return violates the required contract."})
|
|
check = _decision(memory_root, client, [source], [failure])
|
|
|
|
check()
|
|
|
|
assert check.additional["source_locations"] == [source.location]
|
|
assert len(client.prompts) == 2
|
|
assert "must share more than five contiguous characters" in client.prompts[1]
|
|
assert check.additional["failure_code_fragment"] == "return True"
|
|
|
|
|
|
def test_file_blame_uses_fresh_static_calls_without_a_retained_receipt(memory_root: MemoryPath) -> None:
|
|
"""Fresh discovery supplies renderer evidence without a second execution ledger."""
|
|
source = _source("src/editor.py")
|
|
test = _source("tests/test_editor.py", ContentType.Test)
|
|
manager = _BlameFileManager(memory_root)
|
|
manager.src_classes, manager.test_classes = [source], [test]
|
|
definition = "def test_edit():\n assert edit()\n"
|
|
manager.repo_exec.write_text(memory_root / source.location, "def edit():\n return False\n")
|
|
manager.repo_exec.write_text(memory_root / test.location, "from src.editor import edit\n" + definition)
|
|
result = TestRunResult("pytest tests/test_editor.py::test_edit", TestRunOutcome.Failure, 1, "current output", ["tests/test_editor.py:3: AssertionError"], stdout="current output", stderr="captured diagnostic", test_name="test_edit")
|
|
receipt = ValidationReceipt.from_result(test, result, test_definition=definition, generation=1)
|
|
failure = FailingTest.from_result(test, result)
|
|
client = _QueuedClient({"location": source.location, "reason": [" return False"], "explanation": "The production return is wrong."})
|
|
|
|
with patch.object(ContextualFileRenderer, "build", wraps=ContextualFileRenderer.build) as presenter:
|
|
check = FileBlameDecision(client=client, file_manager=manager,
|
|
request=_request(memory_root, [source, test], validation_receipts=ValidationReceipts((receipt,), generation=1)),
|
|
current_cot="Return success.", failure_summary="failure", failed_tests={test.location: failure}, candidates=[source, test])
|
|
check()
|
|
|
|
assert "edit" in presenter.call_args.kwargs["call_trace"]
|
|
assert "current output" in client.prompts[0]
|
|
assert "VALIDATION RECEIPTS" not in client.prompts[0]
|
|
assert check.additional["source_locations"] == [source.location]
|
|
|
|
|
|
def test_file_blame_renders_every_pair_with_failure_and_current_cot(
|
|
memory_root: MemoryPath,
|
|
) -> None:
|
|
"""A pairwise blame comparison sees both files, failure evidence, and current COT."""
|
|
|
|
candidates = [
|
|
_source("band_aider/loop/alpha_loop.py"),
|
|
_source("band_aider/context/memento.py"),
|
|
]
|
|
failure = FailingTest(
|
|
test=_source("tests/test_alpha.py", ContentType.Test),
|
|
error_message="AttributeError: 'Memento' object has no attribute 'clear'",
|
|
stack_trace=[
|
|
"Traceback (most recent call last):",
|
|
"band_aider/loop/alpha_loop.py: source_file.set_or_get_memento(...).clear()",
|
|
"band_aider/context/memento.py: lifecycle error",
|
|
],
|
|
raw_output="pytest raw output with unchanged text",
|
|
)
|
|
client = _QueuedClient({
|
|
"location": "band_aider/loop/alpha_loop.py",
|
|
"reason": ["def owner_band_aider_loop_alpha_loop_py():"],
|
|
"explanation": "The stale call first violates the current COT.",
|
|
})
|
|
check = _decision(memory_root, client, candidates, [failure])
|
|
|
|
assert "BEGIN FAILURE MATCHED SOURCE" in check.prompt
|
|
assert "owner_band_aider_loop_alpha_loop_py" in check.prompt
|
|
assert "owner_band_aider_context_memento_py" in check.prompt
|
|
assert "closed candidate set" in check.prompt
|
|
assert "AttributeError: 'Memento' object has no attribute 'clear'" in check.prompt
|
|
assert "Traceback (most recent call last):" in check.prompt
|
|
assert "pytest raw output with unchanged text" in check.prompt
|
|
assert check.prompt.index("BEGIN FAILED TESTS") < check.prompt.index(
|
|
"BEGIN CANDIDATE SOURCE FILES"
|
|
) < check.prompt.index("BEGIN CURRENT COT")
|
|
|
|
assert check() is True
|
|
assert len(client.prompts) == 1
|
|
comparison_prompt = client.prompts[0]
|
|
assert "Compare exactly two preselected production candidates." in comparison_prompt
|
|
assert "first violates the required contract" in comparison_prompt
|
|
assert "outer wrapper or propagator" in comparison_prompt
|
|
assert "owner_band_aider_loop_alpha_loop_py" in comparison_prompt
|
|
assert "owner_band_aider_context_memento_py" in comparison_prompt
|
|
assert "Repair the failure." in comparison_prompt
|
|
assert "pytest raw output with unchanged text" in comparison_prompt
|
|
assert json.dumps(FILE_BLAME_CANDIDATE_PRECEDENCE_RESPONSE_SHAPE) in comparison_prompt
|
|
assert check.additional["source_locations"] == [
|
|
"band_aider/loop/alpha_loop.py"
|
|
]
|
|
assert check.additional["reason"] == [
|
|
"def owner_band_aider_loop_alpha_loop_py():"
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("retained", (False, True))
|
|
def test_file_blame_carries_historical_file_evidence_through_pairs_and_retry(
|
|
memory_root: MemoryPath, retained: bool, monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Earlier admission supplies current file evidence to every pair without expanding its selectable paths."""
|
|
|
|
from band_aider.context import file_blame_check
|
|
|
|
first, second, editor = map(_source, ("src/first.py", "src/second.py", "src/editor.py"))
|
|
test = _source("tests/test_edit.py", ContentType.Test)
|
|
support = _source("tests/test_contract.py", ContentType.Test)
|
|
manager = _BlameFileManager(memory_root)
|
|
manager.src_classes, manager.test_classes = [first, second, editor], [test, support]
|
|
manager.src_index = {first.location: 0, second.location: 1, editor.location: 2}
|
|
manager.test_index = {test.location: 0, support.location: 1}
|
|
first_body = "def first():\n return False\n"
|
|
second_body = "def second():\n return False\n"
|
|
editor_body = "def edit():\n commit_all()\n validate_all()\n"
|
|
support_body = "def contract():\n return 'all validations first'\n"
|
|
baseline = "def edit():\n previous_operation()\n"
|
|
patch = "@@ -1,2 +1,3 @@\n- previous_operation()\n+ commit_all()\n+ validate_all()\n"
|
|
manager.repo_exec.write_text(memory_root / first.location, first_body)
|
|
manager.repo_exec.write_text(memory_root / second.location, second_body)
|
|
manager.repo_exec.write_text(memory_root / editor.location, editor_body)
|
|
manager.repo_exec.write_text(memory_root / support.location, support_body)
|
|
manager._pristine_sources = {editor.location: baseline}
|
|
monkeypatch.setattr(manager, "git_diff_for", Mock(side_effect=
|
|
lambda location: patch if manager.repo_location(location) == editor.location else ""))
|
|
graph = Provenance.open(repo_root=memory_root, repo_exec=memory_root.repository)
|
|
inspection = InspectionBacklog(first, "earlier inspection",
|
|
upstream_classes=[editor, support] if retained else [], downstream_classes=[first])
|
|
graph.start(Phase.ALPHA, inspection)
|
|
beta = BetaWork(test, "current test work", downstream_classes=[first, second])
|
|
graph.track_provenance(inspection.message_id, Phase.BETA, beta)
|
|
failure = FailingTest(test=test, error_message="commit preceded validation",
|
|
stack_trace=["src/first.py:1 def first():", "src/second.py:1 def second():"])
|
|
responses: list[object] = [
|
|
{"location": editor.location, "reason": [" commit_all()"],
|
|
"explanation": "The retained file is visible, but is outside this comparison's pair."},
|
|
{"location": support.location, "reason": [" return 'all validations first'"],
|
|
"explanation": "Supporting test context is not an originating failed test."},
|
|
{"location": second.location, "reason": ["def second():"],
|
|
"explanation": "The second candidate advances this comparison."},
|
|
]
|
|
# Only the admitted editor participates in the later pair.
|
|
if retained:
|
|
responses.append({"location": editor.location, "reason": [" commit_all()"],
|
|
"explanation": "This member commits before validation under the current task."})
|
|
client = _QueuedClient(*responses)
|
|
comparisons = Mock(wraps=FileBlameCandidatePrecedenceDecision)
|
|
monkeypatch.setattr(file_blame_check, "FileBlameCandidatePrecedenceDecision", comparisons)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.BETA, beta):
|
|
decision = FileBlameDecision(
|
|
client=client, file_manager=manager, request=_request(memory_root, [first, second]),
|
|
current_cot="Validate every member before committing any.",
|
|
failure_summary="The regression remains unresolved.",
|
|
failed_tests={test.location: failure}, candidates=[first, second], backlog=beta,
|
|
)
|
|
assert decision() is True
|
|
|
|
assert tuple(map(lambda call: call.kwargs["candidates"], comparisons.call_args_list)) == (
|
|
([first, second], [second, editor]) if retained else ([first, second],)
|
|
)
|
|
assert decision.additional["source_locations"] == ([editor.location] if retained else [second.location])
|
|
assert client.responses == []
|
|
assert "must identify one of the supplied candidates" in client.prompts[1]
|
|
assert "must identify one of the supplied candidates" in client.prompts[2]
|
|
# Admission changes visibility, while the same initial pair and its validation remain closed.
|
|
for prompt in client.prompts:
|
|
assert (editor_body in prompt) is retained
|
|
assert (support_body in prompt) is retained
|
|
assert (baseline in prompt) is retained
|
|
assert (patch in prompt) is retained
|
|
assert first_body in prompt
|
|
assert second_body in prompt
|
|
assert "Validate every member before committing any." in prompt
|
|
# Context files appear once even when they later become selectable candidates.
|
|
if retained:
|
|
assert all(prompt.count(f"FILE: {editor.location}\n") == 1 for prompt in client.prompts)
|
|
assert all(prompt.count(f"FILE: {support.location}\n") == 1 for prompt in client.prompts)
|
|
assert "ADMITTED FILE CONTEXT" in client.prompts[0]
|
|
assert "does not expand the selectable production pair" in client.prompts[0]
|
|
|
|
|
|
def test_file_blame_preserves_combined_raw_output_in_candidate_evidence(
|
|
memory_root: MemoryPath,
|
|
) -> None:
|
|
"""Every raw test-failure detail remains visible in the candidate scope."""
|
|
|
|
candidate = _source("band_aider/loop/alpha_loop.py")
|
|
first = FailingTest(
|
|
test=_source("tests/test_alpha.py", ContentType.Test),
|
|
error_message="legacy error",
|
|
stack_trace=["band_aider/loop/alpha_loop.py: owner_band_aider_loop_alpha_loop_py()"],
|
|
raw_output="compact pytest failure",
|
|
)
|
|
second = FailingTest(
|
|
test=_source("tests/test_beta.py", ContentType.Test),
|
|
error_message="second error",
|
|
stack_trace=["second stack"],
|
|
raw_output="second pytest failure",
|
|
)
|
|
client = _QueuedClient({
|
|
"location": candidate.location,
|
|
"reason": ["def owner_band_aider_loop_alpha_loop_py():"],
|
|
"explanation": "The sole candidate contains the originating operation.",
|
|
})
|
|
check = _decision(memory_root, client, [candidate], [first, second])
|
|
|
|
assert "compact pytest failure" in check.prompt
|
|
assert "second pytest failure" in check.prompt
|
|
assert "legacy error" in check.prompt
|
|
assert "second stack" in check.prompt
|
|
assert check() is True
|
|
assert check.additional["source_locations"] == [candidate.location]
|
|
assert "Inspect the sole preselected production candidate" in client.prompts[0]
|
|
|
|
|
|
def test_file_blame_pairwise_retry_keeps_only_the_immediate_error(
|
|
memory_root: MemoryPath,
|
|
) -> None:
|
|
"""Pairwise probabilistic failures retry with full context and no error history."""
|
|
|
|
first = _source("band_aider/judgement/cached_file_patcher.py")
|
|
second = _source("band_aider/context/memento.py")
|
|
failure = FailingTest(
|
|
test=_source("tests/test_cached_file_patcher.py", ContentType.Test),
|
|
error_message="LifecycleError: Cannot access pre after rollback.",
|
|
stack_trace=[
|
|
"band_aider/judgement/cached_file_patcher.py: before = definitions(memento.pre)",
|
|
"band_aider/context/memento.py: lifecycle error",
|
|
],
|
|
raw_output="targeted lifecycle failure",
|
|
)
|
|
wrong_shape = {
|
|
"winner": first.location,
|
|
"reason": ["def owner_band_aider_judgement_cached_file_patcher_py():"],
|
|
"explanation": "wrong response key",
|
|
}
|
|
invented = {
|
|
"location": "band_aider/invented.py",
|
|
"reason": ["def invented():"],
|
|
"explanation": "An invented candidate appears responsible.",
|
|
}
|
|
unsupported_reason = {
|
|
"location": second.location,
|
|
"reason": ["smoke = 'weed'"],
|
|
"explanation": "This unsupported line supposedly violates the COT.",
|
|
}
|
|
valid = {
|
|
"location": second.location,
|
|
"reason": ["def owner_band_aider_context_memento_py():"],
|
|
"explanation": "The lifecycle mutation first violates the current COT.",
|
|
}
|
|
client = _QueuedClient(*(wrong_shape,) * 10, invented, unsupported_reason)
|
|
cloud = _QueuedClient(valid)
|
|
runtime = StabilizationEscalation((client, cloud), check_name="FileBlame", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
check = _decision(memory_root, runtime, [first, second], [failure])
|
|
|
|
assert check() is True
|
|
prompts = client.prompts + cloud.prompts
|
|
assert len(client.prompts) == 12
|
|
assert len(cloud.prompts) == 1
|
|
assert len(prompts) == 13
|
|
assert all("Repair the failure." in prompt for prompt in prompts)
|
|
assert all("targeted lifecycle failure" in prompt for prompt in prompts)
|
|
assert all(
|
|
"== BEGIN RESPONSE VALIDATION ERROR ==" in prompt
|
|
for prompt in prompts[1:]
|
|
)
|
|
assert json.dumps(wrong_shape) in prompts[1]
|
|
assert "band_aider/invented.py" in prompts[11]
|
|
assert json.dumps(wrong_shape) not in prompts[11]
|
|
assert "band_aider/invented.py" not in prompts[12]
|
|
assert "smoke = 'weed'" in prompts[12]
|
|
assert "was not found as a complete line in live code" in prompts[12]
|
|
assert check.additional["source_locations"] == [second.location]
|
|
assert check.additional["reason"] == [
|
|
"def owner_band_aider_context_memento_py():"
|
|
]
|
|
assert check.changes == [(check, True)]
|
|
|
|
|
|
def test_file_blame_malformed_json_retry_retains_only_immediate_response(
|
|
memory_root: MemoryPath,
|
|
) -> None:
|
|
"""Malformed JSON correction retains the current rejected text and full task."""
|
|
|
|
candidate = _source("src/origin.py")
|
|
failure = FailingTest(
|
|
test=_source("tests/test_origin.py", ContentType.Test),
|
|
error_message="origin operation failed",
|
|
stack_trace=["src/origin.py: owner_src_origin_py()"],
|
|
raw_output="current failure output",
|
|
)
|
|
valid = {
|
|
"location": candidate.location,
|
|
"reason": ["def owner_src_origin_py():"],
|
|
"explanation": "The sole source contains the originating violation.",
|
|
}
|
|
client = _QueuedClient(
|
|
*("first malformed output",) * 10,
|
|
"second malformed output",
|
|
"immediate malformed output",
|
|
)
|
|
cloud = _QueuedClient(valid)
|
|
runtime = StabilizationEscalation((client, cloud), check_name="FileBlame", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
check = _decision(memory_root, runtime, [candidate], [failure])
|
|
|
|
assert check() is True
|
|
prompts = client.prompts + cloud.prompts
|
|
assert len(client.prompts) == 12
|
|
assert len(cloud.prompts) == 1
|
|
assert len(prompts) == 13
|
|
assert "RESPONSE VALIDATION ERROR" in prompts[1]
|
|
assert "first malformed output" in prompts[1]
|
|
assert "RESPONSE VALIDATION ERROR" in prompts[11]
|
|
assert "second malformed output" in prompts[11]
|
|
assert "first malformed output" not in prompts[11]
|
|
assert "RESPONSE VALIDATION ERROR" in prompts[12]
|
|
assert "immediate malformed output" in prompts[12]
|
|
assert "first malformed output" not in prompts[12]
|
|
assert "second malformed output" not in prompts[12]
|
|
assert "Repair the failure." in prompts[12]
|
|
assert "current failure output" in prompts[12]
|
|
|
|
|
|
@pytest.mark.parametrize("chained", (False, True))
|
|
def test_file_blame_bubbles_tied_candidates_toward_origin_through_cot_renderer(
|
|
memory_root: MemoryPath, chained: bool, monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Every tied pair advances its second candidate without skipping a comparison."""
|
|
from band_aider.context import file_blame_check
|
|
|
|
comparisons = Mock(wraps=file_blame_check.FileBlameCandidatePrecedenceDecision)
|
|
monkeypatch.setattr(file_blame_check, "FileBlameCandidatePrecedenceDecision", comparisons)
|
|
|
|
outer = _source("src/outer.py")
|
|
middle = _source("src/middle.py")
|
|
origin = _source("src/origin.py")
|
|
# Deliberately scramble caller input; FileBlame must restore stack order.
|
|
candidates = [origin, outer, middle]
|
|
prior = CotTaskRecord(
|
|
task="Preserve the earlier lifecycle rule.",
|
|
location=outer.location,
|
|
root=outer,
|
|
)
|
|
current_cot = "Reject the invalid state before it reaches a catcher."
|
|
trace = ["src/outer.py", "src/middle.py", "src/origin.py"]
|
|
# A caught failure exposes the middle/inner chain before the outer rethrow chain.
|
|
if chained:
|
|
trace = ["src/middle.py", "src/origin.py",
|
|
"During handling of the above exception, another exception occurred:",
|
|
"src/outer.py", "src/middle.py"]
|
|
failure = FailingTest(
|
|
test=_source("tests/test_origin.py", ContentType.Test),
|
|
error_message="invalid state reached outer catcher",
|
|
stack_trace=trace,
|
|
raw_output="causal failure evidence",
|
|
)
|
|
client = _QueuedClient(
|
|
"KEEP: COT-0000\nREASON: the lifecycle rule constrains ownership",
|
|
{
|
|
"location": middle.location,
|
|
"reason": ["def owner_src_middle_py():"],
|
|
"explanation": "Tied evidence advances the second candidate toward origin.",
|
|
},
|
|
"KEEP: COT-0000\nREASON: the lifecycle rule constrains ownership",
|
|
{
|
|
"location": origin.location,
|
|
"reason": ["def owner_src_origin_py():"],
|
|
"explanation": "Tied evidence again advances the second candidate toward origin.",
|
|
},
|
|
)
|
|
check = _decision(
|
|
memory_root,
|
|
client,
|
|
candidates,
|
|
[failure],
|
|
current_cot=current_cot,
|
|
previous_cot_records=(prior,),
|
|
)
|
|
|
|
assert check() is True
|
|
assert check.additional["source_locations"] == [origin.location]
|
|
assert tuple(map(lambda call: call.kwargs["candidates"], comparisons.call_args_list)) == (
|
|
[outer, middle], [middle, origin],
|
|
)
|
|
assert len(client.prompts) == 4
|
|
assert "CONTEXTUAL COT SELECTION" in client.prompts[0]
|
|
assert "CONTEXTUAL COT SELECTION" in client.prompts[2]
|
|
assert current_cot in client.prompts[0]
|
|
assert current_cot in client.prompts[2]
|
|
assert "COT AUTHORITY FOR THIS LLM TURN" in client.prompts[1]
|
|
assert "COT AUTHORITY FOR THIS LLM TURN" in client.prompts[3]
|
|
assert outer.location in client.prompts[1]
|
|
assert middle.location in client.prompts[1]
|
|
assert middle.location in client.prompts[3]
|
|
assert origin.location in client.prompts[3]
|
|
assert client.responses == []
|
|
assert "outer-caller-to-failure-source direction" in client.prompts[1]
|
|
assert "choose the second listed candidate" in client.prompts[1]
|
|
|
|
|
|
def test_file_blame_precedence_receives_localized_future_guidance(
|
|
memory_root: MemoryPath,
|
|
) -> None:
|
|
"""A revived observation can change the winner through localized guidance."""
|
|
|
|
candidates = [_source("src/outer.py"), _source("src/origin.py")]
|
|
failure = FailingTest(
|
|
test=_source("tests/test_origin.py", ContentType.Test),
|
|
error_message="origin operation reached the outer catcher",
|
|
stack_trace=["src/outer.py", "src/origin.py: owner_src_origin_py()"],
|
|
raw_output="later rollback evidence",
|
|
)
|
|
guidance = FutureGuidance(
|
|
origin_message_id="later-alpha",
|
|
addressed_message_id="revived-beta",
|
|
message="The outer attribution reached a dead end; reconsider the origin.",
|
|
justification="The origin creates the invalid state before the wrapper sees it.",
|
|
)
|
|
localized = (
|
|
"Prefer src/origin.py when its cited operation creates the invalid state "
|
|
"before src/outer.py observes it."
|
|
)
|
|
client = _QueuedClient(
|
|
{
|
|
"applicable": True,
|
|
"reason": "The revived evidence changes causal precedence in this candidate set.",
|
|
},
|
|
{"localized_prompt": localized},
|
|
{
|
|
"location": candidates[1].location,
|
|
"reason": ["def owner_src_origin_py():"],
|
|
"explanation": "The origin violates the COT before the outer catcher runs.",
|
|
},
|
|
)
|
|
check = _decision(
|
|
memory_root,
|
|
client,
|
|
candidates,
|
|
[failure],
|
|
future_guidance=(guidance,),
|
|
)
|
|
|
|
assert check() is True
|
|
assert check.additional["source_locations"] == [candidates[1].location]
|
|
assert localized in client.prompts[2]
|
|
assert guidance.message not in client.prompts[2]
|
|
assert "CURRENT BACKLOG AND LOCALIZED FUTURE GUIDANCE" in client.prompts[2]
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_pytest_failure_trace_triangulates_the_originating_source_end_to_end(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
) -> None:
|
|
"""A real pytest failure supplies source lines through candidates and blame."""
|
|
|
|
root = tmp_path_factory.mktemp("file-blame-triangulation")
|
|
files = {
|
|
"src/__init__.py": "",
|
|
"src/origin.py": (
|
|
"def originate():\n"
|
|
" raise ValueError(\"TRIANGULATED_FAILURE\")\n"
|
|
),
|
|
"src/outer.py": (
|
|
"from src.origin import originate\n\n"
|
|
"def outer():\n"
|
|
" return originate()\n"
|
|
),
|
|
"tests/test_failure.py": (
|
|
"from src.outer import outer\n\n"
|
|
"def test_failure():\n"
|
|
" outer()\n"
|
|
),
|
|
}
|
|
for location, content in files.items():
|
|
path = root / location
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
candidates = [_source("src/outer.py"), _source("src/origin.py")]
|
|
test = _source("tests/test_failure.py", ContentType.Test)
|
|
client = _QueuedClient({
|
|
"location": candidates[1].location,
|
|
"reason": ['raise ValueError("TRIANGULATED_FAILURE")'],
|
|
"explanation": "The origin creates the invalid state before outer propagates it.",
|
|
})
|
|
manager = _HostBlameFileManager(root, client)
|
|
manager.src_classes = candidates
|
|
manager.test_classes = [test]
|
|
manager.src_index = {
|
|
source.location: index
|
|
for index, source in enumerate(candidates)
|
|
}
|
|
manager.test_index = {test.location: 0}
|
|
|
|
result = PytestRunnerStrategy(manager, manager.repo_exec).run_tests_for([test])
|
|
|
|
assert not result.passed
|
|
assert "--tb=long" in result.command
|
|
assert any(
|
|
'raise ValueError("TRIANGULATED_FAILURE")' in line
|
|
for line in result.stack_trace
|
|
)
|
|
assert any("return originate()" in line for line in result.stack_trace)
|
|
failure = FailingTest(
|
|
test=test,
|
|
error_message=f"Test command failed with return code {result.returncode}.",
|
|
stack_trace=list(result.stack_trace),
|
|
raw_output=result.output,
|
|
)
|
|
trace_candidates = (
|
|
failure_source_candidate_selection_strategy_factory.candidates_for_failure(
|
|
scope=FailureCandidateScope.SOURCE,
|
|
file_manager=manager,
|
|
failure=failure,
|
|
)
|
|
)
|
|
assert [source.location for source in trace_candidates.candidates] == [
|
|
"src/outer.py",
|
|
"src/origin.py",
|
|
]
|
|
|
|
blame = FileBlameDecision(
|
|
client=client,
|
|
file_manager=manager,
|
|
request=AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[],
|
|
readonly_files=[root / source.location for source in trace_candidates.candidates],
|
|
),
|
|
current_cot="Raise the failure at its originating source.",
|
|
failure_summary="The real targeted pytest execution failed.",
|
|
failed_tests={test.location: failure},
|
|
candidates=list(trace_candidates.candidates),
|
|
)
|
|
|
|
assert blame() is True
|
|
assert blame.additional["source_locations"] == ["src/origin.py"]
|
|
assert blame.additional["reason"] == [
|
|
'raise ValueError("TRIANGULATED_FAILURE")'
|
|
]
|
|
assert len(str(blame.additional["failure_code_fragment"])) > 5
|
|
|
|
|
|
@pytest.mark.parametrize("production_caller", [False, True])
|
|
def test_removed_abstract_override_routes_to_blame_guidance_and_real_repair(tmp_path_factory: pytest.TempPathFactory, production_caller: bool) -> None:
|
|
"""An uninstantiable AlphaLoop reaches its production owner and passes only after repair."""
|
|
|
|
root = tmp_path_factory.mktemp("deleted-alpha-ownership")
|
|
removed = " def _prepare_work(self):\n return None\n"
|
|
original = "from src.base import ProtoLoop\nclass AlphaLoop(ProtoLoop):\n" + removed
|
|
files = {
|
|
"src/__init__.py": "", "tests/__init__.py": "",
|
|
"src/base.py": "from abc import ABC, abstractmethod\nclass ProtoLoop(ABC):\n @abstractmethod\n def _prepare_work(self): pass\n",
|
|
"src/alpha.py": original,
|
|
"src/caller.py": "from src.alpha import AlphaLoop\ndef create():\n return AlphaLoop()\n",
|
|
"tests/helper.py": "from src.alpha import AlphaLoop\n",
|
|
"tests/test_alpha.py": "from tests.helper import *\ndef test_alpha():\n assert AlphaLoop()._prepare_work() is None\n",
|
|
}
|
|
if production_caller:
|
|
files["tests/test_alpha.py"] = "from src.caller import create\ndef test_alpha():\n assert create()._prepare_work() is None\n"
|
|
for location, content in files.items():
|
|
path = root / location
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
source = _source("src/alpha.py")
|
|
base = _source("src/base.py")
|
|
caller = _source("src/caller.py")
|
|
test = _source("tests/test_alpha.py", ContentType.Test)
|
|
helper = _source("tests/helper.py", ContentType.Test)
|
|
citation = "<DELETED>" + removed + "</DELETED>"
|
|
client = _RemovedTargetClient(
|
|
{"outcome": "source", "reason": "The goal requires a concrete AlphaLoop.", "additional": {"test_work_needed": ""}},
|
|
{"location": source.location, "reason": [citation], "explanation": "The removed override makes construction abstract before the test can execute its assertion."},
|
|
{"work_needed": "Restore the concrete override, then rerun test_alpha.", "reason": [citation]},
|
|
)
|
|
manager = _HostBlameFileManager(root, client)
|
|
manager.prompt = "AlphaLoop must remain concrete and _prepare_work must return None."
|
|
manager.src_classes = [source, base, caller]
|
|
manager.src_index = {source.location: 0, base.location: 1, caller.location: 2}
|
|
manager.test_classes = [test, helper]
|
|
manager.test_index = {test.location: 0, helper.location: 1}
|
|
maintainer = PythonAstDependencyGraphMaintainer(repo_exec=manager.repo_exec)
|
|
manager.dependency_graph_maintainer = maintainer
|
|
maintainer.build(manager.src_classes, manager.test_classes, root, manager.src_root, manager.test_root)
|
|
manager.capture_frozen_once()
|
|
(root / source.location).write_text("from src.base import ProtoLoop\nclass AlphaLoop(ProtoLoop):\n pass\n", encoding="utf-8")
|
|
maintainer.build(manager.src_classes, manager.test_classes, root, manager.src_root, manager.test_root)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
result = runner.run_tests_for([test])
|
|
assert not result.passed
|
|
assert "abstract" in result.output and "_prepare_work" in result.output
|
|
failure = FailingTest(test=test, error_message="AlphaLoop is abstract", stack_trace=list(result.stack_trace), raw_output=result.output, test_name="test_alpha")
|
|
request = AiderEditRequest(authoritative_prompt=manager.prompt, editable_files=[root / source.location], readonly_files=[root / test.location])
|
|
story = TargetedFailureStory(testing_item=None, raw_output=result.output, work_needed="Repair construction",
|
|
candidate_sources=[source, base], failing=failure, failure_phase="post_alpha", focused_test=test,
|
|
original_cot=manager.prompt, change_type=ChangeType.Update)
|
|
route = TargetedAlphaOrBetaRoutingDecision(client=client, file_manager=manager, request=request, story=story)
|
|
assert route()
|
|
assert route.outcome == "source"
|
|
assert source.location in route.candidate_source_locations
|
|
blame = FileBlameDecision(client=client, file_manager=manager, request=request, current_cot=manager.prompt,
|
|
failure_summary=failure.error_message, failed_tests={test.location: failure}, candidates=[source, base])
|
|
assert blame()
|
|
assert blame.additional["source_locations"] == [source.location]
|
|
assert blame.additional["reason"] == [citation]
|
|
assert blame.additional["failure_code_fragment"] == ""
|
|
guidance = SourceRemediationGuidanceCheck(client=client, file_manager=manager, request=request,
|
|
original_cot=manager.prompt, source=source, failure_evidence=result.output, precedent_diffs="")
|
|
assert guidance()
|
|
assert "Restore the concrete override" in guidance.additional["work_needed"]
|
|
assert all("DELETED" in prompt or "is_removed" in prompt for prompt in client.prompts)
|
|
(root / source.location).write_text(original, encoding="utf-8")
|
|
assert runner.run_tests_for([test]).passed
|
|
|
|
|
|
@pytest.mark.parametrize("restored", [False, True])
|
|
def test_traced_blame_requires_exact_removed_target_evidence(memory_root: MemoryPath, restored: bool) -> None:
|
|
"""A traced candidate may cite its removed target, but unrelated or restored code is rejected."""
|
|
|
|
source = _source("src/alpha.py")
|
|
previous = "class AlphaLoop:\n def _prepare_work(self): return None\n def unrelated(self): return 3\n"
|
|
manager = _BlameFileManager(memory_root)
|
|
manager.src_classes = [source]
|
|
manager.repo_exec.write_text(memory_root / source.location, previous)
|
|
manager.dependency_graph_maintainer = PythonAstDependencyGraphMaintainer(repo_exec=manager.repo_exec)
|
|
manager.capture_frozen_once()
|
|
manager.repo_exec.write_text(memory_root / source.location, previous if restored else "class AlphaLoop: pass\n")
|
|
call = CallTraceNode(source.location, "AlphaLoop._prepare_work", line=2, end_line=2, is_removed=True,
|
|
caller_location="tests/test_alpha.py", call_line=3, import_reference=CallImport("tests/test_alpha.py", 1, "from src.alpha import AlphaLoop", "src.alpha.AlphaLoop"))
|
|
unrelated = "<DELETED> def unrelated(self): return 3\n</DELETED>"
|
|
deleted = "<DELETED> def _prepare_work(self): return None\n</DELETED>"
|
|
client = _QueuedClient(
|
|
{"location": source.location, "reason": [unrelated], "explanation": "Unsupported attribution."},
|
|
{"location": source.location, "reason": [deleted], "explanation": "The required override was removed."},
|
|
)
|
|
request = AiderEditRequest(authoritative_prompt="Keep AlphaLoop concrete.", editable_files=[], readonly_files=[],
|
|
failure_stack_trace=["src/alpha.py:1: in call"])
|
|
decision = FileBlameCandidatePrecedenceDecision(client=client, file_manager=manager, request=request,
|
|
current_cot=request.authoritative_prompt, candidates=[source], failure_evidence="src/alpha.py:1: in call",
|
|
prompt="Attribute the missing implementation.", failure_calls=[call])
|
|
|
|
response, invalid = decision._decide()
|
|
assert not invalid.valid
|
|
assert "pristine declaration" in invalid.stringify()
|
|
_, validated = decision._decide(response, invalid)
|
|
assert validated.valid is not restored
|
|
assert decision.additional.get("failure_code_fragment", "") == ""
|
|
|
|
|
|
def test_host_snapshot_and_deletion_preserve_crlf(tmp_path_factory: pytest.TempPathFactory) -> None:
|
|
"""Universal-newline text reads cannot invent a deletion or normalize the cited baseline."""
|
|
|
|
root = tmp_path_factory.mktemp("deleted-crlf")
|
|
source = _source("src/subject.py")
|
|
path = root / source.location
|
|
path.parent.mkdir()
|
|
original = "def subject():\r\n\treturn 1\r\n"
|
|
path.write_bytes(original.encode("utf-8"))
|
|
manager = _HostBlameFileManager(root, _QueuedClient())
|
|
manager.src_classes = [source]
|
|
manager.dependency_graph_maintainer = PythonAstDependencyGraphMaintainer(repo_exec=manager.repo_exec)
|
|
manager.capture_frozen_once()
|
|
gate = CodeBasedReasoning(file_manager=manager, files=[source], baseline=manager.pristine_sources)
|
|
|
|
assert manager.pristine_sources[source.location] == original
|
|
assert not gate(["<DELETED>" + original + "</DELETED>"])
|
|
path.write_bytes(b"def subject():\n return 2\n")
|
|
assert gate(["<DELETED>" + original + "</DELETED>"])
|
|
assert not gate(["<DELETED>" + original.replace("\r\n", "\n") + "</DELETED>"])
|
|
|
|
|
|
@pytest.mark.parametrize("explicit_cause", (False, True))
|
|
def test_real_pytest_chained_failure_normalizes_outer_test_before_support(
|
|
tmp_path_factory: pytest.TempPathFactory, explicit_cause: bool,
|
|
) -> None:
|
|
"""Actual pytest rendering preserves the test identity across a failing handler."""
|
|
from band_aider.judgement.stack_trace_order_strategy import StackTraceFrame
|
|
from band_aider.strategies.python.python_stack_trace_order_strategy import PythonStackTraceOrderStrategy
|
|
|
|
root = tmp_path_factory.mktemp("chained-traceback")
|
|
cause = " from error" if explicit_cause else ""
|
|
files = {
|
|
"src/__init__.py": "",
|
|
"tests/__init__.py": "",
|
|
"tests/support.py": "def read():\n raise KeyError('snapshot')\n",
|
|
"src/patcher.py": (
|
|
"from tests.support import read\n\n"
|
|
"def apply():\n"
|
|
" try:\n"
|
|
" read()\n"
|
|
" except KeyError as error:\n"
|
|
f" raise RuntimeError('rollback'){cause}\n"
|
|
),
|
|
"tests/test_patcher.py": (
|
|
"from src.patcher import apply\n\n"
|
|
"def test_apply():\n"
|
|
" apply()\n"
|
|
),
|
|
}
|
|
# Use the concrete runner so traceback formatting is independent of the adapter test.
|
|
for location, content in files.items():
|
|
path = root / location
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content, encoding="utf-8")
|
|
source = _source("src/patcher.py")
|
|
test = _source("tests/test_patcher.py", ContentType.Test)
|
|
support = _source("tests/support.py", ContentType.Test)
|
|
manager = _HostBlameFileManager(root, _QueuedClient())
|
|
manager.src_classes = [source]
|
|
manager.test_classes = [support, test]
|
|
result = PytestRunnerStrategy(manager, manager.repo_exec).run_tests_for([test])
|
|
failure = FailingTest.from_result(test, result)
|
|
captured = list(failure.stack_trace)
|
|
adapter = PythonStackTraceOrderStrategy()
|
|
|
|
frames = adapter.frames_for_failure(file_manager=manager, failure=failure)
|
|
candidates = adapter.normalize(file_manager=manager,
|
|
candidates=[support, source, test], stack_trace=failure.stack_trace)
|
|
|
|
assert result.returncode == 1
|
|
assert result.output.index("tests/support.py:2") < result.output.index("tests/test_patcher.py:4")
|
|
assert frames == (
|
|
StackTraceFrame(test.location, None, 4),
|
|
StackTraceFrame(source.location, None, 7),
|
|
StackTraceFrame(source.location, None, 5),
|
|
StackTraceFrame(support.location, None, 2),
|
|
)
|
|
assert candidates == [test, source, support]
|
|
assert failure.test_name == "test_apply"
|
|
assert failure.stack_trace == captured
|
|
assert failure.raw_output == result.output
|
|
|
|
|
|
@pytest.mark.parametrize("paired", [False, True])
|
|
def test_blame_retains_whole_sources_on_rejection(
|
|
memory_root: MemoryPath,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
paired: bool,
|
|
) -> None:
|
|
"""A selector rejecting all callables cannot remove evidence from any blame attempt."""
|
|
|
|
manager = _BlameFileManager(memory_root)
|
|
culprit = _source("src/acceptance.py")
|
|
caller = _source("src/caller.py")
|
|
candidates = [caller, culprit] if paired else [culprit]
|
|
manager.src_classes = candidates
|
|
manager.src_index = dict(map(lambda entry: (entry[1].location, entry[0]), enumerate(candidates)))
|
|
culprit_code = (
|
|
"class Acceptance:\n"
|
|
" def inspect(self, source):\n"
|
|
" descriptor = source.fresh()\n"
|
|
" return descriptor.pre\n"
|
|
"\n"
|
|
" def unrelated(self):\n"
|
|
" return 'also retained'\n"
|
|
)
|
|
caller_code = "def invoke(source):\n return Acceptance().inspect(source)\n"
|
|
manager.repo_exec.write_text(memory_root / culprit.location, culprit_code)
|
|
manager.repo_exec.write_text(memory_root / caller.location, caller_code)
|
|
caller_diff = "@@ -1,2 +1,2 @@\n- return old()\n+ return Acceptance().inspect(source)\n"
|
|
monkeypatch.setattr(
|
|
manager, "git_diff_for",
|
|
Mock(side_effect=lambda location: caller_diff if manager.repo_location(location) == caller.location else ""),
|
|
)
|
|
failure = FailingTest(
|
|
test=_source("tests/test_acceptance.py", ContentType.Test),
|
|
error_message="LifecycleError: descriptor must be active",
|
|
stack_trace=["src/caller.py:2", "src/acceptance.py:4", " return descriptor.pre"],
|
|
)
|
|
client = _QueuedClient(
|
|
{"location": culprit.location, "reason": ["return fabricated.pre"],
|
|
"explanation": "This citation is invalid and must trigger correction."},
|
|
{"location": culprit.location, "reason": ["return descriptor.pre"],
|
|
"explanation": "The fresh descriptor is read before activation."},
|
|
)
|
|
# Reproduce an admitted NO_KEEP without allowing it to weaken blame evidence.
|
|
selector = Mock(return_value=set())
|
|
monkeypatch.setattr(ContextualFileRenderer, "_select_page", selector)
|
|
check = FileBlameDecision(
|
|
client=client, file_manager=manager,
|
|
request=_request(memory_root, candidates),
|
|
current_cot="Enter each fresh descriptor before reading its pre property.",
|
|
failure_summary="The acceptance constructor reads an inert descriptor.",
|
|
failed_tests={failure.test.location: failure}, candidates=candidates,
|
|
)
|
|
|
|
assert check() is True
|
|
assert check.additional["source_locations"] == [culprit.location]
|
|
assert len(client.prompts) == 2
|
|
# Initial evidence and both comparison attempts retain every candidate body.
|
|
for prompt in (check.prompt, *client.prompts):
|
|
assert culprit_code in prompt
|
|
assert "GIT DIFF\n[no git diff]" in prompt
|
|
assert "⋮" not in prompt
|
|
# The two-candidate case must retain the caller as well as the chosen culprit.
|
|
if paired:
|
|
assert caller_code in prompt
|
|
assert caller_diff in prompt
|
|
selector.assert_not_called()
|
|
|
|
|
|
def test_blame_excludes_untraced_source_with_valid_citations(memory_root: MemoryPath) -> None:
|
|
"""Valid code and failure-text overlap cannot make an untraced file eligible."""
|
|
|
|
owner = _source("src/owner.py")
|
|
untraced = _source("src/untraced.py")
|
|
failure = FailingTest(
|
|
test=_source("tests/test_owner.py", ContentType.Test),
|
|
error_message="src/untraced.py is mentioned outside the traceback",
|
|
stack_trace=["src/owner.py:2", " return True"],
|
|
)
|
|
# Both files contain this exact line; only the owner's path appears in the trace.
|
|
client = _QueuedClient(
|
|
{"location": untraced.location, "reason": ["return True"],
|
|
"explanation": "The untraced file also contains the failing statement."},
|
|
{"location": owner.location, "reason": ["return True"],
|
|
"explanation": "The traced source owns the failing statement."},
|
|
)
|
|
check = _decision(memory_root, client, [untraced, owner], [failure])
|
|
|
|
assert check() is True
|
|
assert check.additional["source_locations"] == [owner.location]
|
|
assert len(client.prompts) == 2
|
|
assert "Inspect the sole preselected production candidate" in client.prompts[0]
|
|
assert "must identify one of the supplied candidates: ['src/owner.py']" in client.prompts[1]
|
|
assert all(f"FILE: {untraced.location}" not in prompt for prompt in (check.prompt, *client.prompts))
|
|
|
|
|
|
def test_blame_limits_expanded_pool_to_trace_comparisons(memory_root: MemoryPath) -> None:
|
|
"""The observed 36-source pool requires only the three trace-supported comparisons."""
|
|
|
|
traced = list(map(_source, (
|
|
"band_aider/loop/alpha_loop.py",
|
|
"band_aider/judgement/alpha_edit_acceptance_check.py",
|
|
"band_aider/judgement/prompt_formatter.py",
|
|
"band_aider/context/memento.py",
|
|
)))
|
|
untraced = list(map(lambda number: _source(f"src/context_{number}.py"), range(32)))
|
|
candidates = [*untraced, *reversed(traced)]
|
|
failure = FailingTest(
|
|
test=_source("tests/test_alpha.py", ContentType.Test),
|
|
error_message="LifecycleError",
|
|
stack_trace=[*map(lambda source: f"{source.location}:2", traced), " return True"],
|
|
)
|
|
winner = traced[1]
|
|
response = {"location": winner.location, "reason": ["return True"],
|
|
"explanation": "The acceptance constructor first violates the lifecycle contract."}
|
|
client = _QueuedClient(response, response, response)
|
|
check = _decision(memory_root, client, candidates, [failure])
|
|
|
|
assert check.prompt.count("BEGIN FAILURE MATCHED SOURCE") == 4
|
|
assert check() is True
|
|
assert check.additional["source_locations"] == [winner.location]
|
|
assert len(client.prompts) == 3
|
|
assert f"FILE: {traced[0].location}" in client.prompts[0]
|
|
assert f"FILE: {winner.location}" in client.prompts[0]
|
|
assert f"FILE: {traced[2].location}" in client.prompts[1]
|
|
assert f"FILE: {traced[3].location}" in client.prompts[2]
|
|
assert all("FILE: src/context_" not in prompt for prompt in (check.prompt, *client.prompts))
|
|
|
|
|
|
def test_blame_rejects_empty_admission_without_model_call(memory_root: MemoryPath) -> None:
|
|
"""An untraced supplied pool cannot bypass the existing nonempty-candidate precondition."""
|
|
|
|
client = _QueuedClient()
|
|
failure = FailingTest(
|
|
test=_source("tests/test_owner.py", ContentType.Test),
|
|
error_message="src/untraced.py failed",
|
|
stack_trace=["external/source.py:3"],
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="trace-supported managed production candidate"):
|
|
_decision(memory_root, client, [_source("src/untraced.py")], [failure])
|
|
|
|
assert client.prompts == []
|