379 lines
17 KiB
Python
379 lines
17 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Verify current source after a recovered Alpha item declines further editing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
import json
|
|
import logging
|
|
import pickle
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.context.beta_workflow_story import AcceptedSourceStory, DirectTestCandidate
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.context_factory import AlphaContextReassembly, LoadedCheckpoint
|
|
from band_aider.context.memento import Memento
|
|
from band_aider.context import provenance
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.provenance import FutureGuidance, Phase, Provenance
|
|
from band_aider.context.working_backlog import ContributorBacklog, FailedQAItem, InspectionBacklog, TestingBacklog
|
|
from band_aider.judgement.code_edit_or_leave_it import CodeEditOrLeaveIt
|
|
from band_aider.judgement.code_editor import CodeEditorResult
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from tests.band_aider_loop_impl import (
|
|
BackLog,
|
|
EditingSourceWorker,
|
|
FakeClient,
|
|
PassingRunner,
|
|
TestAlphaContext,
|
|
_alpha_context,
|
|
_alpha_gate,
|
|
_file_manager,
|
|
_language_strategy_set,
|
|
)
|
|
|
|
|
|
class ResumeClient(FakeClient):
|
|
"""Supply independent inspection and edit votes while recording which gates ran."""
|
|
|
|
def __init__(self, *, inspect_work: bool, edit_work: bool = False) -> None:
|
|
"""Store the scenario's two votes and initialize observed judgement calls."""
|
|
|
|
super().__init__()
|
|
self.inspect_work = inspect_work
|
|
self.edit_work = edit_work
|
|
self.inspections = 0
|
|
self.edit_judgements = 0
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Answer actual judgement prompts without invoking a model transport."""
|
|
|
|
# Inspection can close the source assignment before the edit gate is reached.
|
|
if "INSPECTION WORK GUIDANCE" in prompt:
|
|
self.inspections += 1
|
|
return json.dumps({
|
|
"work_required": self.inspect_work,
|
|
"evidence_type": "cot" if self.inspect_work else "none",
|
|
"semantic_divergence": "none",
|
|
"reason": list(self.acceptance_reasons),
|
|
"work_needed": "1. Change foo to return 2. 2. Verify its direct test." if self.inspect_work else "",
|
|
})
|
|
# The final edit vote is independent of the earlier inspection assignment.
|
|
if "EDIT WORTHINESS JUDGEMENT" in prompt:
|
|
self.edit_judgements += 1
|
|
return json.dumps({
|
|
"needs_edit": self.edit_work,
|
|
"evidence_type": "cot" if self.edit_work else "none",
|
|
"semantic_divergence": "none",
|
|
"reason": "The focused source needs the COT change." if self.edit_work else "The focused source already satisfies the COT.",
|
|
})
|
|
return super().complete(prompt, logger)
|
|
|
|
|
|
class PostflightRunner(PassingRunner):
|
|
"""Record real postflight test requests and return the supplied observation."""
|
|
|
|
def __init__(self, result: TestRunResult) -> None:
|
|
"""Store one stationary test outcome and an independent execution history."""
|
|
|
|
self.result = result
|
|
self.calls: list[tuple[str, ...]] = []
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Observe the selected direct tests at the runner boundary."""
|
|
|
|
assert max_failures == 1
|
|
assert not test_names
|
|
self.calls.append(tuple(map(lambda test: test.location, test_files)))
|
|
return self.result
|
|
|
|
|
|
class SourceFailureBacklog(ContributorBacklog):
|
|
"""Attribute fixed failed-test evidence to its source, optionally interrupting routing."""
|
|
|
|
interrupt_routing: bool = False
|
|
|
|
def _route_failed_direct_candidate(
|
|
self,
|
|
story: AcceptedSourceStory,
|
|
candidate: DirectTestCandidate,
|
|
assignment: str,
|
|
*,
|
|
regression_star: bool,
|
|
future_guidance: tuple[FutureGuidance, ...] = (),
|
|
logger: logging.Logger | None = None,
|
|
) -> list[FailedQAItem | TestingBacklog]:
|
|
"""Keep test execution and queue admission real while fixing failure ownership."""
|
|
|
|
assert candidate.failing is not None
|
|
# Model a process stop after a failed test but before its owner is queued.
|
|
if self.interrupt_routing:
|
|
raise KeyboardInterrupt("stopped during failure attribution")
|
|
return [FailedQAItem(
|
|
focused_spotlight=story.current_source,
|
|
summary=assignment,
|
|
downstream_classes=list(story.downstream_classes),
|
|
upstream_classes=list(story.upstream_classes),
|
|
original_cot=story.original_cot,
|
|
failed_test=candidate.failing,
|
|
future_guidance=future_guidance,
|
|
)]
|
|
|
|
|
|
def _restore_alpha(context: TestAlphaContext, checkpoint: Path) -> TestAlphaContext:
|
|
"""Reassemble serialized Alpha state with the production runtime rebinding policy."""
|
|
|
|
saved = pickle.loads(context.file_manager.repo_exec.read_bytes(checkpoint))
|
|
resumed = TestAlphaContext.__new__(TestAlphaContext)
|
|
AlphaContextReassembly().thaw(
|
|
resumed,
|
|
LoadedCheckpoint(version=saved["checkpoint_version"], payload=saved["payload"]),
|
|
{"file_manager": context.file_manager, "backlog": context.backlog},
|
|
)
|
|
resumed.test_runner_strategy = context.test_runner_strategy
|
|
resumed.new_files_callback = context.new_files_callback
|
|
resumed.topology_summary = context.topology_summary
|
|
return resumed
|
|
|
|
|
|
@pytest.mark.parametrize("inspect_work", [False, True], ids=["inspection-no-work", "editor-leave"])
|
|
@pytest.mark.parametrize("passed", [False, True], ids=["partial-source-fails", "source-passes"])
|
|
def test_resumed_alpha_noop_verifies_current_source(inspect_work: bool, passed: bool) -> None:
|
|
"""A resumed no-op must test current bytes and durably route either observed outcome."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
client = ResumeClient(inspect_work=inspect_work)
|
|
manager.client = client
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
raw = BackLog()
|
|
context = _alpha_context(manager, raw, source, test)
|
|
runner = PostflightRunner(TestRunResult("pytest", TestRunOutcome.Success if passed else TestRunOutcome.Failure, 0 if passed else 1, "passed" if passed else "assert 2 == 1"))
|
|
context.backlog = SourceFailureBacklog(
|
|
backlog=raw, get_tests=lambda sources: [test], file_manager=manager,
|
|
parser=manager.parser, language_strategy_set=_language_strategy_set(manager, runner=runner),
|
|
)
|
|
assert context.move_inbound_from_backlog() is not None
|
|
checkpoint = context.persist()
|
|
snapshot = source.set_or_get_memento(manager)
|
|
current_source = "def foo():\n return 1\n" if passed else "def foo():\n return 2\n"
|
|
manager.repo_exec.write_text(manager.source_path(source.location), current_source)
|
|
snapshot.commit()
|
|
assert Memento.current(manager.source_path(source.location), manager.repo_exec) is None
|
|
assert manager.git_diff_for(source.location) == ""
|
|
assert not manager.src_dirty
|
|
resumed = _restore_alpha(context, checkpoint)
|
|
reported_changes: list[tuple[ClassMeta, ...]] = []
|
|
resumed.alpha_postflight_callback = reported_changes.append
|
|
editor = EditingSourceWorker()
|
|
gate = CodeEditOrLeaveIt(
|
|
inner=editor, client=client, repo_root=manager.repo_root,
|
|
prompt_formatter_builder=manager.prompt_formatter_builder, repo_exec=manager.repo_exec,
|
|
)
|
|
gate.file_manager = manager
|
|
|
|
AlphaLoop(gate=_alpha_gate(resumed), context=resumed, aider_worker=gate).run()
|
|
|
|
assert client.inspections == 1
|
|
assert client.edit_judgements == int(inspect_work)
|
|
assert editor.requests == []
|
|
assert runner.calls == [(test.location,)]
|
|
assert manager.repo_exec.read_text(manager.source_path(source.location)) == current_source
|
|
assert resumed.active_inbound is None
|
|
assert not resumed.outbound_queue
|
|
assert not reported_changes
|
|
# Passing verification hands a regression star to Beta; failure returns source remediation.
|
|
if passed:
|
|
assert not raw.remediation
|
|
assert len(raw.testing) == 1
|
|
assert raw.testing[0].focused_spotlight == test
|
|
else:
|
|
assert not raw.testing
|
|
assert len(raw.remediation) == 1
|
|
failure = raw.remediation[0].failed_test
|
|
assert failure is not None
|
|
assert failure.raw_output == "assert 2 == 1"
|
|
assert raw.remediation[0].focused_spotlight == source
|
|
|
|
|
|
@pytest.mark.parametrize("inspect_work", [False, True], ids=["inspection-no-work", "editor-leave"])
|
|
def test_fresh_alpha_noop_does_not_run_postflight(inspect_work: bool) -> None:
|
|
"""An ordinary no-op must retain the existing zero-test behavior."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
client = ResumeClient(inspect_work=inspect_work)
|
|
manager.client = client
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
raw = BackLog()
|
|
context = _alpha_context(manager, raw, source, test)
|
|
runner = PostflightRunner(TestRunResult("pytest", TestRunOutcome.Success, 0, "passed"))
|
|
context.backlog = ContributorBacklog(
|
|
backlog=raw, get_tests=lambda sources: [test], file_manager=manager,
|
|
parser=manager.parser, language_strategy_set=_language_strategy_set(manager, runner=runner),
|
|
)
|
|
editor = EditingSourceWorker()
|
|
gate = CodeEditOrLeaveIt(
|
|
inner=editor, client=client, repo_root=manager.repo_root,
|
|
prompt_formatter_builder=manager.prompt_formatter_builder, repo_exec=manager.repo_exec,
|
|
)
|
|
gate.file_manager = manager
|
|
|
|
AlphaLoop(gate=_alpha_gate(context), context=context, aider_worker=gate).run()
|
|
|
|
assert client.inspections == 1
|
|
assert client.edit_judgements == int(inspect_work)
|
|
assert not editor.requests
|
|
assert not runner.calls
|
|
assert not raw.testing
|
|
assert not raw.remediation
|
|
|
|
|
|
def test_resumed_matching_history_without_leave_still_verifies(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Matching provenance cannot suppress resumed verification when the editor did not return LEAVE."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
failure = FailingTest(test, ["old()"], "AttributeError: old")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
earlier = graph.start(Phase.ALPHA, FailedQAItem(source, "repair", failed_test=failure))
|
|
graph.track_provenance(earlier, Phase.BETA, TestingBacklog(test, "verify", downstream_classes=[source]))
|
|
item = FailedQAItem(source, "repair again", failed_test=failure)
|
|
graph.start(Phase.ALPHA, item)
|
|
raw = BackLog(completion={})
|
|
raw.queue_remediation(item)
|
|
context = _alpha_context(manager, raw, source, test)
|
|
runner = PostflightRunner(TestRunResult("pytest", TestRunOutcome.Success, 0, "passed"))
|
|
context.backlog = SourceFailureBacklog(
|
|
backlog=raw, get_tests=lambda sources: [test], file_manager=manager,
|
|
parser=manager.parser, language_strategy_set=_language_strategy_set(manager, runner=runner),
|
|
)
|
|
with provenance.bind(graph):
|
|
assert context.move_inbound_from_backlog() is not None
|
|
resumed = _restore_alpha(context, context.persist())
|
|
editor = EditingSourceWorker()
|
|
edit = Mock(return_value=CodeEditorResult([], "The editor left source bytes unchanged.", []))
|
|
monkeypatch.setattr(editor, "edit", edit)
|
|
|
|
with provenance.bind(graph):
|
|
AlphaLoop(gate=_alpha_gate(resumed), context=resumed, aider_worker=editor, provenance=graph).run()
|
|
|
|
edit.assert_called_once()
|
|
assert runner.calls == [(test.location,)]
|
|
assert resumed.active_inbound is None
|
|
assert len(raw.testing) == 1
|
|
assert not raw.remediation
|
|
|
|
|
|
def test_resume_verification_is_consumed_before_next_inspection() -> None:
|
|
"""Only the recovered item gets extra verification during an otherwise ordinary Alpha pass."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
manager.repo_exec.write_text(manager.src_root / "bar.py", "def bar():\n return 1\n")
|
|
manager.resync()
|
|
client = ResumeClient(inspect_work=False)
|
|
client.acceptance_reasons = ("return 1",)
|
|
manager.client = client
|
|
source = manager.require_src("src/foo.py")
|
|
next_source = manager.require_src("src/bar.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
raw = BackLog()
|
|
context = _alpha_context(manager, raw, source, test)
|
|
runner = PostflightRunner(TestRunResult("pytest", TestRunOutcome.Success, 0, "passed"))
|
|
context.backlog = ContributorBacklog(
|
|
backlog=raw, get_tests=lambda sources: [test], file_manager=manager,
|
|
parser=manager.parser, language_strategy_set=_language_strategy_set(manager, runner=runner),
|
|
)
|
|
assert context.move_inbound_from_backlog() is not None
|
|
raw.queue_inspection(InspectionBacklog(next_source, "inspect next source"))
|
|
resumed = _restore_alpha(context, context.persist())
|
|
|
|
AlphaLoop(gate=_alpha_gate(resumed), context=resumed).run()
|
|
|
|
assert runner.calls == [(test.location,)]
|
|
assert len(raw.testing) == 1
|
|
assert not raw.inspection
|
|
assert raw.alpha_completed(next_source)
|
|
assert resumed.active_inbound is None
|
|
|
|
|
|
def test_resumed_alpha_edit_runs_postflight_once() -> None:
|
|
"""A resumed actual edit uses the ordinary changed-source postflight exactly once."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
client = ResumeClient(inspect_work=True, edit_work=True)
|
|
manager.client = client
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
raw = BackLog()
|
|
context = _alpha_context(manager, raw, source, test)
|
|
runner = PostflightRunner(TestRunResult("pytest", TestRunOutcome.Success, 0, "passed"))
|
|
context.backlog = ContributorBacklog(
|
|
backlog=raw, get_tests=lambda sources: [test], file_manager=manager,
|
|
parser=manager.parser, language_strategy_set=_language_strategy_set(manager, runner=runner),
|
|
)
|
|
assert context.move_inbound_from_backlog() is not None
|
|
resumed = _restore_alpha(context, context.persist())
|
|
editor = EditingSourceWorker()
|
|
gate = CodeEditOrLeaveIt(
|
|
inner=editor, client=client, repo_root=manager.repo_root,
|
|
prompt_formatter_builder=manager.prompt_formatter_builder, repo_exec=manager.repo_exec,
|
|
)
|
|
gate.file_manager = manager
|
|
|
|
AlphaLoop(gate=_alpha_gate(resumed), context=resumed, aider_worker=gate).run()
|
|
|
|
assert len(editor.requests) == 1
|
|
assert manager.repo_exec.read_text(manager.source_path(source.location)) == "def foo():\n return 2\n"
|
|
assert runner.calls == [(test.location,)]
|
|
assert len(raw.testing) == 1
|
|
|
|
|
|
def test_interrupted_resume_postflight_retries_before_releasing_source() -> None:
|
|
"""Another stop during attribution must retain the source until remediation is durable."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
manager.client = ResumeClient(inspect_work=False)
|
|
source = manager.require_src("src/foo.py")
|
|
test = manager.require_test("tests/test_foo.py")
|
|
raw = BackLog()
|
|
context = _alpha_context(manager, raw, source, test)
|
|
runner = PostflightRunner(TestRunResult("pytest", TestRunOutcome.Failure, 1, "assert 2 == 1"))
|
|
contributor = SourceFailureBacklog(
|
|
backlog=raw, get_tests=lambda sources: [test], file_manager=manager,
|
|
parser=manager.parser, language_strategy_set=_language_strategy_set(manager, runner=runner),
|
|
)
|
|
contributor.interrupt_routing = True
|
|
context.backlog = contributor
|
|
assert context.move_inbound_from_backlog() is not None
|
|
resumed = _restore_alpha(context, context.persist())
|
|
|
|
with pytest.raises(KeyboardInterrupt, match="stopped during failure attribution"):
|
|
AlphaLoop(gate=_alpha_gate(resumed), context=resumed).run()
|
|
|
|
assert resumed.active_inbound is not None
|
|
assert not raw.remediation
|
|
checkpoints = manager.repo_exec.list_files(manager.repo_root / ".band-aider-chkpts", ["TestAlphaContext-*.pkl"], recursive=False)
|
|
restarted = _restore_alpha(resumed, max(checkpoints))
|
|
contributor.interrupt_routing = False
|
|
AlphaLoop(gate=_alpha_gate(restarted), context=restarted).run()
|
|
|
|
assert runner.calls == [(test.location,), (test.location,)]
|
|
assert restarted.active_inbound is None
|
|
assert not restarted.outbound_queue
|
|
assert len(raw.remediation) == 1
|
|
assert raw.remediation[0].focused_spotlight == source
|