335 lines
19 KiB
Python
335 lines
19 KiB
Python
"""Distinguish a completed Beta obligation from the next failure in its file."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import fields, replace
|
|
import logging
|
|
import subprocess
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.context import provenance, resume_state, working_backlog
|
|
from band_aider.context.beta_context import BetaContext
|
|
from band_aider.context.beta_workflow_judgement import BetaWorkflowJudgementService
|
|
from band_aider.context.context_factory import context_factory
|
|
from band_aider.context.memento import Memento
|
|
from band_aider.context.provenance import FutureGuidance, Phase, Provenance
|
|
from band_aider.context.working_backlog import BackLog, ChangeType, FailedQAItem, QABacklog, TestingBacklog
|
|
from band_aider.judgement.code_editor import CodeEditorResult, EditorRecord
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.pytest_runner_strategy import PytestRunnerStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome, TestRunResult
|
|
from band_aider.loop.beta_loop import BetaLoop, TestingItemResult
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
from tests.band_aider import test_beta_prompt_refinement_boundary as boundary
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _CommittingEditor(boundary._RecordingEditor):
|
|
"""Supply the editor's committed-byte contract; Beta must preserve that result.
|
|
|
|
``editor.edit(request)`` retains the candidate and removes its owned snapshot.
|
|
"""
|
|
|
|
def __init__(self, manager: boundary._MemoryFileManager) -> None:
|
|
"""Own the in-memory editor and its successful Work Log."""
|
|
super().__init__("Updated the original assertion.")
|
|
self.manager = manager
|
|
self.record = EditorRecord(self.summary, "assert False", "assert True")
|
|
|
|
def edit(self, request: AiderEditRequest, logger: logging.Logger | None = None) -> CodeEditorResult:
|
|
"""Commit a selected-file edit before returning to Beta's post-edit check."""
|
|
self.requests.append(request)
|
|
path = request.editable_files[0]
|
|
snapshot = Memento.first_available(path, self.manager.repo_exec).capture()
|
|
self.manager.repo_exec.write_text(path, self.manager.repo_exec.read_text(path).replace("assert False", "assert True"))
|
|
snapshot.commit()
|
|
return CodeEditorResult([path], self.summary, [*request.editor_records, self.record])
|
|
|
|
|
|
@pytest.mark.parametrize("owner", ("test", "source", "contradiction"))
|
|
@pytest.mark.parametrize("interruption", ("routing", "flush"))
|
|
def test_new_failure_survives_routing_interruption_without_reopening_completed_work(
|
|
monkeypatch: pytest.MonkeyPatch, owner: str, interruption: str,
|
|
) -> None:
|
|
"""New ownership routing preserves the old commit, copied context and its own durable ID."""
|
|
manager = boundary._MemoryFileManager(
|
|
sources={"src/value.py": "value = 1\n", "src/helper.py": "helper = 2\n", "src/fresh.py": "fresh = 3\n",
|
|
"src/historical.py": "historical = 4\n"},
|
|
tests={"tests/test_value.py": (
|
|
"class TestValue:\n"
|
|
" def test_original(self):\n assert False\n"
|
|
" def test_next(self):\n assert 1 == 2\n"
|
|
)},
|
|
)
|
|
source, helper, fresh, historical = manager.src_classes
|
|
[test] = manager.test_classes
|
|
prior_record = EditorRecord("Earlier setup edit", "before", "after")
|
|
guidance = FutureGuidance("historical-origin", "historical-recipient", "Keep the contract", "Prior observation")
|
|
old = TestingBacklog(
|
|
focused_spotlight=test, summary="REGRESSION STAR", upstream_classes=[helper],
|
|
downstream_classes=[source], future_guidance=(guidance,), editor_records=[prior_record],
|
|
original_cot="Preserve value behavior", previous_beta_guidance="Repair original assertion",
|
|
change_type=ChangeType.Create, failure_test_name="TestValue.test_original",
|
|
failure_error_message="old failure", failure_stack_trace=["old frame"],
|
|
failure_raw_output="old output", failure_outcome=TestRunOutcome.Failure,
|
|
failure_run_info="old command", failure_context_sources=[helper],
|
|
)
|
|
service = Mock(spec=BetaWorkflowJudgementService)
|
|
service.refine_beta_instruction.return_value = "Repair the original assertion."
|
|
service.route_targeted_failure.return_value = boundary._RoutingCheck(outcome="test")
|
|
flush = QABacklog.flush_outbound
|
|
# Interrupt either before judgment or after the new work has been durably staged.
|
|
if interruption == "routing":
|
|
service.route_targeted_failure.side_effect = OSError("routing transport unavailable")
|
|
else:
|
|
monkeypatch.setattr(QABacklog, "flush_outbound", Mock(side_effect=OSError("flush unavailable")))
|
|
backlog = BackLog(completion={}, testing=[old], judgement_service=service)
|
|
qa = QABacklog(backlog, manager.client, manager, Mock(spec=CodeParser))
|
|
context = BetaContext(goal="Preserve value behavior", classes=[], file_manager=manager, backlog=qa)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
context.test_runner_strategy = runner
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
ancestor = TestingBacklog(test, "Prior admitted context", upstream_classes=[historical])
|
|
graph.start(Phase.BETA, ancestor)
|
|
graph.track_provenance(ancestor.message_id, Phase.BETA, old)
|
|
gate = Mock(spec=LoopGate)
|
|
gate.may_enter.return_value = True
|
|
editor = _CommittingEditor(manager)
|
|
loop = BetaLoop(gate, context, editor, provenance=graph)
|
|
resume_state.configure_resume_settings(hydrate=True, checkpoint_dir=manager.repo_root / ".checkpoints", repo_exec=manager.repo_exec)
|
|
monkeypatch.setattr(working_backlog, "curate_failure_context_sources", Mock(return_value=[fresh, helper]))
|
|
assert isinstance(manager.repo_exec, MemoryRepoExec)
|
|
output = "FAILED tests/test_value.py::TestValue::test_next[case] - AssertionError: next failure\n"
|
|
manager.repo_exec.queue_result(subprocess.CompletedProcess(["pytest"], 1, output, ""))
|
|
|
|
with pytest.raises(OSError, match="unavailable"):
|
|
loop.run()
|
|
|
|
assert context.post_edit is not None
|
|
story = context.post_edit
|
|
new = story.testing_item
|
|
assert new is not None
|
|
completed_tests = backlog.completion[source.location].tests
|
|
assert completed_tests is not None and completed_tests[test.location].completed
|
|
assert old.failure_test_name == ""
|
|
assert new.message_id and new.message_id != old.message_id
|
|
assert graph.messages[new.message_id].parent_id == old.message_id
|
|
assert set(graph.messages) == {ancestor.message_id, old.message_id, new.message_id}
|
|
assert context.active_inbound is not None and context.active_inbound.value is new
|
|
assert backlog.testing == backlog.remediation == ()
|
|
assert len(editor.requests) == 1
|
|
assert "assert True" in manager.repo_exec.read_text(manager.test_path(test.location))
|
|
assert Memento.current(manager.test_path(test.location), manager.repo_exec) is None
|
|
assert manager.repo_exec.commands[0][-1] == str(manager.test_path(test.location))
|
|
assert "--maxfail=1" in manager.repo_exec.commands[0]
|
|
assert story.previous_beta_guidance == ""
|
|
assert story.beta_editor_summary == editor.summary
|
|
assert story.failure_phase == "after_beta"
|
|
assert set(map(lambda candidate: candidate.location, story.candidate_sources)) == {
|
|
source.location, helper.location, fresh.location, historical.location,
|
|
}
|
|
expected = {
|
|
"focused_spotlight": test, "summary": "FAILED CANDIDATE",
|
|
"upstream_classes": [helper], "downstream_classes": [source],
|
|
"message_id": new.message_id, "future_guidance": (guidance,),
|
|
"editor_records": [prior_record, editor.record], "original_cot": old.original_cot,
|
|
"previous_beta_guidance": "", "change_type": ChangeType.Update,
|
|
"failure_error_message": story.failing.error_message,
|
|
"failure_stack_trace": list(story.failing.stack_trace), "failure_raw_output": output.rstrip(),
|
|
"failure_test_name": "TestValue.test_next[case]", "failure_outcome": TestRunOutcome.Failure,
|
|
"failure_run_info": story.failing.run_info, "failure_context_sources": [helper, source, historical, fresh],
|
|
"last_exam": False,
|
|
}
|
|
assert {field.name: getattr(new, field.name) for field in fields(TestingBacklog)} == expected
|
|
assert new.upstream_classes is not old.upstream_classes
|
|
assert new.downstream_classes is not old.downstream_classes
|
|
assert new.editor_records is not old.editor_records
|
|
|
|
restored = context_factory.thaw(BetaContext, file_manager=manager, backlog=qa)
|
|
assert restored is not None and restored.post_edit is not None
|
|
restored.test_runner_strategy = runner
|
|
resumed_graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=True)
|
|
service.route_targeted_failure.side_effect = None
|
|
monkeypatch.setattr(QABacklog, "flush_outbound", flush)
|
|
service.route_targeted_failure.return_value = boundary._RoutingCheck(outcome="test" if owner == "test" else "source")
|
|
# FileBlame retains the shared source/contradiction contract; only its final owner is scripted.
|
|
if owner == "source":
|
|
repair = FailedQAItem(
|
|
focused_spotlight=source, summary="Repair next failure", downstream_classes=[source],
|
|
original_cot=old.original_cot, failed_test=story.failing,
|
|
failure_test_name=new.failure_test_name, future_guidance=(guidance,),
|
|
editor_records=list(new.editor_records),
|
|
)
|
|
report = Mock(return_value=[repair])
|
|
else:
|
|
report = Mock(return_value=[replace(new, previous_beta_guidance="Confirmed test ownership")])
|
|
manager.repo_exec.queue_result(subprocess.CompletedProcess(["pytest"], 0, "2 passed", ""))
|
|
monkeypatch.setattr(QABacklog, "report_remediation", report)
|
|
|
|
BetaLoop(gate, restored, editor, provenance=resumed_graph).run()
|
|
|
|
assert restored.post_edit is None and restored.active_inbound is None
|
|
assert "assert True" in manager.repo_exec.read_text(manager.test_path(test.location))
|
|
assert Memento.current(manager.test_path(test.location), manager.repo_exec) is None
|
|
assert service.route_collection_failure.call_count == 0
|
|
assert service.route_targeted_failure.call_args.kwargs["story"].testing_item.message_id == new.message_id
|
|
# Confirmed production returns before a second editor call; Beta repairs only the new identity.
|
|
if owner == "source":
|
|
assert len(editor.requests) == 1
|
|
assert len(backlog.remediation) == 1 and not backlog.testing
|
|
failed_test = backlog.remediation[0].failed_test
|
|
assert failed_test is not None and failed_test.test_name == "TestValue.test_next[case]"
|
|
assert backlog.remediation[0].future_guidance == ()
|
|
assert backlog.remediation[0].editor_records == []
|
|
assert resumed_graph.messages[backlog.remediation[0].message_id].parent_id == new.message_id
|
|
else:
|
|
assert len(editor.requests) == 2
|
|
assert editor.requests[1].failure_test_name == "TestValue.test_next[case]"
|
|
assert backlog.testing == backlog.remediation == ()
|
|
assert set(resumed_graph.messages) == {ancestor.message_id, old.message_id, new.message_id}
|
|
completed_tests = backlog.completion[source.location].tests
|
|
assert completed_tests is not None and completed_tests[test.location].completed
|
|
# The ordinary test verdict avoids production attribution altogether.
|
|
if owner == "test":
|
|
report.assert_not_called()
|
|
else:
|
|
assert report.call_args.kwargs["testing_item"].message_id == new.message_id
|
|
assert report.call_args.kwargs["failed_test"].test_name == "TestValue.test_next[case]"
|
|
|
|
|
|
@pytest.mark.parametrize("observed", ("pass", "same", "new", "error", "timeout"))
|
|
def test_testing_item_reports_validation_without_creating_or_routing_work(
|
|
monkeypatch: pytest.MonkeyPatch, observed: str,
|
|
) -> None:
|
|
"""A specialized repair reports its result; only its workflow may schedule or complete work."""
|
|
manager = boundary._MemoryFileManager(
|
|
sources={"src/value.py": "value = 1\n"},
|
|
tests={"tests/test_value.py": "def test_original():\n assert False\ndef test_next():\n assert False\n"},
|
|
)
|
|
[test] = manager.test_classes
|
|
[source] = manager.src_classes
|
|
item = TestingBacklog(
|
|
test, "repair", downstream_classes=[source], failure_test_name="test_original",
|
|
failure_error_message="original failure", failure_outcome=TestRunOutcome.Failure,
|
|
)
|
|
service = Mock(spec=BetaWorkflowJudgementService)
|
|
service.refine_beta_instruction.return_value = "Repair the original assertion."
|
|
backlog = BackLog(completion={}, testing=[item], judgement_service=service)
|
|
context = BetaContext(goal="repair", classes=[], file_manager=manager,
|
|
backlog=QABacklog(backlog, manager.client, manager, Mock(spec=CodeParser)))
|
|
outcome = {
|
|
"pass": TestRunOutcome.Success, "same": TestRunOutcome.Failure,
|
|
"new": TestRunOutcome.Failure, "error": TestRunOutcome.Error, "timeout": TestRunOutcome.Timeout,
|
|
}[observed]
|
|
name = "test_original" if observed == "same" else "test_next"
|
|
context.test_runner_strategy = boundary._TestRunner([TestRunResult(
|
|
"whole file", outcome, 0 if observed == "pass" else 1,
|
|
"current execution", test_name=name,
|
|
)])
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
graph.start(Phase.BETA, item)
|
|
loop = BetaLoop(Mock(spec=LoopGate), context, _CommittingEditor(manager), provenance=graph)
|
|
monkeypatch.setattr(working_backlog, "curate_failure_context_sources", Mock(return_value=[source]))
|
|
|
|
with provenance.bind(graph):
|
|
with provenance.item_scope(Phase.BETA, item):
|
|
result = loop._handle_testing_item(item)
|
|
|
|
assert isinstance(result, TestingItemResult)
|
|
assert result.passed is (observed in {"pass", "new"})
|
|
assert result.targeted.outcome is outcome
|
|
assert result.targeted.passed is (observed == "pass")
|
|
assert result.applied.testing_item is item
|
|
assert result.applied.work_needed == "Repair the original assertion."
|
|
assert result.applied.summary == "Updated the original assertion."
|
|
assert (result.new_failure is not None) is (observed == "new")
|
|
# Newly observed failure evidence does not replace or clear the still-owned input ticket.
|
|
if result.new_failure is not None:
|
|
assert result.new_failure.test_name == "test_next"
|
|
assert result.new_failure.raw_output == "current execution"
|
|
assert item.failure_test_name == ("" if observed == "pass" else "test_original")
|
|
assert set(graph.messages) == {item.message_id}
|
|
assert backlog.testing == (item,) and backlog.remediation == ()
|
|
assert context.post_edit is context.active_inbound is None
|
|
assert context.inbound_queue == context.outbound_queue == []
|
|
service.route_targeted_failure.assert_not_called()
|
|
service.route_collection_failure.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", (TestRunOutcome.Failure, TestRunOutcome.Error, TestRunOutcome.Timeout))
|
|
def test_unfinished_failure_releases_claim_without_changing_identity(
|
|
monkeypatch: pytest.MonkeyPatch, outcome: TestRunOutcome,
|
|
) -> None:
|
|
"""Same-name failures and file errors do not become successful old obligations."""
|
|
manager = boundary._MemoryFileManager(
|
|
sources={"src/value.py": "value = 1\n"},
|
|
tests={"tests/test_value.py": "def test_original():\n assert False\n"},
|
|
)
|
|
[test] = manager.test_classes
|
|
[source] = manager.src_classes
|
|
item = TestingBacklog(test, "repair", downstream_classes=[source], failure_test_name="test_original")
|
|
service = Mock(spec=BetaWorkflowJudgementService)
|
|
service.refine_beta_instruction.return_value = "Repair the assertion."
|
|
service.route_targeted_failure.side_effect = OSError("routing interrupted")
|
|
service.route_collection_failure.side_effect = OSError("routing interrupted")
|
|
backlog = BackLog(completion={}, testing=[item], judgement_service=service)
|
|
context = BetaContext(goal="repair", classes=[], file_manager=manager,
|
|
backlog=QABacklog(backlog, manager.client, manager, Mock(spec=CodeParser)))
|
|
context.test_runner_strategy = boundary._TestRunner([TestRunResult(
|
|
"whole file", outcome, 1, "new error text", test_name="test_original" if outcome is TestRunOutcome.Failure else "test_other",
|
|
)])
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
gate = Mock(spec=LoopGate)
|
|
gate.may_enter.return_value = True
|
|
editor = _CommittingEditor(manager)
|
|
monkeypatch.setattr(working_backlog, "curate_failure_context_sources", Mock(return_value=[source]))
|
|
monkeypatch.setattr("band_aider.loop.beta_loop.curate_failure_context_sources", Mock(return_value=[source]))
|
|
monkeypatch.setattr(BetaContext, "persist", Mock())
|
|
|
|
with pytest.raises(OSError, match="routing interrupted"):
|
|
BetaLoop(gate, context, editor, provenance=graph).run()
|
|
|
|
assert context.post_edit is None
|
|
assert context.active_inbound is not None and context.active_inbound.value is item
|
|
assert set(graph.messages) == {item.message_id}
|
|
completed_tests = backlog.completion[source.location].tests
|
|
assert completed_tests is not None and test.location not in completed_tests
|
|
assert item.failure_raw_output == "new error text"
|
|
assert "assert True" in manager.repo_exec.read_text(manager.test_path(test.location))
|
|
assert service.route_targeted_failure.call_count == int(outcome is TestRunOutcome.Failure)
|
|
assert service.route_collection_failure.call_count == int(outcome.is_error)
|
|
|
|
|
|
@pytest.mark.parametrize("code", (0, 5))
|
|
def test_whole_file_success_or_removed_test_completes_without_new_work(
|
|
monkeypatch: pytest.MonkeyPatch, code: int,
|
|
) -> None:
|
|
"""A clean file or confirmed empty file discharges the old name without fabricating a receipt."""
|
|
manager = boundary._MemoryFileManager(tests={"tests/test_value.py": "# old test was removed\n"})
|
|
[test] = manager.test_classes
|
|
item = TestingBacklog(test, "repair", downstream_classes=list(manager.src_classes), failure_test_name="test_removed")
|
|
service = Mock(spec=BetaWorkflowJudgementService)
|
|
service.refine_beta_instruction.return_value = "Preserve the intended removal."
|
|
backlog = BackLog(completion={}, testing=[item], judgement_service=service)
|
|
context = BetaContext(goal="repair", classes=[], file_manager=manager,
|
|
backlog=QABacklog(backlog, manager.client, manager, Mock(spec=CodeParser)))
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
context.test_runner_strategy = runner
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
gate = Mock(spec=LoopGate)
|
|
gate.may_enter.return_value = True
|
|
assert isinstance(manager.repo_exec, MemoryRepoExec)
|
|
manager.repo_exec.queue_result(subprocess.CompletedProcess(["pytest"], code, "passed" if code == 0 else "no tests collected", ""))
|
|
monkeypatch.setattr(BetaContext, "persist", Mock())
|
|
|
|
BetaLoop(gate, context, boundary._RecordingEditor("preserved removal"), provenance=graph).run()
|
|
|
|
assert context.post_edit is None and context.active_inbound is None
|
|
assert item.failure_test_name == ""
|
|
assert backlog.testing == backlog.remediation == ()
|
|
assert set(graph.messages) == {item.message_id}
|
|
service.route_targeted_failure.assert_not_called()
|
|
service.route_collection_failure.assert_not_called()
|