1333 lines
56 KiB
Python
1333 lines
56 KiB
Python
"""Beta refinement boundaries exercised through nominal in-memory collaborators."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
import inspect
|
|
import json
|
|
import logging
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.context.beta_context import BetaContext
|
|
from band_aider.context.beta_workflow_judgement import BetaWorkflowJudgementService
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
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 import provenance
|
|
from band_aider.context.provenance import FutureGuidance, Phase, Provenance
|
|
from band_aider.context.targeted_failure_owner_check import TargetedAlphaOrBetaRoutingDecision
|
|
from band_aider.context.targeted_test_collection_failure_owner_check import (
|
|
CollectionFailureAssignment,
|
|
TargetedCollectionAlphaOrBetaRoutingDecision,
|
|
)
|
|
from band_aider.context.working_backlog import (
|
|
BackLog,
|
|
BacklogLane,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
QABacklog,
|
|
StagedInboundItem,
|
|
TestingBacklog,
|
|
curate_failure_context_sources,
|
|
)
|
|
from band_aider.judgement.beta_contract_guidance_check import BetaContractGuidanceDecision
|
|
from band_aider.judgement.code_edit_or_leave_it import (
|
|
CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN,
|
|
CodeEditLeaveSummary,
|
|
)
|
|
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult, EditorRecord
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import BiDependencyDiscoveryStrategy
|
|
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.stabilization_decision import StabilizationDecision
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck, VibeCodeCheck
|
|
from band_aider.loop.beta_loop import AppliedTestWork, BetaLoop
|
|
from band_aider.loop.exceptions import BetaRemediationRequested, BetaRequeueRequested
|
|
from band_aider.strategies import LanguageStrategySet
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _QueuedClient(OpenAICompatibleClient):
|
|
"""Return queued JSON results without an external endpoint."""
|
|
|
|
def __init__(self, responses: Sequence[object] = ()) -> None:
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
self.responses = [json.dumps(response) for response in responses]
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record and serve one stationary response."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
if not self.responses:
|
|
raise AssertionError("Test setup did not queue a completion response.")
|
|
return self.responses.pop(0)
|
|
|
|
|
|
class _MemoryFileManager(FileManager):
|
|
"""Expose FileManager's API over lexical files only held in memory."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
client: OpenAICompatibleClient | None = None,
|
|
sources: dict[str, str] | None = None,
|
|
tests: dict[str, str] | None = None,
|
|
diffs: dict[str, str] | None = None,
|
|
) -> None:
|
|
"""Seed explicit source/test exhibits without calling a host filesystem."""
|
|
|
|
self._pristine_sources = {}
|
|
|
|
self.repo_root = Path("/testbed")
|
|
self.src_root = self.repo_root / "src"
|
|
self.test_root = self.repo_root / "tests"
|
|
self.repo_exec = MemoryRepoExec(self.repo_root)
|
|
self.client = client or _QueuedClient(({"include": False},))
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
self.actor = "test"
|
|
self.prompt = "Original stabilization goal."
|
|
self.upstream_file_manager: FileManager | None = None
|
|
self.src_dirty: dict[str, bool] = {}
|
|
self.test_dirty: dict[str, bool] = {}
|
|
self.src_classes: list[ClassMeta] = []
|
|
self.test_classes: list[ClassMeta] = []
|
|
self.src_index: dict[str, int] = {}
|
|
self.test_index: dict[str, int] = {}
|
|
self._diffs = dict(diffs or {})
|
|
self.refreshed_summaries: list[str] = []
|
|
for location, source in (sources or {}).items():
|
|
self.repo_exec.write_text(self.source_path(location), source)
|
|
self._add_source(location)
|
|
for location, source in (tests or {}).items():
|
|
self.repo_exec.write_text(self.test_path(location), source)
|
|
self._add_test(location)
|
|
|
|
def _repository_path(self, location: str | Path) -> Path:
|
|
path = Path(location)
|
|
return path if path.is_absolute() else self.repo_root / path
|
|
|
|
def repo_path(self, location: str | Path) -> Path:
|
|
"""Resolve one lexical repository path."""
|
|
|
|
return self._repository_path(location)
|
|
|
|
def repo_location(self, location: str | Path) -> str:
|
|
"""Return a stable repository-relative path string."""
|
|
|
|
return self._repository_path(location).relative_to(self.repo_root).as_posix()
|
|
|
|
def source_path(self, location: str | Path) -> Path:
|
|
"""Resolve source locations with or without their root prefix."""
|
|
|
|
path = Path(location)
|
|
if path.is_absolute():
|
|
return path
|
|
return self.repo_root / path if path.parts[:1] == ("src",) else self.src_root / path
|
|
|
|
def test_path(self, location: str | Path) -> Path:
|
|
"""Resolve test locations with or without their root prefix."""
|
|
|
|
path = Path(location)
|
|
if path.is_absolute():
|
|
return path
|
|
return self.repo_root / path if path.parts[:1] == ("tests",) else self.test_root / path
|
|
|
|
def source_location(self, location: str | Path) -> str:
|
|
"""Return one repository-relative source location."""
|
|
|
|
return self.repo_location(self.source_path(location))
|
|
|
|
def test_location(self, location: str | Path) -> str:
|
|
"""Return one repository-relative test location."""
|
|
|
|
return self.repo_location(self.test_path(location))
|
|
|
|
def _add_source(self, location: str) -> ClassMeta:
|
|
meta = ClassMeta(self.source_location(location), "source", [], {}, content_type=ContentType.Source)
|
|
self.src_index[meta.location] = len(self.src_classes)
|
|
self.src_classes.append(meta)
|
|
return meta
|
|
|
|
def _add_test(self, location: str) -> ClassMeta:
|
|
meta = ClassMeta(self.test_location(location), "test", [], {}, content_type=ContentType.Test)
|
|
self.test_index[meta.location] = len(self.test_classes)
|
|
self.test_classes.append(meta)
|
|
return meta
|
|
|
|
def current_src(self, location: str) -> ClassMeta | None:
|
|
"""Return one current source record."""
|
|
|
|
index = self.src_index.get(location)
|
|
return self.src_classes[index] if index is not None else None
|
|
|
|
def current_test(self, location: str) -> ClassMeta | None:
|
|
"""Return one current test record."""
|
|
|
|
index = self.test_index.get(location)
|
|
return self.test_classes[index] if index is not None else None
|
|
|
|
def refresh_test(self, location: str, prompt: str, tailing_logs: str = "") -> ClassMeta:
|
|
"""Record refresh evidence and retain one current test record."""
|
|
|
|
del prompt
|
|
self.refreshed_summaries.append(tailing_logs)
|
|
normalized = self.test_location(location)
|
|
current = self.current_test(normalized)
|
|
return current if current is not None else self._add_test(normalized)
|
|
|
|
def git_diff_for(self, location: str | Path) -> str:
|
|
"""Return only test-supplied diff evidence."""
|
|
|
|
return self._diffs.get(self.repo_location(location), "")
|
|
|
|
|
|
class _TestRunner(TestRunnerStrategy):
|
|
"""Record targeted requests and return prearranged results."""
|
|
|
|
def failure_reports(self, test_files, result):
|
|
"""Return the scripted known test scope without running report discovery."""
|
|
return [FailingTest.from_result(test_files[0], result)]
|
|
|
|
def matching_tests(self, test_files: list[ClassMeta], result: TestRunResult) -> dict[str, tuple[str, ...]]:
|
|
"""Reject unexpected final-exam discovery in this isolated runner scenario."""
|
|
raise AssertionError("This scenario does not provide failed-suite method discovery.")
|
|
|
|
def __init__(self, results: Sequence[TestRunResult] = ()) -> None:
|
|
self.results = list(results)
|
|
self.calls: list[tuple[list[ClassMeta], int | None, tuple[str, ...]]] = []
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Record a stationary test invocation."""
|
|
|
|
self.calls.append((test_files, max_failures, tuple(test_names)))
|
|
return self.results.pop(0) if self.results else TestRunResult("pytest", TestRunOutcome.Success, 0, "")
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Return a stationary whole-suite result."""
|
|
|
|
del test_root
|
|
return TestRunResult("pytest", TestRunOutcome.Success, 0, "")
|
|
|
|
|
|
class _RecordingEditor(CodeEditor):
|
|
"""Capture direct edit requests without a file or system edit."""
|
|
|
|
def __init__(self, summary: str, changed_paths: Sequence[Path] = ()) -> None:
|
|
self.summary = summary
|
|
self.changed_paths = list(changed_paths)
|
|
self.requests: list[AiderEditRequest] = []
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Declare this as a root editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Return the configured edit result."""
|
|
|
|
del logger
|
|
self.requests.append(request)
|
|
return CodeEditorResult(list(self.changed_paths), self.summary, [])
|
|
|
|
|
|
class _RoutingCheck(ReasonableVibeCheck):
|
|
"""Represent one pre-established routing outcome."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
outcome: str = "test",
|
|
reason: str = "routing explanation",
|
|
assignments: Sequence[CollectionFailureAssignment] = (),
|
|
) -> None:
|
|
super().__init__(outcome=outcome, reason=reason)
|
|
self.assignments = tuple(assignments)
|
|
|
|
|
|
class _BoundaryRoutingCheck(_RoutingCheck):
|
|
"""Supply the completed targeted-routing result to Beta's explicit boundary."""
|
|
|
|
def __init__(self, decision_name: str) -> None:
|
|
"""Construct one test-owned routed outcome with its request evidence."""
|
|
|
|
super().__init__()
|
|
self._decision_name = decision_name
|
|
self.request = AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[Path("/testbed/tests/test_widget.py")],
|
|
readonly_files=[],
|
|
)
|
|
|
|
def last_result(self) -> dict[str, object]:
|
|
"""Present the admitted routing evidence required by the boundary decision."""
|
|
|
|
return {
|
|
"decision": self._decision_name,
|
|
"outcome": self.outcome,
|
|
"reason": self.reason,
|
|
"additional": {},
|
|
}
|
|
|
|
|
|
class _TestQABacklog(QABacklog):
|
|
"""Supply only the concrete judgement surface required by the Beta scenarios."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
service: BetaWorkflowJudgementService | None = None,
|
|
routing: _RoutingCheck | None = None,
|
|
) -> None:
|
|
self._service = service or BetaWorkflowJudgementService()
|
|
self._routing = routing or _RoutingCheck()
|
|
|
|
@property
|
|
def judgement_service(self) -> BetaWorkflowJudgementService:
|
|
"""Return the scenario-owned refinement service."""
|
|
|
|
return self._service
|
|
|
|
def targeted_alpha_or_beta_routing_check(
|
|
self,
|
|
*,
|
|
story: TargetedFailureStory,
|
|
logger: logging.Logger | None = None,
|
|
) -> ReasonableVibeCheck:
|
|
"""Return the ordinary targeted routing answer."""
|
|
|
|
del story, logger
|
|
return self._routing
|
|
|
|
def targeted_collection_alpha_or_beta_routing_check(
|
|
self,
|
|
*,
|
|
story: TargetedFailureStory,
|
|
logger: logging.Logger | None = None,
|
|
) -> ReasonableVibeCheck:
|
|
"""Return the collection-routing answer."""
|
|
|
|
del story, logger
|
|
return self._routing
|
|
|
|
|
|
def _context(
|
|
manager: _MemoryFileManager,
|
|
*,
|
|
service: BetaWorkflowJudgementService | None = None,
|
|
routing: _RoutingCheck | None = None,
|
|
) -> BetaContext:
|
|
"""Create a real Beta context over test-owned in-memory boundaries."""
|
|
|
|
return BetaContext(
|
|
goal="test beta refinement",
|
|
classes=[],
|
|
file_manager=manager,
|
|
backlog=_TestQABacklog(service=service, routing=routing),
|
|
)
|
|
|
|
|
|
class _BetaProbe(BetaLoop):
|
|
"""Use inherited Beta operations without starting a parent loop."""
|
|
|
|
def __init__(
|
|
self,
|
|
context: BetaContext,
|
|
*,
|
|
runner: TestRunnerStrategy | None = None,
|
|
editor: CodeEditor | None = None,
|
|
) -> None:
|
|
self.context = context
|
|
self.test_runner_strategy = runner or _TestRunner()
|
|
self.aider_worker = editor
|
|
self.frontier_provider = None
|
|
self._blocked_summary = ""
|
|
self._logger_stack = [logging.getLogger(__name__)]
|
|
|
|
|
|
class _RequeueBetaProbe(_BetaProbe):
|
|
"""Own a stationary failure recording boundary for requeue scenarios."""
|
|
|
|
def __init__(
|
|
self,
|
|
context: BetaContext,
|
|
failure_sources: Sequence[ClassMeta],
|
|
routing: _RoutingCheck,
|
|
) -> None:
|
|
super().__init__(context)
|
|
self.failure_sources = list(failure_sources)
|
|
context.backlog = _TestQABacklog(routing=routing)
|
|
|
|
def _record_targeted_failure(
|
|
self,
|
|
testing_item: TestingBacklog,
|
|
result: TestRunResult,
|
|
) -> FailingTest:
|
|
"""Record a complete local test failure without discovery side effects."""
|
|
|
|
focused = testing_item.focused_spotlight
|
|
if focused is None:
|
|
raise AssertionError("Scenario must supply a focused test.")
|
|
failure = FailingTest(
|
|
focused,
|
|
list(result.stack_trace),
|
|
f"Test command failed with return code {result.returncode}.",
|
|
result.output,
|
|
)
|
|
testing_item.record_failure(failure, self.failure_sources)
|
|
return failure
|
|
|
|
|
|
class _GuidanceProbe(BetaContractGuidanceDecision):
|
|
"""Exercise the guidance reducer without irrelevant contextual rendering."""
|
|
|
|
def __init__(self, client: _QueuedClient, *, status: str) -> None:
|
|
manager = _MemoryFileManager(client=client)
|
|
StabilizationDecision.__init__(
|
|
self, client=client, file_manager=manager,
|
|
request=AiderEditRequest(
|
|
authoritative_prompt="", editable_files=[], readonly_files=[],
|
|
),
|
|
)
|
|
self._original_cot = "ORIGINAL REQUIREMENT"
|
|
self._accepted_source_diff = "ACCEPTED DIFF"
|
|
self._previous_beta_guidance = ""
|
|
self._test_status = status
|
|
self.beta_instruction = ""
|
|
|
|
|
|
def _source(location: str = "src/widget.py") -> ClassMeta:
|
|
"""Return a source-owned record."""
|
|
|
|
return ClassMeta(location, "source", [], {}, content_type=ContentType.Source)
|
|
|
|
|
|
def _test(location: str = "tests/test_widget.py") -> ClassMeta:
|
|
"""Return a test-owned record."""
|
|
|
|
return ClassMeta(location, "test", [], {}, content_type=ContentType.Test)
|
|
|
|
|
|
def _testing_item(*, source: ClassMeta | None = None) -> TestingBacklog:
|
|
"""Return a focused Beta ticket."""
|
|
|
|
return TestingBacklog(
|
|
focused_spotlight=_test(),
|
|
summary="ticket provenance",
|
|
downstream_classes=[] if source is None else [source],
|
|
)
|
|
|
|
|
|
def _applied(item: TestingBacklog, summary: str = "editor summary") -> AppliedTestWork:
|
|
"""Return a real immutable Beta edit result."""
|
|
|
|
focused = item.focused_spotlight
|
|
if focused is None:
|
|
raise AssertionError("Scenario must supply a focused test.")
|
|
return AppliedTestWork(item, "instruction", Path("/testbed") / focused.location, (), summary)
|
|
|
|
|
|
def test_beta_retry_has_no_parallel_retry_state() -> None:
|
|
assert "_retry_context" not in BetaLoop.__dataclass_fields__
|
|
assert "retry_context" not in inspect.signature(BetaRequeueRequested).parameters
|
|
|
|
|
|
def test_refined_instruction_is_the_apply_work_input() -> None:
|
|
handle = inspect.signature(BetaLoop._handle_testing_item)
|
|
apply = inspect.signature(BetaLoop._apply_test_work)
|
|
assert list(handle.parameters) == ["self", "testing_item"]
|
|
assert "work_needed" in apply.parameters
|
|
|
|
|
|
def test_beta_refinement_failure_retains_repair_without_stopping(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A refinement failure retains the same work identity for another repair cycle."""
|
|
|
|
context = _context(_MemoryFileManager())
|
|
item = _testing_item()
|
|
item.message_id = "same-beta-work"
|
|
refinement = Mock(side_effect=ValueError("incomplete refinement evidence"))
|
|
monkeypatch.setattr(context.backlog.judgement_service, "refine_beta_instruction", refinement)
|
|
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
_BetaProbe(context)._refine_beta_instruction(item)
|
|
|
|
assert raised.value.testing_item is item
|
|
assert item.message_id == "same-beta-work"
|
|
assert raised.value.stop is False
|
|
assert "incomplete refinement evidence" in raised.value.summary
|
|
|
|
|
|
def test_passing_regression_star_retries_adapt_with_semantic_error_until_add_succeeds() -> None:
|
|
rejected = {
|
|
"asserted_contract": "Existing behavior.",
|
|
"desired_contract": "New behavior.",
|
|
"divergence": "Missing distinct regression.",
|
|
"contract_similarity": 0.80,
|
|
"action": "adapt",
|
|
"beta_instruction": "Adapt the existing passing test.",
|
|
}
|
|
valid = {
|
|
**rejected,
|
|
"action": "add",
|
|
"beta_instruction": "1. Add a new test method.\n2. Preserve the passing test.",
|
|
}
|
|
client = _QueuedClient((rejected, rejected, valid))
|
|
check = _GuidanceProbe(client, status="PASSING")
|
|
check.prompt = "beta refinement prompt"
|
|
|
|
assert check() is True
|
|
assert len(client.prompts) == 3
|
|
assert client.prompts[0] == check.prompt
|
|
assert client.prompts[1].startswith(f"{check.prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert client.prompts[1] == client.prompts[2]
|
|
assert "Missing or regression-star coverage requires add." in client.prompts[1]
|
|
assert json.dumps(rejected) in client.prompts[1]
|
|
assert check.beta_instruction.startswith("1. Add a new test method")
|
|
|
|
|
|
def test_refinement_prompt_prohibits_raw_input_copy() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="requirement",
|
|
accepted_source_diff="diff",
|
|
focused_context="test context",
|
|
assignment_type="REGRESSION STAR",
|
|
test_status="PASSING",
|
|
failure_evidence="",
|
|
)
|
|
assert "Do not copy the ORIGINAL STABILIZATION GOAL" in prompt
|
|
assert "beta_instruction" in prompt
|
|
|
|
|
|
def test_passing_regression_star_prompt_requires_a_new_test_method() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="requirement",
|
|
accepted_source_diff="diff",
|
|
focused_context="test context",
|
|
assignment_type="REGRESSION STAR",
|
|
test_status="PASSING",
|
|
failure_evidence="",
|
|
)
|
|
assert 'ASSIGNMENT TYPE is REGRESSION STAR and FOCUSED TEST STATUS is PASSING, action must be "add"' in prompt
|
|
assert '"add" means create a new test method' in prompt
|
|
assert 'Do not return "adapt"' in prompt
|
|
assert "Existing passing tests are evidence and must be preserved" in prompt
|
|
|
|
|
|
@pytest.mark.parametrize(("similarity", "action"), [(0.20, "delete"), (0.80, "adapt")])
|
|
def test_failed_test_delete_and_adapt_predicates_remain_unchanged(
|
|
similarity: float,
|
|
action: str,
|
|
) -> None:
|
|
response = {
|
|
"asserted_contract": "Prior behavior.",
|
|
"desired_contract": "Changed behavior.",
|
|
"divergence": "Selected treatment.",
|
|
"contract_similarity": similarity,
|
|
"action": action,
|
|
"beta_instruction": "Treat the failed test according to similarity.",
|
|
}
|
|
decision = _GuidanceProbe(_QueuedClient((response,)), status="FAILED")
|
|
_, validation = decision._decide()
|
|
assert validation.valid
|
|
assert decision.additional["action"] == action
|
|
|
|
|
|
def test_refinement_prompt_separates_predecessor_from_assignment_and_failure() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="goal",
|
|
accepted_source_diff="diff",
|
|
focused_context="test context",
|
|
assignment_type="ticket provenance",
|
|
previous_beta_guidance="P0 concrete instruction",
|
|
test_status="FAILED",
|
|
failure_evidence="F0 observed failure",
|
|
)
|
|
assert "ASSIGNMENT TYPE\nticket provenance" in prompt
|
|
assert "PREVIOUS BETA INSTRUCTION\nP0 concrete instruction" in prompt
|
|
assert "FAILURE EVIDENCE\nF0 observed failure" in prompt
|
|
|
|
|
|
def test_testing_backlog_predecessor_defaults_empty() -> None:
|
|
item = TestingBacklog(focused_spotlight=None, summary="ticket provenance")
|
|
assert item.previous_beta_guidance == ""
|
|
|
|
|
|
def test_beta_targeted_run_requests_one_failure() -> None:
|
|
runner = _TestRunner()
|
|
beta = _BetaProbe(_context(_MemoryFileManager()), runner=runner)
|
|
focused = _test()
|
|
beta._run_targeted_tests(TestingBacklog(summary="ticket provenance", focused_spotlight=focused))
|
|
assert runner.calls == [([focused], 1, ())]
|
|
|
|
|
|
def test_targeted_failure_story_uses_executed_beta_instruction() -> None:
|
|
source = _source()
|
|
manager = _MemoryFileManager(
|
|
sources={source.location: "value = 1\n"},
|
|
tests={"tests/test_widget.py": ""},
|
|
)
|
|
item = _testing_item(source=source)
|
|
beta = _BetaProbe(_context(manager))
|
|
result = TestRunResult("pytest", TestRunOutcome.Failure, 1, "failure output", ["failure trace"])
|
|
failing = FailingTest(_test(), ["failure trace"], "failure", "failure output")
|
|
story = beta._targeted_failure_story(item, "P0 concrete instruction", _applied(item), result, failing)
|
|
assert story.work_needed == "P0 concrete instruction"
|
|
assert story.previous_beta_guidance == "P0 concrete instruction"
|
|
assert isinstance(story.testing_item, TestingBacklog)
|
|
assert story.testing_item.summary == "ticket provenance"
|
|
|
|
|
|
def test_beta_requeue_preserves_predecessor_without_routing_guidance() -> None:
|
|
"""A retry preserves identity and carries the latest abandoned edit into its next instruction."""
|
|
|
|
source = _source()
|
|
item = _testing_item(source=source)
|
|
item.message_id = "revived-beta"
|
|
item.editor_records = [EditorRecord("retained setup", "old setup", "new setup")]
|
|
rejection = "Production caller blocks migration.\nRolled-back candidate diff:\n-setup_old\n+setup_active"
|
|
future_guidance = FutureGuidance(
|
|
origin_message_id="future-observation",
|
|
addressed_message_id="revived-beta",
|
|
message="Reconsider the originating source.",
|
|
justification="The later observation invalidated the outer attribution.",
|
|
)
|
|
item.future_guidance = (future_guidance,)
|
|
beta = _RequeueBetaProbe(_context(_MemoryFileManager()), [source], _RoutingCheck())
|
|
result = TestRunResult("pytest", TestRunOutcome.Failure, 1, "F0 output", ["F0 trace"])
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
beta._raise_for_targeted_test_failure(item, "P0 concrete instruction", result, _applied(item, rejection))
|
|
retry = raised.value.testing_item
|
|
assert retry.previous_beta_guidance == "P0 concrete instruction\n\n" + rejection
|
|
assert retry.message_id == item.message_id
|
|
assert retry.failure_error_message == "Test command failed with return code 1."
|
|
assert retry.failure_stack_trace == ["F0 trace"]
|
|
assert retry.failure_context_sources == [source]
|
|
assert retry.future_guidance == (future_guidance,)
|
|
assert retry.editor_records == item.editor_records
|
|
assert retry.editor_records is not item.editor_records
|
|
with pytest.raises(BetaRequeueRequested) as raised_again:
|
|
beta._raise_for_targeted_test_failure(retry, "P1 concrete instruction", result, _applied(retry))
|
|
assert raised_again.value.testing_item.previous_beta_guidance == "P1 concrete instruction\n\neditor summary"
|
|
assert rejection not in raised_again.value.testing_item.previous_beta_guidance
|
|
assert raised_again.value.testing_item.message_id == item.message_id
|
|
assert raised_again.value.testing_item.future_guidance == (future_guidance,)
|
|
assert raised_again.value.testing_item.editor_records == item.editor_records
|
|
|
|
|
|
def test_consolidated_beta_requeue_does_not_persist_routing_prose() -> None:
|
|
source = _source()
|
|
assignment = CollectionFailureAssignment(
|
|
"test",
|
|
"R consolidated routing explanation",
|
|
test_work_needed="Q routing instruction",
|
|
)
|
|
beta = _RequeueBetaProbe(
|
|
_context(_MemoryFileManager()),
|
|
[source],
|
|
_RoutingCheck(assignments=(assignment,)),
|
|
)
|
|
item = _testing_item(source=source)
|
|
item.message_id = "collection-beta"
|
|
item.editor_records = [EditorRecord("retained setup", "old setup", "new setup")]
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
beta._raise_for_targeted_test_failure(
|
|
item,
|
|
"P0 concrete instruction",
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 1, "F0 output", ["F0 trace"]),
|
|
_applied(item),
|
|
)
|
|
retry = raised.value.testing_item
|
|
assert retry.summary == "ticket provenance"
|
|
assert retry.previous_beta_guidance == "P0 concrete instruction\n\neditor summary"
|
|
assert retry.message_id == item.message_id
|
|
assert retry.editor_records == item.editor_records
|
|
assert "R consolidated routing explanation" not in retry.summary
|
|
assert "Q routing instruction" not in retry.previous_beta_guidance
|
|
|
|
|
|
@pytest.mark.parametrize("failure_outcome", (TestRunOutcome.Error, TestRunOutcome.Timeout))
|
|
@pytest.mark.parametrize("owner", ("source", "test"))
|
|
def test_beta_handoff_preserves_selected_owner_with_both_assignment_kinds(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
failure_outcome: TestRunOutcome,
|
|
owner: str,
|
|
) -> None:
|
|
"""Postflight transfers the selected owner even when both kinds of evidence are retained."""
|
|
|
|
source = _source()
|
|
manager = _MemoryFileManager(
|
|
sources={source.location: "def widget():\n return 1\n"},
|
|
tests={"tests/test_widget.py": "def test_widget():\n assert False\n"},
|
|
)
|
|
item = _testing_item(source=source)
|
|
item.message_id = "same-beta-work"
|
|
assignments = (
|
|
CollectionFailureAssignment(
|
|
outcome="test", reason="A test expectation also needs migration.",
|
|
test_work_needed="Migrate the expectation without weakening the contract.",
|
|
),
|
|
CollectionFailureAssignment(
|
|
outcome="source", reason="Production violates the required candidate contract.",
|
|
source_clues=({"location": source.location, "reason": "Candidate state was restored before validation."},),
|
|
),
|
|
)
|
|
context = _context(manager, routing=_RoutingCheck(outcome=owner, assignments=assignments))
|
|
beta = _BetaProbe(context)
|
|
result = TestRunResult(
|
|
"pytest", failure_outcome, 1, "captured postflight failure",
|
|
["tests/test_widget.py:2", "src/widget.py:2"],
|
|
run_info="Captured invocation and deadline.",
|
|
)
|
|
remediation = FailedQAItem(
|
|
focused_spotlight=source, summary="Repair the required contract.",
|
|
failed_test=FailingTest.from_result(_test(), result),
|
|
)
|
|
report = Mock(return_value=[remediation])
|
|
stage = Mock()
|
|
monkeypatch.setattr(context.backlog, "report_remediation", report)
|
|
monkeypatch.setattr(context, "stage_remediation_outbound", stage)
|
|
monkeypatch.setattr(context, "persist", Mock())
|
|
|
|
# A source verdict must reach production attribution instead of an assignment-count override.
|
|
if owner == "source":
|
|
with pytest.raises(BetaRemediationRequested):
|
|
beta._raise_for_targeted_test_failure(item, "Verify the required contract.", result, _applied(item))
|
|
report.assert_called_once()
|
|
assert report.call_args.kwargs["failed_test"].outcome is failure_outcome
|
|
assert report.call_args.kwargs["failed_test"].test.location == "tests/test_widget.py"
|
|
assert report.call_args.kwargs["source_guidance"] == {
|
|
source.location: "Candidate state was restored before validation.",
|
|
}
|
|
stage.assert_called_once_with(remediation)
|
|
# A test verdict must retain Beta even when a separate production attribution is present.
|
|
else:
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
beta._raise_for_targeted_test_failure(item, "Verify the required contract.", result, _applied(item))
|
|
assert raised.value.testing_item.message_id == item.message_id
|
|
assert raised.value.testing_item.failure_outcome is failure_outcome
|
|
assert raised.value.testing_item.failure_raw_output == result.output
|
|
assert raised.value.stop is False
|
|
report.assert_not_called()
|
|
stage.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("failure_outcome", (TestRunOutcome.Failure, TestRunOutcome.Error, TestRunOutcome.Timeout))
|
|
def test_beta_requeues_confirmed_file_blame_contradiction_without_new_identity(
|
|
monkeypatch: pytest.MonkeyPatch, failure_outcome: TestRunOutcome,
|
|
) -> None:
|
|
"""Confirmed Beta ownership retains active work for all targeted-failure branches."""
|
|
|
|
source = _source()
|
|
manager = _MemoryFileManager(sources={source.location: "def widget():\n return 2\n"},
|
|
tests={"tests/test_widget.py": "def test_widget():\n assert False\n"})
|
|
item = _testing_item(source=source)
|
|
item.message_id = "same-beta-work"
|
|
result = TestRunResult("pytest", failure_outcome, 1, "captured failure", ["tests/test_widget.py:2", "src/widget.py:2"])
|
|
retry = TestingBacklog(focused_spotlight=_test(), summary=item.summary, message_id=item.message_id,
|
|
previous_beta_guidance="Highest review confirms an obsolete test expectation.")
|
|
retry.record_failure(FailingTest.from_result(_test(), result), [source])
|
|
context = _context(manager, routing=_RoutingCheck(outcome="source"))
|
|
beta = _BetaProbe(context)
|
|
report, stage = Mock(return_value=[retry]), Mock()
|
|
monkeypatch.setattr(context.backlog, "report_remediation", report)
|
|
monkeypatch.setattr(context, "stage_remediation_outbound", stage)
|
|
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
beta._raise_for_targeted_test_failure(item, "Verify widget.", result, _applied(item))
|
|
|
|
assert raised.value.testing_item is retry
|
|
assert retry.message_id == "same-beta-work"
|
|
assert retry.failure_outcome is failure_outcome
|
|
assert retry.failure_raw_output == result.output
|
|
assert raised.value.stop is False
|
|
report.assert_called_once()
|
|
stage.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("output", [
|
|
"ordinary targeted assertion failure",
|
|
"ERROR collecting tests/test_widget.py",
|
|
])
|
|
def test_beta_targeted_routing_uses_result_despite_restart_history(
|
|
output: str,
|
|
) -> None:
|
|
"""A completed Beta-owned route cannot be reinterpreted as a dead end."""
|
|
|
|
source = _source()
|
|
client = _QueuedClient(())
|
|
manager = _MemoryFileManager(client=client)
|
|
item = _testing_item(source=source)
|
|
beta = _RequeueBetaProbe(
|
|
_context(manager),
|
|
[source],
|
|
_BoundaryRoutingCheck("RootRoutingDecision"),
|
|
)
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
result = TestRunResult("pytest", TestRunOutcome.Failure, 1, output, ["F0 trace"])
|
|
prior_beta = graph.start(Phase.BETA, _testing_item(source=source))
|
|
alpha = graph.track_provenance(
|
|
prior_beta,
|
|
Phase.ALPHA,
|
|
InspectionBacklog(source, "Repair the source before retesting."),
|
|
)
|
|
graph.track_provenance(alpha, Phase.BETA, item)
|
|
|
|
with provenance.bind(graph):
|
|
with provenance.item_scope(Phase.BETA, item):
|
|
with pytest.raises(BetaRequeueRequested):
|
|
beta._raise_for_targeted_test_failure(
|
|
item,
|
|
"P0 concrete instruction",
|
|
result,
|
|
_applied(item),
|
|
)
|
|
|
|
assert client.prompts == []
|
|
|
|
|
|
class _CapturingService(BetaWorkflowJudgementService):
|
|
"""Capture refinement inputs without starting a judgement client."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.calls: list[dict[str, object]] = []
|
|
|
|
def refine_beta_instruction(
|
|
self,
|
|
*,
|
|
file_manager: FileManager,
|
|
focused_test: ClassMeta,
|
|
original_cot: str,
|
|
accepted_source_diff: str,
|
|
assignment_type: str,
|
|
test_status: str,
|
|
previous_beta_guidance: str = "",
|
|
failure_error_message: str = "",
|
|
failure_stack_trace: Sequence[str] = (),
|
|
failure_run_info: str = "",
|
|
readonly_sources: Sequence[ClassMeta] = (),
|
|
previous_cot_records: tuple[CotTaskRecord, ...] = (),
|
|
editor_records: Sequence[EditorRecord] = (),
|
|
logger: logging.Logger | None = None,
|
|
) -> str:
|
|
del logger
|
|
self.calls.append(
|
|
{
|
|
"file_manager": file_manager,
|
|
"focused_test": focused_test,
|
|
"original_cot": original_cot,
|
|
"accepted_source_diff": accepted_source_diff,
|
|
"assignment_type": assignment_type,
|
|
"test_status": test_status,
|
|
"previous_beta_guidance": previous_beta_guidance,
|
|
"failure_error_message": failure_error_message,
|
|
"failure_stack_trace": list(failure_stack_trace),
|
|
"failure_run_info": failure_run_info,
|
|
"readonly_sources": list(readonly_sources),
|
|
"previous_cot_records": previous_cot_records,
|
|
"editor_records": list(editor_records),
|
|
}
|
|
)
|
|
return "P1 concrete instruction"
|
|
|
|
|
|
def test_loop_refinement_forwards_backlog_predecessor() -> None:
|
|
"""Refinement retains prior guidance and the current invocation evidence."""
|
|
service = _CapturingService()
|
|
source = _source()
|
|
manager = _MemoryFileManager(sources={source.location: "value = 1\n"})
|
|
item = _testing_item(source=source)
|
|
item.original_cot = "goal"
|
|
item.previous_beta_guidance = "P0 concrete instruction"
|
|
item.editor_records = [EditorRecord("retained setup", "old setup", "new setup")]
|
|
item.failure_error_message = "F0 failure"
|
|
item.failure_run_info = "PROCESS DEADLINE SECONDS: 60"
|
|
beta = _BetaProbe(_context(manager, service=service))
|
|
assert beta._refine_beta_instruction(item) == "P1 concrete instruction"
|
|
assert service.calls[0]["previous_beta_guidance"] == "P0 concrete instruction"
|
|
assert service.calls[0]["failure_run_info"] == item.failure_run_info
|
|
assert service.calls[0]["editor_records"] == item.editor_records
|
|
|
|
|
|
@pytest.mark.parametrize("with_source", (False, True))
|
|
def test_full_suite_test_tooling_failure_reaches_beta_repair(
|
|
monkeypatch: pytest.MonkeyPatch, with_source: bool,
|
|
) -> None:
|
|
"""A failed testing randomizer reaches repair regardless of source relationships."""
|
|
|
|
evidence: str = "tests/test_randomizer.py:5: in test_randomizer\nassert randomizer() == expected\nAssertionError"
|
|
client = _QueuedClient()
|
|
manager = _MemoryFileManager(
|
|
client=client,
|
|
sources={"src/value.py": "value = 1\n"} if with_source else {},
|
|
tests={"tests/test_randomizer.py": "def randomizer():\n return 1\n\ndef test_randomizer():\n assert randomizer() == 2\n"},
|
|
diffs={"src/value.py": "accepted source change"},
|
|
)
|
|
[test] = manager.test_classes
|
|
dependency = Mock(spec=BiDependencyDiscoveryStrategy)
|
|
dependency.upstream_sources_for.return_value = manager.src_classes
|
|
strategies = LanguageStrategySet.get(manager)
|
|
strategies.bindings[BiDependencyDiscoveryStrategy] = dependency
|
|
monkeypatch.setattr(LanguageStrategySet, "instance", lambda *_args, **_kwargs: strategies)
|
|
service = _CapturingService()
|
|
monkeypatch.setattr(service, "route_targeted_failure", Mock(return_value=_RoutingCheck(outcome="test")))
|
|
backlog = BackLog(completion={}, judgement_service=service)
|
|
context = BetaContext(
|
|
goal="Repair testing randomizer", classes=[], file_manager=manager,
|
|
backlog=QABacklog(backlog, client, manager, Mock(spec=CodeParser)),
|
|
)
|
|
context.move_inbound(StagedInboundItem(
|
|
lane=BacklogLane.Testing, value=TestingBacklog(None, "Full suite", last_exam=True),
|
|
inbound_id="full-suite",
|
|
))
|
|
monkeypatch.setattr(BetaContext, "persist", Mock())
|
|
editor = _RecordingEditor("Repaired selected test")
|
|
runner = _TestRunner((TestRunResult("test randomizer", TestRunOutcome.Success, 0, "passed"),))
|
|
beta = _BetaProbe(context, editor=editor, runner=runner)
|
|
|
|
output = f"suite preamble λ \n\n{evidence}\n\nretained trailer\t\n"
|
|
trace = ["", *evidence.splitlines(), "retained trace "]
|
|
runner.results.insert(0, TestRunResult("single randomizer", TestRunOutcome.Failure, 1, output, trace))
|
|
monkeypatch.setattr(runner, "matching_tests", Mock(return_value={test.location: ("test_randomizer",)}))
|
|
monkeypatch.setattr(runner, "failure_reports", Mock(return_value=[FailingTest(
|
|
test, trace, "Test command failed with return code 1.", output,
|
|
test_name="test_randomizer", outcome=TestRunOutcome.Failure,
|
|
)]))
|
|
beta._report_all_tests_failure(TestRunResult("all tests", TestRunOutcome.Failure, 1, output, trace))
|
|
context.move_outbound_to_backlog()
|
|
[item] = backlog.testing
|
|
assert item.focused_spotlight == test
|
|
assert item.downstream_classes == manager.src_classes
|
|
assert test in item.failure_context_sources
|
|
assert item.failure_error_message == "Test command failed with return code 1."
|
|
assert item.failure_stack_trace == trace
|
|
assert item.failure_raw_output == output
|
|
|
|
result = beta._handle_testing_item(item)
|
|
|
|
assert result.passed and result.new_failure is None
|
|
assert "judgement PASS" in result.summary
|
|
assert service.calls[0]["accepted_source_diff"] == ("accepted source change" if with_source else "")
|
|
assert service.calls[0]["test_status"] == "FAILED"
|
|
assert service.calls[0]["failure_stack_trace"] == trace
|
|
assert runner.calls == [([test], 1, ("test_randomizer",)), ([test], 1, ())]
|
|
[request] = editor.requests
|
|
assert request.editable_files == [manager.test_path(test.location)]
|
|
assert not request.is_remediation
|
|
assert request.remediation_originator_test is None
|
|
assert item.failure_stack_trace == []
|
|
|
|
|
|
def test_beta_apply_receives_only_refined_instruction() -> None:
|
|
manager = _MemoryFileManager(tests={"tests/test_widget.py": "def test_widget(): pass\n"})
|
|
editor = _RecordingEditor("edited")
|
|
beta = _BetaProbe(_context(manager), editor=editor)
|
|
item = _testing_item()
|
|
item.previous_beta_guidance = "P0 concrete instruction"
|
|
beta._apply_test_work(item, "P1 concrete instruction")
|
|
assert editor.requests[0].backlog_guidance[0] == "P1 concrete instruction"
|
|
assert "P0 concrete instruction" not in editor.requests[0].backlog_guidance
|
|
|
|
|
|
def test_beta_consumes_runtime_leave_control_before_test_refresh_or_retry_state() -> None:
|
|
manager = _MemoryFileManager(tests={"tests/test_widget.py": "def test_widget(): pass\n"})
|
|
editor = _RecordingEditor(
|
|
CodeEditLeaveSummary(
|
|
"CODE EDIT OR LEAVE IT\n"
|
|
+ CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN
|
|
+ "\nJudgement summary:\nNo test mutation is needed."
|
|
)
|
|
)
|
|
beta = _BetaProbe(_context(manager), editor=editor)
|
|
applied = beta._apply_test_work(_testing_item(), "P1 concrete instruction")
|
|
assert CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN not in applied.summary
|
|
assert manager.refreshed_summaries
|
|
assert CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN not in manager.refreshed_summaries[0]
|
|
|
|
|
|
def test_failure_context_curation_skips_selection_without_failure_evidence() -> None:
|
|
manager = _MemoryFileManager()
|
|
failure = FailingTest(_test(), [], "", "")
|
|
assert curate_failure_context_sources(file_manager=manager, failure=failure) == []
|
|
|
|
|
|
def test_failure_context_curation_selects_matched_candidates_without_completion() -> None:
|
|
"""Only a path-matched source candidate enters context, without semantic admission."""
|
|
|
|
first, second = _source("src/first.py"), _source("src/second.py")
|
|
client = _QueuedClient(())
|
|
manager = _MemoryFileManager(
|
|
client=client,
|
|
sources={first.location: "first = 1\n", second.location: "second = 2\n"},
|
|
)
|
|
selected = curate_failure_context_sources(
|
|
file_manager=manager,
|
|
failure=FailingTest(_test(), [], "failed in src/second.py:12", "output"),
|
|
)
|
|
assert selected == [second]
|
|
assert client.prompts == []
|
|
|
|
|
|
def test_beta_apply_expands_and_dedupes_existing_readonly_paths() -> None:
|
|
manager = _MemoryFileManager(tests={"tests/test_widget.py": "def test_widget(): pass\n"})
|
|
editor = _RecordingEditor("edited")
|
|
beta = _BetaProbe(_context(manager), editor=editor)
|
|
beta._apply_test_work(
|
|
TestingBacklog(
|
|
focused_spotlight=_test(),
|
|
summary="ticket",
|
|
upstream_classes=[_source("src/owner.py"), _source("src/shared.py")],
|
|
failure_context_sources=[_source("src/shared.py"), _source("src/failure.py"), _test(), _test("tests/toolkit.py")],
|
|
),
|
|
"instruction",
|
|
)
|
|
assert editor.requests[0].readonly_files == [
|
|
Path("/testbed/src/owner.py"),
|
|
Path("/testbed/src/shared.py"),
|
|
Path("/testbed/src/failure.py"),
|
|
Path("/testbed/tests/toolkit.py"),
|
|
]
|
|
|
|
|
|
def test_refinement_prompt_scopes_one_assignment_to_the_whole_selected_file() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="repair accepted widget behavior",
|
|
accepted_source_diff="diff",
|
|
focused_context="tests/example/test_widget.py",
|
|
assignment_type="FAILED CANDIDATE",
|
|
test_status="FAILED",
|
|
failure_evidence="tests/example/test_widget.py::test_a failed\ntests/example/test_widget.py::test_b failed",
|
|
)
|
|
assert "one test-file assignment" in prompt
|
|
assert "multiple test functions or regions within the selected file" in prompt
|
|
assert "Do not select or edit another file" in prompt
|
|
|
|
|
|
def test_beta_guidance_requires_all_cited_selected_file_diagnostics() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="repair accepted widget behavior",
|
|
accepted_source_diff="diff",
|
|
focused_context="tests/example/test_widget.py",
|
|
assignment_type="FAILED CANDIDATE",
|
|
test_status="FAILED",
|
|
failure_evidence="_Manager is not assignable to FileManager\nduplicate test declaration",
|
|
)
|
|
assert "all supplied failure, compiler, parser, reference, and patch evidence" in prompt
|
|
assert "every cited blocking problem that belongs to the selected test file" in prompt
|
|
assert "not only the original\nbacklog failure" in prompt
|
|
|
|
|
|
class _CapturingGuidance(BetaContractGuidanceDecision):
|
|
"""Capture service construction through its nominal extension point."""
|
|
|
|
captured_request: AiderEditRequest | None = None
|
|
captured_focused_test: ClassMeta | None = None
|
|
captured_previous_guidance = ""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
client: OpenAICompatibleClient,
|
|
file_manager: FileManager,
|
|
request: AiderEditRequest,
|
|
focused_test: ClassMeta,
|
|
original_cot: str,
|
|
accepted_source_diff: str,
|
|
assignment_type: str,
|
|
test_status: str,
|
|
previous_beta_guidance: str = "",
|
|
readonly_sources: list[ClassMeta] | None = None,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
del client, file_manager, original_cot, accepted_source_diff, assignment_type, test_status, readonly_sources, logger
|
|
type(self).captured_request = request
|
|
type(self).captured_focused_test = focused_test
|
|
type(self).captured_previous_guidance = previous_beta_guidance
|
|
self.beta_instruction = "Repair test_a, test_b, and test_c in tests/example/test_widget.py."
|
|
self.prompt = "BETA CONTRACT GUIDANCE\nCaptured focused test evidence."
|
|
self.outcome = "guidance"
|
|
self.reason = "The focused test needs one consolidated assignment."
|
|
self.additional = {"beta_instruction": self.beta_instruction}
|
|
|
|
def __call__(self, *args: object, **kwargs: object) -> bool:
|
|
del args, kwargs
|
|
return True
|
|
|
|
|
|
def test_refine_beta_instruction_preserves_file_scoped_multi_failure_obligation() -> None:
|
|
focused = _test("tests/example/test_widget.py")
|
|
client = _QueuedClient(())
|
|
manager = _MemoryFileManager(client=client)
|
|
service = BetaWorkflowJudgementService(guidance_cls=_CapturingGuidance)
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
item = TestingBacklog(focused_spotlight=focused, summary="repair widget behavior")
|
|
prior_beta = graph.start(Phase.BETA, _testing_item())
|
|
alpha = graph.track_provenance(
|
|
prior_beta,
|
|
Phase.ALPHA,
|
|
InspectionBacklog(_source(), "Repair the source before refining the test."),
|
|
)
|
|
graph.track_provenance(alpha, Phase.BETA, item)
|
|
with provenance.bind(graph):
|
|
with provenance.item_scope(Phase.BETA, item):
|
|
instruction = service.refine_beta_instruction(
|
|
file_manager=manager,
|
|
focused_test=focused,
|
|
original_cot="repair widget behavior",
|
|
accepted_source_diff="accepted source diff",
|
|
assignment_type="FAILED CANDIDATE",
|
|
test_status="FAILED",
|
|
previous_beta_guidance="P0 concrete instruction",
|
|
failure_error_message="test_a, test_b, and test_c failed",
|
|
failure_stack_trace=[
|
|
"tests/example/test_widget.py::test_a",
|
|
"tests/example/test_widget.py::test_b",
|
|
"tests/example/test_widget.py::test_c",
|
|
],
|
|
)
|
|
request = _CapturingGuidance.captured_request
|
|
assert request is not None
|
|
assert "test_a, test_b, and test_c" in instruction
|
|
assert _CapturingGuidance.captured_focused_test is focused
|
|
assert _CapturingGuidance.captured_previous_guidance == "P0 concrete instruction"
|
|
assert request.authoritative_prompt == ""
|
|
assert request.backlog_guidance == []
|
|
assert "tests/example/test_widget.py::test_a" in request.failure_stack_trace
|
|
assert client.prompts == []
|
|
|
|
|
|
class _CapturingTargetedRouting(TargetedAlphaOrBetaRoutingDecision):
|
|
"""Capture a targeted routing carrier without invoking a client."""
|
|
|
|
request: AiderEditRequest | None = None
|
|
story: TargetedFailureStory | None = None
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
client: CompletionClient,
|
|
file_manager: FileManager,
|
|
request: AiderEditRequest,
|
|
story: TargetedFailureStory,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
del client, file_manager, logger
|
|
type(self).request = request
|
|
type(self).story = story
|
|
ReasonableVibeCheck.__init__(self, outcome="test", reason="Need more evidence.")
|
|
|
|
def __call__(self, *args: object, **kwargs: object) -> bool:
|
|
del args, kwargs
|
|
return False
|
|
|
|
|
|
class _CapturingCollectionRouting(TargetedCollectionAlphaOrBetaRoutingDecision):
|
|
"""Capture a collection routing carrier without invoking a client."""
|
|
|
|
request: AiderEditRequest | None = None
|
|
story: TargetedFailureStory | None = None
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
client: CompletionClient,
|
|
file_manager: FileManager,
|
|
request: AiderEditRequest,
|
|
story: TargetedFailureStory,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
del client, file_manager, logger
|
|
type(self).request = request
|
|
type(self).story = story
|
|
ReasonableVibeCheck.__init__(self, outcome="test", reason="Need more evidence.")
|
|
|
|
def __call__(self, *args: object, **kwargs: object) -> bool:
|
|
del args, kwargs
|
|
return False
|
|
|
|
|
|
def _story(*, phase: str, work_needed: str) -> tuple[_MemoryFileManager, TargetedFailureStory]:
|
|
"""Build one fully typed routing story."""
|
|
|
|
source, focused = _source(), _test()
|
|
manager = _MemoryFileManager(
|
|
sources={source.location: "value = 1\n"},
|
|
tests={focused.location: ""},
|
|
)
|
|
item = TestingBacklog(
|
|
focused_spotlight=focused,
|
|
summary="ticket provenance",
|
|
downstream_classes=[source],
|
|
original_cot="immutable collection CoT",
|
|
)
|
|
failing = FailingTest(focused, [focused.location, source.location, f"{phase} stack"], f"{phase} failure", f"{phase} raw output")
|
|
return manager, TargetedFailureStory(
|
|
testing_item=item,
|
|
raw_output=failing.raw_output,
|
|
work_needed=work_needed,
|
|
candidate_sources=[source],
|
|
failing=failing,
|
|
failure_phase=phase,
|
|
accepted_alpha_summary="accepted Alpha summary",
|
|
structural_evidence={},
|
|
semantic_evidence={},
|
|
previous_beta_guidance="previous Beta guidance",
|
|
beta_edit_diff="Beta diff",
|
|
beta_editor_summary="Beta editor summary",
|
|
)
|
|
|
|
|
|
def test_targeted_routing_keeps_cot_and_work_out_of_the_carrier_request() -> None:
|
|
manager, story = _story(phase="after_beta", work_needed="directly applicable prior Beta work")
|
|
story.editor_records.append(EditorRecord("retained setup", "old setup", "new setup"))
|
|
BetaWorkflowJudgementService(targeted_routing_cls=_CapturingTargetedRouting).route_targeted_failure(
|
|
file_manager=manager,
|
|
story=story,
|
|
)
|
|
request = _CapturingTargetedRouting.request
|
|
captured_story = _CapturingTargetedRouting.story
|
|
assert request is not None
|
|
assert captured_story is story
|
|
assert captured_story is not None
|
|
assert request.authoritative_prompt == ""
|
|
assert request.backlog_guidance == []
|
|
assert request.editable_files == [Path("/testbed/tests/test_widget.py")]
|
|
assert request.readonly_files == [Path("/testbed/src/widget.py")]
|
|
assert request.failure_error_message == ""
|
|
assert captured_story.original_cot == "immutable collection CoT"
|
|
assert captured_story.work_needed == "directly applicable prior Beta work"
|
|
assert request.validation_receipts.receipts == ()
|
|
assert request.editor_records == story.editor_records
|
|
|
|
|
|
def test_collection_routing_keeps_cot_and_work_out_of_the_carrier_request() -> None:
|
|
manager, story = _story(phase="collection", work_needed="directly applicable collection work")
|
|
story.editor_records.append(EditorRecord("retained setup", "old setup", "new setup"))
|
|
BetaWorkflowJudgementService(collection_routing_cls=_CapturingCollectionRouting).route_collection_failure(
|
|
file_manager=manager,
|
|
story=story,
|
|
)
|
|
request = _CapturingCollectionRouting.request
|
|
captured_story = _CapturingCollectionRouting.story
|
|
assert request is not None
|
|
assert captured_story is story
|
|
assert captured_story is not None
|
|
assert request.authoritative_prompt == ""
|
|
assert request.backlog_guidance == []
|
|
assert request.editable_files == [Path("/testbed/tests/test_widget.py")]
|
|
assert request.readonly_files == [Path("/testbed/src/widget.py")]
|
|
assert request.failure_error_message == ""
|
|
assert captured_story.work_needed == "directly applicable collection work"
|
|
assert captured_story.failing.error_message == "collection failure"
|
|
assert request.validation_receipts.receipts == ()
|
|
assert request.editor_records == story.editor_records
|
|
|
|
|
|
def test_beta_authoritative_prompt_allows_multiple_regions_but_not_other_files() -> None:
|
|
beta = _BetaProbe(_context(_MemoryFileManager()))
|
|
prompt = beta._authoritative_prompt([Path("/testbed/tests/example/test_widget.py")])
|
|
assert "multiple functions or regions within that file" in prompt
|
|
assert "another test file" in prompt
|
|
|
|
|
|
def test_refinement_prompt_preserves_contract_when_classifying_obsolete_tests() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="replacement contract",
|
|
accepted_source_diff="diff",
|
|
focused_context="test context",
|
|
assignment_type="FAILED CANDIDATE",
|
|
test_status="FAILED",
|
|
failure_evidence="stale caller",
|
|
)
|
|
assert "AUTHORITATIVE TASK > COMPATIBLE GUIDANCE > APPLICABLE PRE-EXISTING PRECEDENTS" in prompt
|
|
assert "OBSOLETE TEST: a previous test expectation conflicts with the contract required" in prompt
|
|
assert "PRODUCTION REGRESSION: production violates a requirement" in prompt
|
|
assert "DELETE only the obsolete test function unless the entire file contract is obsolete under the shared definition" in prompt
|
|
assert "stale production caller" in prompt
|
|
|
|
|
|
def test_beta_refinement_prompt_derives_asserted_contract_before_test_mutation() -> None:
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot="managed entry captures and construction is inert",
|
|
accepted_source_diff="accepted source uses managed entry",
|
|
focused_context="def test_capture(): assert public_capture()",
|
|
assignment_type="FAILED CANDIDATE",
|
|
test_status="FAILED",
|
|
previous_beta_guidance="rename capture",
|
|
failure_evidence="test_capture failed",
|
|
)
|
|
assert "WHAT BEHAVIOR DOES THE FOCUSED TEST OR MISSING COVERAGE SEMANTICALLY REPRESENT?" in prompt
|
|
assert "asserted_contract" in prompt
|
|
assert "desired_contract" in prompt
|
|
assert "contract_similarity" in prompt
|
|
assert "A factual diagnosis in guidance still requires verification." in prompt
|
|
assert "Do not preserve a factual premise merely because earlier guidance asserted it." in prompt
|
|
|
|
|
|
@pytest.mark.parametrize("language,focused_test", [
|
|
("C++", "TEST(Client, isolated) { FakeTransport transport; EXPECT_EQ(Client(transport).read(), 2); }"),
|
|
("JavaScript", "it('isolates the client', () => expect(new Client(fakeTransport).read()).toBe(2));"),
|
|
])
|
|
def test_refinement_prompt_preserves_target_language_and_unit_isolation(language: str, focused_test: str) -> None:
|
|
"""General contract instructions retain the supplied target language and isolated test evidence."""
|
|
|
|
goal = f"The {language} client must return 2 and its unit tests must use a controlled transport."
|
|
prompt = BetaContractGuidanceDecision._prompt(
|
|
original_cot=goal, accepted_source_diff="Client now accepts a transport dependency.",
|
|
focused_context=focused_test, assignment_type="FAILED CANDIDATE", test_status="FAILED",
|
|
failure_evidence="The fixture contacted a remote service and timed out.",
|
|
)
|
|
|
|
assert goal in prompt
|
|
assert focused_test in prompt
|
|
assert "do not assume the target is Band-Aider or Python" in prompt
|
|
assert "A unit test must run in isolation" in prompt
|
|
assert "controlled substitutes" in prompt
|
|
assert "ClassMeta" not in prompt
|
|
assert "Memento.capture" not in prompt
|
|
|
|
|
|
def test_targeted_failure_preserves_exact_captured_evidence() -> None:
|
|
"""Beta recording projects the same canonical fields without trimming or replacing text."""
|
|
|
|
manager = _MemoryFileManager(
|
|
client=_QueuedClient(()), sources={},
|
|
tests={"tests/test_randomizer.py": "def test_randomizer(): pass\n"}, diffs={},
|
|
)
|
|
[test] = manager.test_classes
|
|
backlog = BackLog(completion={})
|
|
context = BetaContext(
|
|
goal="repair test", classes=[], file_manager=manager,
|
|
backlog=QABacklog(backlog, manager.client, manager, Mock(spec=CodeParser)),
|
|
)
|
|
beta = _BetaProbe(context, editor=_RecordingEditor("unused"), runner=_TestRunner(()))
|
|
item = TestingBacklog(test, "repair")
|
|
result = TestRunResult("targeted", TestRunOutcome.Failure, 1, " λ\r\n\n", ["tests/test_randomizer.py:1 ", "", "λ\r"])
|
|
failure = beta._record_targeted_failure(item, result)
|
|
assert failure.to_record() == FailingTest.from_result(test, result).to_record()
|
|
assert item.failure_error_message == "Test command failed with return code 1."
|
|
assert item.failure_stack_trace == result.stack_trace
|
|
assert item.failure_raw_output == result.output
|