1235 lines
45 KiB
Python
1235 lines
45 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 inspect
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from io import StringIO
|
|
import pytest
|
|
from typing import Callable, Iterable, Sequence, TextIO, cast
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.context.alpha_context import AlphaContext
|
|
from band_aider.context.beta_context import BetaContext
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.context.resume_state import configure_resume_settings
|
|
from band_aider.context.work_completion import WorkCompletion
|
|
from band_aider.context.working_backlog import (
|
|
BackLog as _BackLog,
|
|
ChangeType,
|
|
ContributorBacklog,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
QABacklog,
|
|
TestingBacklog,
|
|
)
|
|
from band_aider.judgement.coverage_strategy import CoverageStrategy
|
|
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.code_dependency_graph import DependencySelector
|
|
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import BiDependencyDiscoveryStrategy
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.python_ast_dependency_strategy import PythonAstDependencyStrategy
|
|
from band_aider.judgement.python_coverage_strategy import (
|
|
MINIMUM_BRANCH_COVERAGE,
|
|
PythonCoverageStrategy,
|
|
)
|
|
from band_aider.judgement.retry_vibe_check import RetryVibeCheck
|
|
from band_aider.judgement.source_discovery_strategy import SourceDiscoveryStrategy
|
|
from band_aider.judgement.test_creation_strategy import NamingStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy
|
|
from band_aider.judgement.repo_exec import current_repo_exec
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from band_aider.loop.beta_loop import BetaLoop
|
|
from band_aider.loop.mechanical_loop_gate import MechanicalLoopGate
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
from band_aider.strategies import LanguageStrategySet, language_strategy_set_factory
|
|
from band_aider.context.aider_config import AiderConfig
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class BackLog(_BackLog):
|
|
"""Real backlog with a test-friendly default completion ledger."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
completion: dict[str, WorkCompletion] | None = None,
|
|
remediation: Iterable[FailedQAItem] | None = None,
|
|
inspection: Iterable[InspectionBacklog] | None = None,
|
|
testing: Iterable[TestingBacklog] | None = None,
|
|
) -> None:
|
|
"""Initialize the production backlog without requiring an irrelevant ledger fixture."""
|
|
|
|
super().__init__(
|
|
completion={} if completion is None else completion,
|
|
remediation=remediation,
|
|
inspection=inspection,
|
|
testing=testing,
|
|
)
|
|
|
|
|
|
# Shared loop test helpers extracted from test_loop_strategies.py.
|
|
|
|
class FakeClient(OpenAICompatibleClient):
|
|
"""FakeClient returns strict JSON for loop metadata prompts."""
|
|
|
|
def __init__(self, *, acceptance_reasons: tuple[str, ...] = ("def foo():",)) -> None:
|
|
"""Initialize an OpenAI-compatible shape without a reachable endpoint."""
|
|
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
self.acceptance_reasons = acceptance_reasons
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return a minimal JSON response matching the requested prompt shape."""
|
|
|
|
del logger
|
|
|
|
# Metadata fields embedded in evidence cannot change the active response contract.
|
|
if prompt.startswith("Only the CDC task is active."):
|
|
return json.dumps({"cdc": "Metadata refreshed."})
|
|
# These loop scenarios require no supplemental dependency reference cards.
|
|
if "BEGIN DEPENDENCY REFERENCE SELECTION" in prompt.splitlines()[0]:
|
|
return json.dumps({"references": []})
|
|
|
|
# Scenario decisions already carry fixed work and attribution evidence.
|
|
if "BEGIN FUTURE GUIDANCE APPLICABILITY" in prompt.splitlines()[0]:
|
|
return json.dumps({"applicable": False, "reason": "The fixed local evidence already supplies this observation."})
|
|
|
|
if (
|
|
"ACCEPTANCE ROLE" in prompt
|
|
and "ALPHA EDIT ACCEPTANCE" in prompt
|
|
and "DECISION RULE" in prompt
|
|
):
|
|
return json.dumps(
|
|
{
|
|
"outcome": "pass",
|
|
"reason": list(self.acceptance_reasons),
|
|
"instruction": "",
|
|
}
|
|
)
|
|
if "FILE BLAME CANDIDATE PRECEDENCE" in prompt:
|
|
return json.dumps(
|
|
{
|
|
"location": "src/foo.py",
|
|
"reason": list(self.acceptance_reasons),
|
|
"explanation": (
|
|
"The selected source is closer to the originating COT violation."
|
|
),
|
|
}
|
|
)
|
|
if "FAILURE CONTEXT SELECTION" in prompt:
|
|
return json.dumps({"include": False})
|
|
if "Judge Alpha-or-Beta routing for this targeted test collection/discovery/setup/import failure." in prompt:
|
|
return json.dumps(
|
|
{
|
|
"outcome": "test",
|
|
"reason": "The targeted collection output indicates missing or unrunnable test discovery work.",
|
|
"additional": {
|
|
"failure_phase": "collection",
|
|
"failed_test": "tests/test_foo.py",
|
|
"error_message": "collected 0 items",
|
|
"relevant_stack_trace": ["collected 0 items"],
|
|
"test_work_needed": "Create or repair runnable pytest tests in the target file.",
|
|
"source_clues": [],
|
|
"source_remediation_locations": [],
|
|
"import_chain": [],
|
|
},
|
|
}
|
|
)
|
|
if "Judge Alpha-or-Beta routing for this targeted test failure." in prompt:
|
|
return json.dumps(
|
|
{
|
|
"outcome": "source",
|
|
"reason": "Readonly source behavior appears incorrect relative to the test intent.",
|
|
"additional": {"test_work_needed": ""},
|
|
}
|
|
)
|
|
if "BETA TEST PROMPT REFINEMENT" in prompt:
|
|
return json.dumps(
|
|
{
|
|
"asserted_contract": "The focused test protects the accepted observable source behavior.",
|
|
"desired_contract": "The current COT and accepted source retain that observable behavior.",
|
|
"divergence": "The test needs only the current contract assertion.",
|
|
"contract_similarity": 0.80,
|
|
"action": "adapt" if "FOCUSED TEST STATUS\nFAILED" in prompt else "add",
|
|
"beta_instruction": (
|
|
"Update the focused test with the smallest regression assertion "
|
|
"for the accepted source behavior."
|
|
)
|
|
}
|
|
)
|
|
if "BETA CONTRACT GUIDANCE ROLE" in prompt:
|
|
return json.dumps(
|
|
{
|
|
"needs_beta_work": True,
|
|
"reason": "Update the directly connected test for the accepted source contract.",
|
|
"new_contract": "The source now exposes the accepted behavior.",
|
|
"invalidated_contract": "The prior expectation is obsolete when contradicted by the accepted source diff.",
|
|
"regression_obligation": "The direct regression must continue to prove the public behavior.",
|
|
"existing_tests_to_update": [
|
|
{
|
|
"test_beat": "tests/test_foo.py::test_foo",
|
|
"required_change": "Assert the accepted observable source behavior.",
|
|
}
|
|
],
|
|
"existing_tests_to_preserve": [
|
|
{
|
|
"test_beat": "tests/test_foo.py::test_foo",
|
|
"reason": "The regression remains the direct source contract guard.",
|
|
}
|
|
],
|
|
"new_tests": [],
|
|
"failure_mode": "",
|
|
"beta_instruction": "Update only the directly connected test beat for the accepted source contract.",
|
|
}
|
|
)
|
|
if "Return exactly one JSON array of repo-relative source paths." in prompt:
|
|
return "[]"
|
|
if "Return exactly one JSON array of repo-relative .py test paths." in prompt:
|
|
return "[]"
|
|
if '"satisfied"' in prompt:
|
|
return json.dumps(
|
|
{"satisfied": True, "reason": "closed", "remaining_work": ""}
|
|
)
|
|
if "incomplete" in prompt.lower():
|
|
return json.dumps({"summary": "Incomplete missing coverage."})
|
|
if '"work_required"' in prompt:
|
|
return json.dumps(
|
|
{
|
|
"work_required": True,
|
|
"evidence_type": "cot",
|
|
"semantic_divergence": "none",
|
|
"reason": list(self.acceptance_reasons),
|
|
"work_needed": "1. Identify the COT-owned source operation.\n2. Align it with the COT.\n3. Verify the focused contract.",
|
|
}
|
|
)
|
|
if '"work_needed"' in prompt:
|
|
return json.dumps({"work_needed": "Refresh metadata.", "reason": list(self.acceptance_reasons)})
|
|
if '"change_type"' in prompt:
|
|
return json.dumps(
|
|
{"summary": "Validate source behavior.", "change_type": "Update"}
|
|
)
|
|
if '"summary"' in prompt and '"files"' in prompt:
|
|
return json.dumps({"summary": "Squashed metadata.", "files": {}})
|
|
if '"summary"' in prompt:
|
|
return json.dumps({"summary": "Completed metadata."})
|
|
|
|
return json.dumps({"cdc": "Metadata refreshed."})
|
|
|
|
class FakeParser(CodeParser):
|
|
"""FakeParser returns deterministic ClassMeta records without model calls."""
|
|
|
|
def __init__(self, src_root: Path, test_root: Path) -> None:
|
|
"""Store source and test roots for relative location calculation."""
|
|
|
|
self.src_root = src_root
|
|
self.test_root = test_root
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return metadata for one source or test path."""
|
|
|
|
path = Path(file_name)
|
|
root = self.src_root if self.src_root in path.parents else self.test_root
|
|
relative = str(path.relative_to(root))
|
|
symbols = {"foo": "public symbol"} if relative.endswith("foo.py") else {}
|
|
return ClassMeta(
|
|
location=relative,
|
|
summary=f"summary for {relative}",
|
|
relevance=[prompt] if prompt else [],
|
|
relevant_symbols=symbols,
|
|
content_type=content_type,
|
|
)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Parse a stream through the same lexical metadata rule as a file."""
|
|
|
|
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]:
|
|
"""Parse each supplied source path in deterministic order."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
class PreflightNewSourceClient(FakeClient):
|
|
"""PreflightNewSourceClient asks Alpha to provision one new source file."""
|
|
|
|
def __init__(self, location: str = "src/pkg/new_file.py") -> None:
|
|
"""Store the requested source preflight location."""
|
|
|
|
super().__init__()
|
|
self.location = location
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return a preflight source path for the Alpha preflight prompt."""
|
|
|
|
if "Return exactly one JSON array of repo-relative source paths." in prompt:
|
|
return json.dumps([self.location])
|
|
return super().complete(prompt, logger)
|
|
|
|
class AlphaPreflightClient(FakeClient):
|
|
"""AlphaPreflightClient records one-source prompts and returns per-source responses."""
|
|
|
|
def __init__(self, responses_by_source: dict[str, list[str]]) -> None:
|
|
"""Store per-source responses and every captured prompt."""
|
|
|
|
super().__init__()
|
|
self.responses_by_source = responses_by_source
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the configured response for the one source file in this prompt."""
|
|
|
|
# Select the source-specific reply from the preflight's rendered evidence.
|
|
if "Return exactly one JSON array of repo-relative source paths." in prompt:
|
|
self.prompts.append(prompt)
|
|
location = prompt.split("====== BEGIN LOCATION ======\n", 1)[1].split(
|
|
"\n====== END LOCATION ======", 1
|
|
)[0]
|
|
return json.dumps(self.responses_by_source.get(location, []))
|
|
return super().complete(prompt, logger)
|
|
|
|
class WorkNeededPromptClient(FakeClient):
|
|
"""WorkNeededPromptClient records ContributorBacklog work-needed prompts."""
|
|
|
|
def __init__(self, response: str = "summarized work") -> None:
|
|
"""Store the configured work-needed response and captured prompts."""
|
|
|
|
super().__init__()
|
|
self.response = response
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the configured work-needed response for contributor prompts."""
|
|
|
|
if "Determine whether this focused file requires source work" in prompt:
|
|
self.prompts.append(prompt)
|
|
return json.dumps(
|
|
{
|
|
"work_required": True,
|
|
"evidence_type": "cot",
|
|
"semantic_divergence": "none",
|
|
"reason": list(self.acceptance_reasons),
|
|
"work_needed": self.response,
|
|
}
|
|
)
|
|
return super().complete(prompt, logger)
|
|
|
|
class BetaPreflightClient(FakeClient):
|
|
"""BetaPreflightClient returns a configured test preflight response."""
|
|
|
|
def __init__(self, response: str) -> None:
|
|
"""Store the Beta preflight response and prompts."""
|
|
|
|
super().__init__()
|
|
self.response = response
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the configured response for Beta test preflight prompts."""
|
|
|
|
if "Return exactly one JSON array of repo-relative .py test paths." in prompt:
|
|
self.prompts.append(prompt)
|
|
return self.response
|
|
return super().complete(prompt, logger)
|
|
|
|
class PostflightBlameClient(FakeClient):
|
|
"""PostflightBlameClient returns configured source blame locations."""
|
|
|
|
def __init__(self, response: list[str]) -> None:
|
|
"""Store the post-flight blame response."""
|
|
|
|
super().__init__()
|
|
self.response = response
|
|
self.prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return configured blame locations for post-flight remediation prompts."""
|
|
|
|
if "Given this test failure, choose which available source files are faulted" in prompt:
|
|
self.prompts.append(prompt)
|
|
return json.dumps(
|
|
{
|
|
"outcome": "blame" if self.response else "none",
|
|
"reason": "configured file blame response",
|
|
"additional": {"source_locations": self.response},
|
|
}
|
|
)
|
|
return super().complete(prompt, logger)
|
|
|
|
class FailureEvidenceClient(FakeClient):
|
|
"""FailureEvidenceClient returns configured failure evidence and blame responses."""
|
|
|
|
def __init__(
|
|
self,
|
|
evidence_response: dict[str, object] | str,
|
|
ownership_response: dict[str, object] | str | None = None,
|
|
blame_response: list[str] | None = None,
|
|
collection_ownership_response: dict[str, object] | str | None = None,
|
|
) -> None:
|
|
"""Store configured extraction responses and captured prompts."""
|
|
|
|
super().__init__()
|
|
self.evidence_response = evidence_response
|
|
self.ownership_response = ownership_response
|
|
self.blame_response = ["src/foo.py"] if blame_response is None else blame_response
|
|
self.collection_ownership_response = collection_ownership_response
|
|
self.evidence_prompts: list[str] = []
|
|
self.ownership_prompts: list[str] = []
|
|
self.collection_ownership_prompts: list[str] = []
|
|
self.blame_prompts: list[str] = []
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return configured responses for failure extraction and blame prompts."""
|
|
|
|
if "Extract structured failure evidence from this raw test output." in prompt:
|
|
self.evidence_prompts.append(prompt)
|
|
if isinstance(self.evidence_response, str):
|
|
return self.evidence_response
|
|
return json.dumps(self.evidence_response)
|
|
if "Judge Alpha-or-Beta routing for this targeted test collection/discovery/setup/import failure." in prompt:
|
|
self.collection_ownership_prompts.append(prompt)
|
|
response = self.collection_ownership_response
|
|
if response is None:
|
|
response = self.ownership_response
|
|
if response is None:
|
|
return super().complete(prompt, logger)
|
|
if isinstance(response, str):
|
|
return response
|
|
return json.dumps(response)
|
|
if "Judge Alpha-or-Beta routing for this targeted test failure." in prompt:
|
|
self.ownership_prompts.append(prompt)
|
|
if self.ownership_response is None:
|
|
return super().complete(prompt, logger)
|
|
if isinstance(self.ownership_response, str):
|
|
return self.ownership_response
|
|
return json.dumps(self.ownership_response)
|
|
if "Given this test failure, choose which available source files are faulted" in prompt:
|
|
self.blame_prompts.append(prompt)
|
|
return json.dumps(
|
|
{
|
|
"outcome": "blame" if self.blame_response else "none",
|
|
"reason": "configured file blame response",
|
|
"additional": {"source_locations": self.blame_response},
|
|
}
|
|
)
|
|
return super().complete(prompt, logger)
|
|
|
|
class EmptyDependencyStrategy:
|
|
"""EmptyDependencyStrategy reports no deterministic dependency evidence."""
|
|
|
|
@property
|
|
def is_frozen(self) -> bool:
|
|
"""Return whether the strategy is frozen."""
|
|
|
|
return False
|
|
|
|
def freeze(self) -> None:
|
|
"""Allow the empty strategy to satisfy the dependency protocol."""
|
|
|
|
return None
|
|
|
|
def direct_tests_for(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
all_test_files: list[ClassMeta],
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> list[ClassMeta]:
|
|
"""Return no direct tests."""
|
|
|
|
return []
|
|
|
|
def downstream_sources_for(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta],
|
|
src_root: Path,
|
|
recursive: bool = False,
|
|
dependency_cache=None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
"""Return no downstream sources."""
|
|
|
|
return []
|
|
|
|
def upstream_sources_for(
|
|
self,
|
|
files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta],
|
|
src_root: Path,
|
|
test_root: Path,
|
|
recursive: bool = False,
|
|
dependency_cache=None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
"""Return no upstream sources."""
|
|
|
|
return []
|
|
|
|
def joint_beam_sources_for(
|
|
self,
|
|
current_frontier: list[ClassMeta],
|
|
deferred_blames: list[ClassMeta],
|
|
all_source_files: list[ClassMeta],
|
|
src_root: Path,
|
|
) -> list[ClassMeta]:
|
|
"""Return no deferred-guided beam sources."""
|
|
|
|
return []
|
|
|
|
class FailingTargetedRunner(TestRunnerStrategy):
|
|
"""FailingTargetedRunner fails targeted tests and passes all-tests."""
|
|
|
|
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 run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Return a failing targeted test result."""
|
|
|
|
del test_files, max_failures, test_names
|
|
return TestRunResult("targeted", TestRunOutcome.Failure, 1, "failed")
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Return a passing all-tests result."""
|
|
|
|
return TestRunResult("all", TestRunOutcome.Success, 0, "ok")
|
|
|
|
class SequenceTargetedRunner(TestRunnerStrategy):
|
|
"""SequenceTargetedRunner returns configured targeted results in order and then passes."""
|
|
|
|
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: list[TestRunResult], all_tests: TestRunResult | None = None) -> None:
|
|
"""Store targeted results and the all-tests fallback result."""
|
|
|
|
self.results = list(results)
|
|
self.all_tests = all_tests or TestRunResult("all", TestRunOutcome.Success, 0, "ok")
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Return the next targeted result or a passing result when exhausted."""
|
|
|
|
del test_files, max_failures, test_names
|
|
if self.results:
|
|
return self.results.pop(0)
|
|
return TestRunResult("targeted", TestRunOutcome.Success, 0, "ok")
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Return the configured all-tests result."""
|
|
|
|
return self.all_tests
|
|
|
|
class DiscoveryFailingTargetedRunner(FailingTargetedRunner):
|
|
"""DiscoveryFailingTargetedRunner fails targeted tests with configured output."""
|
|
|
|
def __init__(self, output: str) -> None:
|
|
"""Store targeted failure output."""
|
|
|
|
self.output = output
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Return a failing targeted discovery result."""
|
|
|
|
del test_files, max_failures, test_names
|
|
return TestRunResult(
|
|
"targeted",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
self.output,
|
|
stack_trace=[
|
|
line.rstrip()
|
|
for line in self.output.splitlines()
|
|
if line.strip()
|
|
],
|
|
)
|
|
|
|
class PassingRunner(TestRunnerStrategy):
|
|
"""PassingRunner passes targeted tests and all-tests."""
|
|
|
|
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 run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Return a passing targeted test result."""
|
|
|
|
del test_files, max_failures, test_names
|
|
return TestRunResult("targeted", TestRunOutcome.Success, 0, "ok")
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Return a passing all-tests result."""
|
|
|
|
return TestRunResult("all", TestRunOutcome.Success, 0, "ok")
|
|
|
|
class FakeCoverageStrategy:
|
|
"""FakeCoverageStrategy supplies direct-test and branch-coverage evidence."""
|
|
|
|
def __init__(self, branch_coverage: float = 1.0) -> None:
|
|
"""Store the branch coverage value returned by the fake strategy."""
|
|
|
|
self.branch_coverage = branch_coverage
|
|
|
|
def corresponding_tests_for(
|
|
self,
|
|
source_file: ClassMeta,
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> list[ClassMeta]:
|
|
"""Return every supplied test file as corresponding coverage evidence."""
|
|
|
|
return list(test_files)
|
|
|
|
def branch_coverage_for(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> float:
|
|
"""Return the configured branch coverage value."""
|
|
|
|
return self.branch_coverage
|
|
|
|
def coverage_is_sufficient(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> bool:
|
|
"""Return whether configured branch coverage satisfies the gate."""
|
|
|
|
return self.branch_coverage >= MINIMUM_BRANCH_COVERAGE
|
|
|
|
class HandlingAlphaLoop(AlphaLoop):
|
|
"""HandlingAlphaLoop accepts all assigned source files through the protected source seam."""
|
|
|
|
def _run_source_work(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
work_needed_by_location: dict[str, str],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
failing_test=None,
|
|
*,
|
|
is_remediation: bool,
|
|
) -> tuple[list[ClassMeta], str]:
|
|
"""Return assigned source files as handled and record backlog guidance."""
|
|
|
|
del is_remediation
|
|
self.work_needed_by_location = work_needed_by_location
|
|
return list(source_files), "handled source files"
|
|
|
|
class EditingSourceWorker(CodeEditor):
|
|
"""EditingSourceWorker edits the first source file requested by Alpha."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize request and callback tracking."""
|
|
|
|
self.requests: list[AiderEditRequest] = []
|
|
self.new_files_callback: Callable[..., bool] | None = None
|
|
self.file_manager: FileManager | None = None
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Expose this nominal fake as an undecorated editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Write one editable source file and return it as changed."""
|
|
|
|
del logger
|
|
self.requests.append(request)
|
|
path = request.editable_files[0]
|
|
current_repo_exec().write_text(path, "def foo():\n return 2\n")
|
|
return CodeEditorResult([path], "source edited", [])
|
|
|
|
class RecordingSourceWorker(CodeEditor):
|
|
"""RecordingSourceWorker records the Alpha edit request without mutating files."""
|
|
|
|
def __init__(self, events: list[tuple[str, object]]) -> None:
|
|
"""Store shared ordering events and recorded requests."""
|
|
|
|
self.events = events
|
|
self.requests: list[AiderEditRequest] = []
|
|
self.new_files_callback: Callable[..., bool] | None = None
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Expose this nominal fake as an undecorated editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Record the edit request after preflight has run."""
|
|
|
|
del logger
|
|
self.events.append(("edit", list(request.editable_files)))
|
|
self.requests.append(request)
|
|
return CodeEditorResult([], "no changes", [])
|
|
|
|
class NoChangeWorker(CodeEditor):
|
|
"""NoChangeWorker records edit requests without changing files."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize request and callback tracking."""
|
|
|
|
self.requests: list[AiderEditRequest] = []
|
|
self.new_files_callback: Callable[..., bool] | None = None
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Expose this nominal fake as an undecorated editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Return no changed paths."""
|
|
|
|
del logger
|
|
self.requests.append(request)
|
|
return CodeEditorResult([], "no changes", [])
|
|
|
|
class EditingTestWorker(CodeEditor):
|
|
"""EditingTestWorker edits editable test files and records ordering events."""
|
|
|
|
def __init__(self, events: list[str] | None = None) -> None:
|
|
"""Initialize event, request, and callback tracking."""
|
|
|
|
self.events = events if events is not None else []
|
|
self.requests: list[AiderEditRequest] = []
|
|
self.new_files_callback: Callable[..., bool] | None = None
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Expose this nominal fake as an undecorated editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Write editable test files and return them as changed."""
|
|
|
|
del logger
|
|
self.events.append("edit")
|
|
self.requests.append(request)
|
|
for path in request.editable_files:
|
|
current_repo_exec().write_text(path, "def test_foo():\n assert True\n")
|
|
return CodeEditorResult(list(request.editable_files), "test edited", [])
|
|
|
|
class EventPassingRunner(PassingRunner):
|
|
"""EventPassingRunner records test execution order while passing tests."""
|
|
|
|
def __init__(self, events: list[str]) -> None:
|
|
"""Store the shared ordering event list."""
|
|
|
|
self.events = events
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Record targeted execution and return a passing result."""
|
|
|
|
del test_names
|
|
self.events.append("targeted")
|
|
return super().run_tests_for(test_files, max_failures=max_failures)
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Record all-test execution and return a passing result."""
|
|
|
|
self.events.append("all")
|
|
return super().run_all_tests_under(test_root)
|
|
|
|
|
|
def _language_strategy_set(
|
|
file_manager: FileManager,
|
|
*,
|
|
runner=None,
|
|
coverage=None,
|
|
dependency_strategy=None,
|
|
) -> LanguageStrategySet:
|
|
"""Return one mutable language strategy set for deterministic loop tests."""
|
|
|
|
builder = cast(
|
|
Callable[[FileManager], LanguageStrategySet],
|
|
language_strategy_set_factory.build_language_set("python"),
|
|
)
|
|
strategy_set = builder(file_manager)
|
|
if dependency_strategy is not None:
|
|
strategy_set.bindings[DependencyDiscoveryStrategy] = dependency_strategy
|
|
if runner is not None:
|
|
strategy_set.bindings[TestRunnerStrategy] = runner
|
|
if coverage is not None:
|
|
strategy_set.bindings[CoverageStrategy] = coverage
|
|
return strategy_set
|
|
|
|
|
|
def _idle_language_strategy_set() -> LanguageStrategySet:
|
|
"""Return one inert language strategy set for constructor-only tests."""
|
|
|
|
class _IdleNamingStrategy:
|
|
"""_IdleNamingStrategy fails if a constructor-only test starts using naming behavior."""
|
|
|
|
def get_idiomatic_test_name_for(self, class_meta: ClassMeta):
|
|
raise AssertionError("Constructor-only tests should not ask for test naming.")
|
|
|
|
class _IdleSourceDiscovery:
|
|
"""_IdleSourceDiscovery fails if a constructor-only test starts using beat discovery."""
|
|
|
|
def searchable_beats(self, location: str, content: str):
|
|
raise AssertionError("Constructor-only tests should not ask for searchable beats.")
|
|
|
|
return LanguageStrategySet(
|
|
bindings={
|
|
SourceDiscoveryStrategy: _IdleSourceDiscovery(),
|
|
TestRunnerStrategy: object(),
|
|
CoverageStrategy: FakeCoverageStrategy(),
|
|
BiDependencyDiscoveryStrategy: object(),
|
|
NamingStrategy: _IdleNamingStrategy(),
|
|
}
|
|
)
|
|
|
|
class FakeMissingTestWorker(CodeEditor):
|
|
"""FakeMissingTestWorker edits prepared missing test files."""
|
|
|
|
def __init__(self, new_files_callback: Callable[[dict[Path, str]], bool]) -> None:
|
|
"""Store the new-file callback and initialize call tracking."""
|
|
|
|
self.new_files_callback = new_files_callback
|
|
self.requests: list[AiderEditRequest] = []
|
|
self.callback_payloads: list[dict[Path, str]] = []
|
|
|
|
@property
|
|
def inner(self) -> CodeEditor | None:
|
|
"""Expose this nominal fake as an undecorated editor."""
|
|
|
|
return None
|
|
|
|
def edit(
|
|
self,
|
|
request: AiderEditRequest,
|
|
logger: logging.Logger | None = None,
|
|
) -> CodeEditorResult:
|
|
"""Write editable files and return changed paths."""
|
|
|
|
del logger
|
|
self.requests.append(request)
|
|
editable = list(request.editable_files)
|
|
payload = {
|
|
path: "prepared editable"
|
|
for path in editable
|
|
if current_repo_exec().exists(path) and not current_repo_exec().read_text(path)
|
|
}
|
|
for path in editable:
|
|
current_repo_exec().write_text(path, "def test_missing_coverage_for_source_file():\n assert True\n")
|
|
|
|
if payload and not self.new_files_callback(payload):
|
|
raise ValueError("callback rejected requested files")
|
|
|
|
self.callback_payloads.append(payload)
|
|
return CodeEditorResult(editable, "test edited", [])
|
|
|
|
class FakeGate:
|
|
"""FakeGate records that StabilizationLoop preserves injected parent gates."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the fake gate call counters."""
|
|
|
|
self.may_enter_calls = 0
|
|
|
|
def may_enter(self) -> bool:
|
|
"""Return True and record one may-enter call."""
|
|
|
|
self.may_enter_calls += 1
|
|
return True
|
|
|
|
def should_retry(self) -> bool:
|
|
"""Return False for retry checks."""
|
|
|
|
return False
|
|
|
|
def should_continue(self) -> bool:
|
|
"""Return False for continuation checks."""
|
|
|
|
return False
|
|
|
|
|
|
class TestAlphaContext(AlphaContext):
|
|
"""Alpha context with the test-only collaborators exercised by loop scenarios."""
|
|
|
|
new_files_callback: Callable[..., bool]
|
|
topology_summary: str
|
|
test_runner_strategy: TestRunnerStrategy
|
|
|
|
|
|
class TestBetaContext(BetaContext):
|
|
"""Beta context with the test-only runner exercised by loop scenarios."""
|
|
|
|
test_runner_strategy: TestRunnerStrategy
|
|
|
|
|
|
class TestFileManager(FileManager):
|
|
"""FileManager fixture whose seeded metadata must be present in each scenario."""
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Initialize stationary in-memory observations before indexing the scenario tree."""
|
|
|
|
self.squash_summaries: list[str] = []
|
|
self.diffs_by_location: dict[str, str] = {}
|
|
super().__post_init__()
|
|
|
|
def git_diff_for(self, location: str | Path) -> str:
|
|
"""Return the explicitly seeded stationary diff for one scenario location."""
|
|
|
|
return self.diffs_by_location.get(str(location), "")
|
|
|
|
def squash_to_upstream(self, summary_hint: str = "") -> None:
|
|
"""Record a logical squash without invoking a repository command."""
|
|
|
|
self.squash_summaries.append(summary_hint)
|
|
|
|
def set_diff(self, location: str | Path, diff: str) -> None:
|
|
"""Seed one deterministic source diff for a semantic scenario."""
|
|
|
|
self.diffs_by_location[str(location)] = diff
|
|
|
|
def require_src(self, location: str) -> ClassMeta:
|
|
"""Return one seeded source record or fail the scenario at its setup boundary."""
|
|
|
|
current = super().current_src(location)
|
|
if current is None:
|
|
raise AssertionError(f"Test setup did not seed source metadata for {location}.")
|
|
return current
|
|
|
|
def require_test(self, location: str) -> ClassMeta:
|
|
"""Return one seeded test record or fail the scenario at its setup boundary."""
|
|
|
|
current = super().current_test(location)
|
|
if current is None:
|
|
raise AssertionError(f"Test setup did not seed test metadata for {location}.")
|
|
return current
|
|
|
|
|
|
def _file_manager(repo_root: Path) -> TestFileManager:
|
|
"""Create a FileManager over a small source/test tree."""
|
|
|
|
repo_exec = MemoryRepoExec(repo_root)
|
|
src_root = repo_root / "src"
|
|
test_root = repo_root / "tests"
|
|
repo_exec.mkdir(src_root)
|
|
repo_exec.mkdir(test_root)
|
|
repo_exec.write_text(src_root / "foo.py", "def foo():\n return 1\n")
|
|
repo_exec.write_text(
|
|
test_root / "test_foo.py",
|
|
"from foo import foo\n\n"
|
|
"def test_foo():\n"
|
|
" assert foo() == 1\n"
|
|
)
|
|
configure_resume_settings(
|
|
hydrate=False,
|
|
checkpoint_dir=repo_root / ".band-aider-chkpts",
|
|
repo_exec=repo_exec,
|
|
)
|
|
file_manager = TestFileManager(
|
|
parser=FakeParser(src_root, test_root),
|
|
client=FakeClient(),
|
|
repo_root=repo_root,
|
|
src_root=src_root,
|
|
test_root=test_root,
|
|
actor="test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
prompt="goal",
|
|
repo_exec=repo_exec,
|
|
)
|
|
LanguageStrategySet.set_instance(
|
|
cast(
|
|
Callable[..., LanguageStrategySet],
|
|
language_strategy_set_factory.build_language_set("python"),
|
|
)
|
|
)
|
|
return file_manager
|
|
|
|
|
|
def _scoped_file_manager(upstream: TestFileManager) -> TestFileManager:
|
|
"""Create the real stabilization scope over the same in-memory repository."""
|
|
|
|
scoped = TestFileManager(
|
|
parser=upstream.parser,
|
|
client=upstream.client,
|
|
repo_root=upstream.repo_root,
|
|
src_root=upstream.src_root,
|
|
test_root=upstream.test_root,
|
|
actor=upstream.actor,
|
|
prompt_formatter_builder=upstream.prompt_formatter_builder,
|
|
prompt=upstream.prompt,
|
|
repo_exec=upstream.repo_exec,
|
|
)
|
|
scoped._frozen = upstream._frozen
|
|
return scoped
|
|
|
|
def _alpha_context(
|
|
file_manager: FileManager,
|
|
backlog: _BackLog,
|
|
source: ClassMeta,
|
|
test: ClassMeta,
|
|
) -> TestAlphaContext:
|
|
"""Create an AlphaContext over the real ContributorBacklog presentation layer."""
|
|
|
|
if not backlog.inspection and not backlog.remediation:
|
|
inspection = InspectionBacklog(
|
|
source,
|
|
"inspect source",
|
|
downstream_classes=[source],
|
|
)
|
|
backlog.queue_inspection(inspection)
|
|
|
|
contributor_backlog = ContributorBacklog(
|
|
backlog=backlog,
|
|
get_tests=lambda source_files: [test],
|
|
file_manager=file_manager,
|
|
parser=file_manager.parser,
|
|
language_strategy_set=_language_strategy_set(file_manager, runner=PassingRunner()),
|
|
)
|
|
context = TestAlphaContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
file_manager=file_manager,
|
|
backlog=contributor_backlog,
|
|
)
|
|
context.new_files_callback = lambda **files: False
|
|
context.topology_summary = "alpha"
|
|
context.test_runner_strategy = PassingRunner()
|
|
return context
|
|
|
|
def _alpha_gate(context: AlphaContext) -> MechanicalLoopGate:
|
|
"""Create a simple mechanical Alpha gate for one test pass."""
|
|
|
|
return MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: False,
|
|
)
|
|
|
|
def _testing_item(source: ClassMeta, test: ClassMeta) -> TestingBacklog:
|
|
"""Create one testing backlog item for a source and test pair."""
|
|
|
|
return TestingBacklog(
|
|
focused_spotlight=test,
|
|
summary="validate source",
|
|
downstream_classes=[source],
|
|
change_type=ChangeType.Update,
|
|
)
|
|
|
|
def _beta_context(
|
|
file_manager: FileManager,
|
|
qa_backlog: QABacklog,
|
|
runner,
|
|
) -> TestBetaContext:
|
|
"""Create a BetaContext over the supplied QA presentation layer."""
|
|
|
|
context = TestBetaContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
file_manager=file_manager,
|
|
backlog=qa_backlog,
|
|
)
|
|
context.test_runner_strategy = runner
|
|
return context
|
|
|
|
def _beta_gate(context: BetaContext) -> MechanicalLoopGate:
|
|
"""Create a simple mechanical Beta gate for one test pass."""
|
|
|
|
return MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
|
|
def _stabilization_loop(
|
|
file_manager: TestFileManager,
|
|
backlog: _BackLog,
|
|
coverage: FakeCoverageStrategy,
|
|
worker=None,
|
|
cot_index: int = 0,
|
|
total_cot: int = 2,
|
|
) -> StabilizationLoop:
|
|
"""Create a StabilizationLoop with fake deterministic strategies."""
|
|
|
|
source = file_manager.src_classes[0]
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=source,
|
|
cot_index=cot_index,
|
|
total_cot=total_cot,
|
|
completion=backlog.completion,
|
|
backlog=backlog,
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
return StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=coverage,
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
aider_worker=worker or EditingSourceWorker(),
|
|
)
|
|
|
|
__all__ = [
|
|
'annotations',
|
|
'inspect',
|
|
'json',
|
|
'Path',
|
|
'pytest',
|
|
'AlphaContext',
|
|
'BetaContext',
|
|
'TargetedFailureStory',
|
|
'ClassMeta',
|
|
'StabilizationContext',
|
|
'BackLog',
|
|
'ChangeType',
|
|
'ContributorBacklog',
|
|
'FailedQAItem',
|
|
'FailingTest',
|
|
'InspectionBacklog',
|
|
'QABacklog',
|
|
'TestingBacklog',
|
|
'FileManager',
|
|
'TestFileManager',
|
|
'_scoped_file_manager',
|
|
'PythonAstDependencyStrategy',
|
|
'PythonCoverageStrategy',
|
|
'RetryVibeCheck',
|
|
'ReasonableVibeCheck',
|
|
'TestRunResult',
|
|
'AlphaLoop',
|
|
'BetaLoop',
|
|
'MechanicalLoopGate',
|
|
'StabilizationLoop',
|
|
'AiderConfig',
|
|
'FakeClient',
|
|
'FakeParser',
|
|
'PreflightNewSourceClient',
|
|
'AlphaPreflightClient',
|
|
'WorkNeededPromptClient',
|
|
'BetaPreflightClient',
|
|
'PostflightBlameClient',
|
|
'FailureEvidenceClient',
|
|
'EmptyDependencyStrategy',
|
|
'FailingTargetedRunner',
|
|
'SequenceTargetedRunner',
|
|
'DiscoveryFailingTargetedRunner',
|
|
'PassingRunner',
|
|
'FakeCoverageStrategy',
|
|
'HandlingAlphaLoop',
|
|
'EditingSourceWorker',
|
|
'RecordingSourceWorker',
|
|
'NoChangeWorker',
|
|
'EditingTestWorker',
|
|
'EventPassingRunner',
|
|
'FakeMissingTestWorker',
|
|
'FakeGate',
|
|
'_idle_language_strategy_set',
|
|
'_language_strategy_set',
|
|
'_file_manager',
|
|
'_alpha_context',
|
|
'_alpha_gate',
|
|
'_testing_item',
|
|
'_beta_context',
|
|
'_beta_gate',
|
|
'_stabilization_loop',
|
|
]
|