1280 lines
55 KiB
Python
1280 lines
55 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Scope and dependency-graph behavior over a wholly in-memory repository."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Sequence, TextIO
|
|
from unittest.mock import Mock
|
|
from band_aider.judgement.vibe_check import ReasonableVibeCheck
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
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 import provenance
|
|
from band_aider.context.provenance import Phase, Provenance
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.context.work_completion import WorkCompletion
|
|
from band_aider.context.working_backlog import (
|
|
BackLog,
|
|
ChangeType,
|
|
ContributorBacklog,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
QABacklog,
|
|
TestingBacklog,
|
|
curate_failure_context_sources,
|
|
)
|
|
from band_aider.judgement.code_dependency_graph import CodeDependencyGraph, DependencySelector
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.coverage_strategy import CoverageStrategy
|
|
from band_aider.judgement.dependency_discovery_strategy import DependencyTraversalCache
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.file_salience_check import FileSalienceCheck
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.python_ast_dependency_graph_maintainer import (
|
|
PythonAstDependencyGraphMaintainer,
|
|
)
|
|
from band_aider.judgement.python_ast_dependency_strategy import PythonAstDependencyStrategy
|
|
from band_aider.judgement.python_coverage_strategy import PythonCoverageStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy
|
|
from band_aider.loop.mechanical_loop_gate import MechanicalLoopGate
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
from band_aider.strategies import LanguageStrategySet, failure_source_candidate_selection_strategy_factory
|
|
from band_aider.strategies.python import build_python_language_strategy_set
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
from tests.topology_test_support import service_topology
|
|
|
|
|
|
pytestmark = pytest.mark.usefixtures("without_embedding_model")
|
|
|
|
|
|
def _meta(location: str) -> ClassMeta:
|
|
"""Create correctly owned metadata for one repository-relative path."""
|
|
|
|
ownership = ContentType.Test if Path(location).parts[0] == "tests" else ContentType.Source
|
|
return ClassMeta(location, f"summary {location}", [], {}, content_type=ownership)
|
|
|
|
|
|
class _Parser(CodeParser):
|
|
"""Create deterministic metadata from lexical memory-repository paths."""
|
|
|
|
def __init__(self, src_root: Path, test_root: Path) -> None:
|
|
"""Store the two ownership roots used by the parser."""
|
|
|
|
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 from the supplied path only."""
|
|
|
|
path = Path(file_name)
|
|
root = self.src_root if path == self.src_root or self.src_root in path.parents else self.test_root
|
|
return ClassMeta(
|
|
str(path.relative_to(root)),
|
|
f"summary {path.name}",
|
|
[prompt] if prompt else [],
|
|
{},
|
|
content_type=content_type,
|
|
)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Apply the same parser rule to an open stream."""
|
|
|
|
del stream
|
|
return self.parse_file(file_name, prompt, content_type)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Parse every explicitly requested path in deterministic order."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _Client(OpenAICompatibleClient):
|
|
"""Supply stationary valid model responses without external transport."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
cdc: str = "metadata refreshed",
|
|
precedence_winner: str = "",
|
|
) -> None:
|
|
"""Store the exact scenario response values."""
|
|
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
self._cdc = cdc
|
|
self._precedence_winner = precedence_winner
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the response selected by the production judgement role."""
|
|
|
|
del logger
|
|
if (
|
|
"Compare exactly two preselected production candidates" in prompt
|
|
or "Inspect the sole preselected production candidate" in prompt
|
|
):
|
|
return json.dumps(
|
|
{
|
|
"location": self._precedence_winner,
|
|
"reason": [
|
|
f"def culprit_{Path(self._precedence_winner).stem}():"
|
|
],
|
|
"explanation": (
|
|
"The configured candidate creates the invalid state before "
|
|
"the other candidate observes it."
|
|
),
|
|
}
|
|
)
|
|
if "BETA CONTRACT GUIDANCE ROLE" in prompt:
|
|
return json.dumps(
|
|
{
|
|
"needs_beta_work": True,
|
|
"reason": "The direct regression remains relevant.",
|
|
"new_contract": "The source behavior is current.",
|
|
"invalidated_contract": "",
|
|
"regression_obligation": "Keep the direct regression executable.",
|
|
"existing_tests_to_update": [],
|
|
"existing_tests_to_preserve": [],
|
|
"new_tests": [],
|
|
"failure_mode": "",
|
|
"beta_instruction": "Maintain the selected direct regression.",
|
|
}
|
|
)
|
|
if "Return exactly one JSON array" in prompt:
|
|
return "[]"
|
|
return json.dumps({"cdc": self._cdc})
|
|
|
|
|
|
class _EvidenceClient(_Client):
|
|
"""Accept a CDC request only when its semantic evidence reaches the client."""
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Enforce history and diff inputs through the fake's stationary contract."""
|
|
|
|
if "existing change history" not in prompt:
|
|
raise AssertionError("CDC request omitted current ClassMeta history.")
|
|
if "Current git diff for this file, authoritative for what changed:" not in prompt:
|
|
raise AssertionError("CDC request omitted current file-scoped diff evidence.")
|
|
return super().complete(prompt, logger)
|
|
|
|
|
|
class _MemoryFileManager(FileManager):
|
|
"""Use normal FileManager behavior with scenario-owned Git diff evidence."""
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Initialize explicit diff evidence before normal source discovery."""
|
|
|
|
self._diff_by_location: dict[str, str] = {}
|
|
super().__post_init__()
|
|
|
|
def set_diff(self, location: str, diff: str) -> None:
|
|
"""Set the current diff evidence for one location."""
|
|
|
|
self._diff_by_location[location] = diff
|
|
|
|
def git_diff_for(self, location: str | Path) -> str:
|
|
"""Return only explicitly supplied in-memory diff evidence."""
|
|
|
|
return self._diff_by_location.get(self.repo_location(location), "")
|
|
|
|
|
|
class _Runner(TestRunnerStrategy):
|
|
"""Return a deterministic test result and enforce bounded direct execution."""
|
|
|
|
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, result: TestRunResult | None = None) -> None:
|
|
"""Store the targeted result to be observed by production workflow code."""
|
|
|
|
self._result = result or TestRunResult("targeted", TestRunOutcome.Success, 0, "ok")
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Require the workflow's one-failure bound and return the stationary result."""
|
|
|
|
del test_files, test_names
|
|
if max_failures != 1:
|
|
raise AssertionError("Direct candidate runs must remain bounded to one failure.")
|
|
return self._result
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Return a passing whole-suite result without starting a child process."""
|
|
|
|
del test_root
|
|
return TestRunResult("all", TestRunOutcome.Success, 0, "ok")
|
|
|
|
|
|
class _NoCoverage(CoverageStrategy):
|
|
"""Represent a stable no-coverage answer without process execution."""
|
|
|
|
def corresponding_tests_for(
|
|
self,
|
|
source_file: ClassMeta,
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> list[ClassMeta]:
|
|
"""Return no direct coverage candidates."""
|
|
|
|
del source_file, test_files, repo_root, src_root, test_root
|
|
return []
|
|
|
|
def coverage_is_sufficient(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> bool:
|
|
"""Return the fixed insufficient coverage disposition."""
|
|
|
|
del source_files, test_files, repo_root, src_root, test_root
|
|
return False
|
|
|
|
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 fixed zero branch-coverage observation."""
|
|
|
|
del source_files, test_files, repo_root, src_root, test_root
|
|
return 0.0
|
|
|
|
|
|
class _DirectMapStrategy(PythonAstDependencyStrategy):
|
|
"""Provide direct upstream relationships while retaining the real strategy contract."""
|
|
|
|
def __init__(self, edges: dict[str, list[ClassMeta]]) -> None:
|
|
"""Store explicit direct edges over a nominal in-memory maintainer."""
|
|
|
|
super().__init__(PythonAstDependencyGraphMaintainer(repo_exec=MemoryRepoExec()))
|
|
self._edges = edges
|
|
|
|
def upstream_sources_for(
|
|
self,
|
|
files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
test_root: Path | None = None,
|
|
recursive: bool = False,
|
|
dependency_cache: DependencyTraversalCache | None = None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
"""Return the configured direct upstream union only."""
|
|
|
|
del all_source_files, src_root, test_root, recursive, dependency_cache, selector
|
|
found = {
|
|
source.location: source
|
|
for file in files
|
|
for source in self._edges.get(file.location, [])
|
|
}
|
|
return [found[location] for location in sorted(found)]
|
|
|
|
|
|
class _ReadBoundaryRepoExec(MemoryRepoExec):
|
|
"""Reject a repository content read after the graph-build phase closes."""
|
|
|
|
def __init__(self, root: Path) -> None:
|
|
"""Allow reads during setup and forbid them during graph-only queries."""
|
|
|
|
super().__init__(root)
|
|
self.reads_allowed = True
|
|
|
|
def read_text(self, path: Path, *, encoding: str = "utf-8") -> str:
|
|
"""Fail the scenario if a query unexpectedly performs repository IO."""
|
|
|
|
if not self.reads_allowed:
|
|
raise AssertionError("Graph query attempted a repository content read.")
|
|
return super().read_text(path, encoding=encoding)
|
|
|
|
|
|
def _manager(
|
|
memory_root: Path,
|
|
source_files: dict[str, str],
|
|
test_files: dict[str, str],
|
|
*,
|
|
client: OpenAICompatibleClient | None = None,
|
|
repo_exec: MemoryRepoExec | None = None,
|
|
) -> _MemoryFileManager:
|
|
"""Build a real FileManager over only in-memory source and test files."""
|
|
|
|
executor = repo_exec or MemoryRepoExec(memory_root)
|
|
src_root = memory_root / "src"
|
|
test_root = memory_root / "tests"
|
|
executor.mkdir(src_root)
|
|
executor.mkdir(test_root)
|
|
for location, content in source_files.items():
|
|
executor.write_text(src_root / location, content)
|
|
for location, content in test_files.items():
|
|
executor.write_text(test_root / location, content)
|
|
LanguageStrategySet.set_instance(build_python_language_strategy_set)
|
|
return _MemoryFileManager(
|
|
parser=_Parser(src_root, test_root),
|
|
client=client or _Client(),
|
|
repo_root=memory_root,
|
|
src_root=src_root,
|
|
test_root=test_root,
|
|
actor="test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
prompt="goal",
|
|
repo_exec=executor,
|
|
)
|
|
|
|
|
|
def _source(file_manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return one required current source metadata record."""
|
|
|
|
current = file_manager.current_src(location)
|
|
assert current is not None
|
|
return current
|
|
|
|
|
|
def _test(file_manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return one required current test metadata record."""
|
|
|
|
current = file_manager.current_test(location)
|
|
assert current is not None
|
|
return current
|
|
|
|
|
|
def _focused(item: FailedQAItem) -> ClassMeta:
|
|
"""Return the required focused metadata from a workflow item."""
|
|
|
|
focused = item.focused_spotlight
|
|
assert focused is not None
|
|
return focused
|
|
|
|
|
|
def _maintainer(file_manager: FileManager) -> PythonAstDependencyGraphMaintainer:
|
|
"""Return the concrete Python maintainer installed by strategy construction."""
|
|
|
|
maintainer = file_manager.dependency_graph_maintainer
|
|
assert isinstance(maintainer, PythonAstDependencyGraphMaintainer)
|
|
return maintainer
|
|
|
|
|
|
def _strategy_set(
|
|
file_manager: FileManager,
|
|
*,
|
|
runner: TestRunnerStrategy | None = None,
|
|
dependency: PythonAstDependencyStrategy | None = None,
|
|
) -> LanguageStrategySet:
|
|
"""Build concrete Python strategies and replace only owned scenario boundaries."""
|
|
|
|
strategy_set = build_python_language_strategy_set(file_manager)
|
|
if runner is not None:
|
|
strategy_set.bindings[TestRunnerStrategy] = runner
|
|
if dependency is not None:
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import BiDependencyDiscoveryStrategy
|
|
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
|
|
|
|
strategy_set.bindings[DependencyDiscoveryStrategy] = dependency
|
|
strategy_set.bindings[BiDependencyDiscoveryStrategy] = dependency
|
|
return strategy_set
|
|
|
|
|
|
def _loop(
|
|
file_manager: FileManager,
|
|
root: ClassMeta,
|
|
*,
|
|
topology: TopologicalLiveliness | None = None,
|
|
dependency: PythonAstDependencyStrategy | None = None,
|
|
) -> tuple[StabilizationLoop, StabilizationContext]:
|
|
"""Build a normally initialized stabilization loop over a memory manager."""
|
|
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
file_manager=file_manager,
|
|
root_change=root,
|
|
topology=topology or TopologicalLiveliness(),
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: False),
|
|
context=context,
|
|
language_strategy_set=_strategy_set(file_manager, runner=_Runner(), dependency=dependency),
|
|
)
|
|
return loop, context
|
|
|
|
|
|
def _contributor(
|
|
file_manager: FileManager,
|
|
backlog: BackLog,
|
|
direct_tests: list[ClassMeta],
|
|
*,
|
|
runner: TestRunnerStrategy | None = None,
|
|
) -> ContributorBacklog:
|
|
"""Build the normal contributor presentation layer for one scenario."""
|
|
|
|
return ContributorBacklog(
|
|
backlog=backlog,
|
|
get_tests=lambda _sources: list(direct_tests),
|
|
file_manager=file_manager,
|
|
parser=file_manager.parser,
|
|
language_strategy_set=_strategy_set(file_manager, runner=runner or _Runner()),
|
|
)
|
|
|
|
|
|
def test_refresh_source_normalizes_location_and_updates_graph(memory_root: Path) -> None:
|
|
"""Refreshing a source preserves its owned location and replaces graph edges."""
|
|
|
|
manager = _manager(memory_root, {"a.py": "from b import b\n", "b.py": "pass\n", "c.py": "pass\n"}, {})
|
|
stale = _source(manager, "src/a.py")
|
|
manager.repo_exec.write_text(manager.source_path(stale.location), "from c import c\n")
|
|
|
|
refreshed = manager.refresh_src(stale.location, "repair")
|
|
|
|
assert refreshed.location == "src/a.py"
|
|
assert _maintainer(manager).graph.source_dependencies[refreshed.location] == {"src/c.py"}
|
|
|
|
|
|
def test_refresh_test_normalizes_location_and_marks_test_dirty(memory_root: Path) -> None:
|
|
"""Refreshing a test keeps it in test metadata and records its dirty state."""
|
|
|
|
manager = _manager(memory_root, {"app.py": "def app():\n return 1\n"}, {"test_app.py": "from app import app\n"})
|
|
|
|
refreshed = manager.refresh_test("tests/test_app.py", "repair")
|
|
|
|
assert refreshed.location == "tests/test_app.py"
|
|
assert manager.test_dirty[refreshed.location] is True
|
|
assert refreshed is _test(manager, refreshed.location)
|
|
|
|
|
|
def test_refresh_rejects_malformed_content_type(memory_root: Path) -> None:
|
|
"""A malformed metadata owner cannot cross the source/test refresh boundary."""
|
|
|
|
manager = _manager(memory_root, {"app.py": "pass\n"}, {})
|
|
malformed = _source(manager, "src/app.py")
|
|
object.__setattr__(malformed, "content_type", object())
|
|
|
|
with pytest.raises(ValueError, match="Unsupported ClassMeta content type"):
|
|
manager.refresh_class_meta(malformed, "repair")
|
|
|
|
|
|
def test_resync_discovers_only_python_source_and_test_files(memory_root: Path) -> None:
|
|
"""Language discovery omits irrelevant file types from both managed domains."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "pass\n", "notes.md": "ignored\n"}, {"test_foo.py": "pass\n", "notes.txt": "ignored\n"})
|
|
|
|
assert [item.location for item in manager.src_classes] == ["src/foo.py"]
|
|
assert [item.location for item in manager.test_classes] == ["tests/test_foo.py"]
|
|
|
|
|
|
def test_cdc_uses_carried_history_and_stationary_diff(memory_root: Path) -> None:
|
|
"""CDC refresh receives history and scenario diff evidence without Git execution."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "def foo():\n return 1\n"}, {}, client=_EvidenceClient(cdc="source changed"))
|
|
source = _source(manager, "src/foo.py")
|
|
source.cdc.append((datetime.now(), "actor", "existing change history"))
|
|
manager.set_diff(source.location, "- return 1\n+ return 2")
|
|
manager.repo_exec.write_text(manager.source_path(source.location), "def foo():\n return 2\n")
|
|
latest = manager.parser.parse_file(str(manager.source_path(source.location)), "goal")
|
|
latest.location = source.location
|
|
|
|
manager.upsert_src(latest, "goal")
|
|
|
|
assert _source(manager, source.location).cdc[-1][2] == "source changed"
|
|
|
|
|
|
def test_salience_reader_delegates_to_repo_exec(memory_root: Path) -> None:
|
|
"""Salience content lookup reads through the repository execution boundary."""
|
|
|
|
manager = _manager(memory_root, {"pkg/foo.py": "def foo():\n return 1\n"}, {})
|
|
check = FileSalienceCheck(
|
|
changes=[],
|
|
prompt="goal",
|
|
repo_root=manager.repo_root,
|
|
repo_exec=manager.repo_exec,
|
|
client=_Client(),
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
|
|
assert "def foo" in check._read_file("src/pkg/foo.py")
|
|
|
|
|
|
def test_stabilization_adds_direct_upstream_and_surrogate_context(memory_root: Path) -> None:
|
|
"""Inspection receives direct closed upstreams and a root surrogate as readonly context."""
|
|
|
|
manager = _manager(memory_root, {"a.py": "pass\n", "surrogate.py": "pass\n", "b.py": "pass\n"}, {})
|
|
root = _source(manager, "src/a.py")
|
|
surrogate = _source(manager, "src/surrogate.py")
|
|
focused = _source(manager, "src/b.py")
|
|
topology = TopologicalLiveliness(frontiers=[[focused]], bfs_depth=1, visited={root.location})
|
|
loop, context = _loop(manager, root, topology=topology, dependency=_DirectMapStrategy({focused.location: [root]}))
|
|
context.surrogate_roots = [surrogate]
|
|
|
|
loop._enqueue_current_frontier_to_backlog()
|
|
|
|
assert context.backlog is not None
|
|
item = context.backlog.inspection[0]
|
|
assert [source.location for source in item.upstream_classes] == [root.location, surrogate.location]
|
|
assert [source.location for source in item.downstream_classes] == [focused.location]
|
|
|
|
|
|
def test_stabilization_excludes_transitive_closed_ancestors(memory_root: Path) -> None:
|
|
"""Only a direct prerequisite reaches the active inspection work item."""
|
|
|
|
manager = _manager(memory_root, {"root.py": "pass\n", "direct.py": "pass\n", "leaf.py": "pass\n"}, {})
|
|
root = _source(manager, "src/root.py")
|
|
direct = _source(manager, "src/direct.py")
|
|
leaf = _source(manager, "src/leaf.py")
|
|
topology = TopologicalLiveliness(frontiers=[[direct], [leaf]], bfs_depth=2, visited={root.location, direct.location})
|
|
loop, context = _loop(manager, root, topology=topology, dependency=_DirectMapStrategy({leaf.location: [direct]}))
|
|
|
|
loop._enqueue_current_frontier_to_backlog()
|
|
|
|
assert context.backlog is not None
|
|
assert [source.location for source in context.backlog.inspection[0].upstream_classes] == [direct.location]
|
|
|
|
|
|
def test_contributor_copies_explicit_topology_scope(memory_root: Path) -> None:
|
|
"""Contributor-created regression work copies supplied upstream/downstream scope."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "def foo():\n return 2\n", "base.py": "pass\n", "downstream.py": "pass\n"}, {"test_foo.py": "from foo import foo\n\ndef test_foo():\n assert foo() == 2\n"})
|
|
source = _source(manager, "src/foo.py")
|
|
base = _source(manager, "src/base.py")
|
|
downstream = _source(manager, "src/downstream.py")
|
|
test = _test(manager, "tests/test_foo.py")
|
|
manager.set_diff(source.location, "source change")
|
|
manager.src_dirty[source.location] = True
|
|
|
|
created = _contributor(manager, BackLog(completion={}), [test]).create_outbound_for(source, upstream_classes=[base], downstream_classes=[source, downstream], original_cot="goal")
|
|
|
|
assert len(created) == 1
|
|
assert created[0].upstream_classes == [base]
|
|
assert created[0].downstream_classes == [source, downstream]
|
|
|
|
|
|
def test_contributor_creates_idiomatic_missing_coverage_ticket(memory_root: Path) -> None:
|
|
"""No direct test yields a focused deterministic create ticket."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "def foo():\n return 1\n"}, {})
|
|
source = _source(manager, "src/foo.py")
|
|
manager.set_diff(source.location, "source change")
|
|
manager.src_dirty[source.location] = True
|
|
|
|
created = _contributor(manager, BackLog(completion={}), []).create_outbound_for(source, original_cot="goal")
|
|
|
|
assert len(created) == 1
|
|
assert created[0].focused_spotlight is not None
|
|
assert created[0].focused_spotlight.location == "tests/test_foo.py"
|
|
assert created[0].change_type is ChangeType.Create
|
|
|
|
|
|
def test_alpha_claim_rolls_back_full_source_completion_tree() -> None:
|
|
"""An uncommitted Alpha scope clears its source completion and child tests."""
|
|
|
|
source = _meta("src/foo.py")
|
|
completion = {source.location: WorkCompletion(location=source.location, tests={"tests/test_foo.py": WorkCompletion(location="tests/test_foo.py", completed=True)})}
|
|
backlog = BackLog(completion=completion, inspection=[InspectionBacklog(source, "inspect")])
|
|
claimed = backlog.pop_inspection()
|
|
assert claimed is not None
|
|
|
|
with claimed:
|
|
pass
|
|
|
|
assert source.location not in completion
|
|
|
|
|
|
def test_beta_claim_rolls_back_only_selected_test_completion(memory_root: Path) -> None:
|
|
"""An uncommitted Beta scope retains unrelated completed test children."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "pass\n"}, {"test_foo.py": "pass\n"})
|
|
source = _meta("src/foo.py")
|
|
test = _meta("tests/test_foo.py")
|
|
completion = {source.location: WorkCompletion(location=source.location, completed=True, tests={"tests/test_other.py": WorkCompletion(location="tests/test_other.py", completed=True)})}
|
|
backlog = BackLog(completion=completion, testing=[TestingBacklog(focused_spotlight=test, summary="validate", downstream_classes=[source])])
|
|
qa = QABacklog(backlog, manager.client, manager, manager.parser)
|
|
claimed = backlog.pop_testing()
|
|
assert claimed is not None
|
|
|
|
with claimed:
|
|
qa.claim_testing_pair(claimed.value)
|
|
|
|
retained = completion[source.location].tests
|
|
assert retained is not None
|
|
assert "tests/test_other.py" in retained
|
|
assert test.location not in retained
|
|
|
|
|
|
def test_testing_enqueue_skips_pair_already_pending() -> None:
|
|
"""A queued source/test pair remains represented by one testing item."""
|
|
|
|
source = _meta("src/foo.py")
|
|
test = _meta("tests/test_foo.py")
|
|
pending = TestingBacklog(
|
|
focused_spotlight=test,
|
|
summary="first validation",
|
|
downstream_classes=[source],
|
|
)
|
|
duplicate = TestingBacklog(
|
|
focused_spotlight=test,
|
|
summary="duplicate validation",
|
|
downstream_classes=[source],
|
|
)
|
|
backlog = BackLog(completion={})
|
|
|
|
assert backlog.queue_testing(pending) is pending
|
|
assert backlog.queue_testing(duplicate) is None
|
|
assert backlog.testing == (pending,)
|
|
|
|
|
|
def test_testing_enqueue_keeps_uncovered_pair_beside_pending_pair() -> None:
|
|
"""Pending work for one source does not suppress another source/test pair."""
|
|
|
|
first_source = _meta("src/first.py")
|
|
second_source = _meta("src/second.py")
|
|
test = _meta("tests/test_subject.py")
|
|
pending = TestingBacklog(
|
|
focused_spotlight=test,
|
|
summary="validate first source",
|
|
downstream_classes=[first_source],
|
|
)
|
|
uncovered = TestingBacklog(
|
|
focused_spotlight=test,
|
|
summary="validate second source",
|
|
downstream_classes=[second_source],
|
|
)
|
|
backlog = BackLog(completion={})
|
|
|
|
assert backlog.queue_testing(pending) is pending
|
|
assert backlog.queue_testing(uncovered) is uncovered
|
|
assert backlog.testing == (pending, uncovered)
|
|
|
|
|
|
def test_remediation_reduces_candidates_to_one_blame_with_its_upstream(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""The root culprit alone receives its own direct dependency topology."""
|
|
|
|
client = _Client(precedence_winner="src/two.py")
|
|
manager = _manager(
|
|
memory_root,
|
|
{
|
|
"base_one.py": "pass\n",
|
|
"base_two.py": "pass\n",
|
|
"one.py": "def culprit_one():\n return False\n",
|
|
"two.py": "def culprit_two():\n return True\n",
|
|
},
|
|
{"test_subject.py": "from one import subject\nfrom two import subject as other\n"},
|
|
client=client,
|
|
)
|
|
base_one = _source(manager, "src/base_one.py")
|
|
base_two = _source(manager, "src/base_two.py")
|
|
one = _source(manager, "src/one.py")
|
|
two = _source(manager, "src/two.py")
|
|
test = _test(manager, "tests/test_subject.py")
|
|
dependency = _DirectMapStrategy({
|
|
test.location: [one, two],
|
|
one.location: [base_one],
|
|
two.location: [base_two],
|
|
})
|
|
LanguageStrategySet.set_instance(
|
|
lambda file_manager: _strategy_set(file_manager, dependency=dependency)
|
|
)
|
|
failed = FailingTest(
|
|
test=test,
|
|
error_message="culprit_two produced boom",
|
|
stack_trace=["src/two.py: culprit_two()"],
|
|
raw_output="raw",
|
|
)
|
|
ticket = TestingBacklog(focused_spotlight=test, summary="repair", downstream_classes=[one], original_cot="goal")
|
|
story = TargetedFailureStory(testing_item=ticket, raw_output="raw", work_needed="repair", candidate_sources=[one, two], failing=failed, failure_phase="after_beta", downstream_classes=[one], original_cot="goal")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec)
|
|
prior_beta = graph.start(Phase.BETA, TestingBacklog(test, "prior test work"))
|
|
alpha = graph.track_provenance(
|
|
prior_beta,
|
|
Phase.ALPHA,
|
|
InspectionBacklog(one, "repair the previously blamed source"),
|
|
)
|
|
graph.track_provenance(alpha, Phase.BETA, ticket)
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.BETA, ticket):
|
|
created = BackLog(completion={}).report_remediation(
|
|
ai_client=client,
|
|
file_manager=manager,
|
|
story=story,
|
|
candidates=[two],
|
|
summary="repair",
|
|
queue=False,
|
|
)
|
|
|
|
[item] = created
|
|
assert isinstance(item, FailedQAItem)
|
|
assert _focused(item).location == two.location
|
|
assert item.failed_test == failed
|
|
assert created[0].downstream_classes == [two]
|
|
assert created[0].upstream_classes == [base_two]
|
|
|
|
|
|
def test_remediation_retains_routed_candidate_when_dependency_candidates_are_empty(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A routed source obligation survives empty dependency discovery."""
|
|
|
|
client = _Client(precedence_winner="src/foo.py")
|
|
manager = _manager(
|
|
memory_root,
|
|
{"foo.py": "def culprit_foo():\n return True\n"},
|
|
{"test_foo.py": "from foo import subject\n"},
|
|
client=client,
|
|
)
|
|
source = _source(manager, "src/foo.py")
|
|
test = _test(manager, "tests/test_foo.py")
|
|
failed = FailingTest(
|
|
test=test,
|
|
error_message="culprit_foo produced boom",
|
|
stack_trace=["src/foo.py: culprit_foo()"],
|
|
raw_output="raw",
|
|
)
|
|
ticket = TestingBacklog(focused_spotlight=test, summary="repair", downstream_classes=[source])
|
|
story = TargetedFailureStory(testing_item=ticket, raw_output="raw", work_needed="repair", candidate_sources=[source], failing=failed, failure_phase="after_beta", downstream_classes=[source])
|
|
LanguageStrategySet.set_instance(
|
|
lambda file_manager: _strategy_set(
|
|
file_manager,
|
|
dependency=_DirectMapStrategy({test.location: []}),
|
|
)
|
|
)
|
|
|
|
created = BackLog(completion={}).report_remediation(
|
|
ai_client=client,
|
|
file_manager=manager,
|
|
story=story,
|
|
candidates=[source],
|
|
summary="repair",
|
|
queue=False,
|
|
)
|
|
|
|
[item] = created
|
|
assert isinstance(item, FailedQAItem)
|
|
assert _focused(item).location == source.location
|
|
|
|
|
|
@pytest.mark.parametrize("include_context", [False, True])
|
|
def test_remediation_keeps_context_out_of_trace_candidate_pool(
|
|
memory_root: Path,
|
|
include_context: bool,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Neither context admission nor dependency/topology membership authorizes untraced blame."""
|
|
|
|
client = _Client(precedence_winner="src/owner.py")
|
|
manager = _manager(
|
|
memory_root,
|
|
{"owner.py": "def culprit_owner():\n return True\n", "context.py": "def helper():\n return True\n"},
|
|
{"test_owner.py": "from owner import culprit_owner\nfrom context import helper\n"},
|
|
client=client,
|
|
)
|
|
owner = _source(manager, "src/owner.py")
|
|
context = _source(manager, "src/context.py")
|
|
test = _test(manager, "tests/test_owner.py")
|
|
dependency = _DirectMapStrategy({test.location: [context, owner], owner.location: [context]})
|
|
LanguageStrategySet.set_instance(
|
|
lambda file_manager: _strategy_set(file_manager, dependency=dependency)
|
|
)
|
|
complete = client.complete
|
|
responses = Mock(side_effect=complete)
|
|
monkeypatch.setattr(client, "complete", responses)
|
|
failure = FailingTest(
|
|
test=test,
|
|
error_message="culprit_owner failed in src/context.py" if include_context else "culprit_owner failed",
|
|
stack_trace=["src/owner.py:1", "def culprit_owner():"],
|
|
)
|
|
curated = curate_failure_context_sources(file_manager=manager, failure=failure)
|
|
assert (context in curated) is include_context
|
|
story = TargetedFailureStory(
|
|
testing_item=None, raw_output="", work_needed="repair", candidate_sources=[context],
|
|
failing=failure, failure_phase="after_alpha", upstream_classes=[context],
|
|
downstream_classes=[context], failure_context_sources=curated,
|
|
focused_test=test, change_type=ChangeType.Update,
|
|
)
|
|
|
|
created = BackLog(completion={}).report_remediation(
|
|
ai_client=client, file_manager=manager, story=story, candidates=[owner], summary="repair", queue=False,
|
|
)
|
|
|
|
[item] = created
|
|
assert isinstance(item, FailedQAItem)
|
|
assert _focused(item).location == owner.location
|
|
assert created[0].upstream_classes == [context]
|
|
assert created[0].failure_context_sources == curated
|
|
comparisons = list(filter(
|
|
lambda call: "== BEGIN FILE BLAME CANDIDATE PRECEDENCE ==" in call.args[0],
|
|
responses.call_args_list,
|
|
))
|
|
assert len(comparisons) == 1
|
|
assert "Inspect the sole preselected production candidate" in comparisons[0].args[0]
|
|
assert "FILE: src/context.py" not in comparisons[0].args[0]
|
|
|
|
|
|
def test_remediation_does_not_fall_back_to_untraced_inventory(
|
|
memory_root: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A source-owned failure without managed trace paths cannot manufacture a culprit."""
|
|
|
|
client = _Client(precedence_winner="src/owner.py")
|
|
manager = _manager(
|
|
memory_root, {"owner.py": "def culprit_owner():\n return True\n"},
|
|
{"test_owner.py": "pass\n"}, client=client,
|
|
)
|
|
test = _test(manager, "tests/test_owner.py")
|
|
failure = FailingTest(test=test, error_message="external failure", stack_trace=["external.py:3"])
|
|
story = TargetedFailureStory(
|
|
testing_item=None, raw_output="", work_needed="repair", candidate_sources=[],
|
|
failing=failure, failure_phase="after_alpha",
|
|
focused_test=test, change_type=ChangeType.Update,
|
|
)
|
|
backlog = BackLog(completion={})
|
|
completion = Mock(side_effect=AssertionError("No eligible source may reach model completion."))
|
|
monkeypatch.setattr(client, "complete", completion)
|
|
|
|
with pytest.raises(ValueError, match="trace-supported managed production candidate"):
|
|
backlog.report_remediation(ai_client=client, file_manager=manager, story=story, candidates=[], summary="repair")
|
|
|
|
assert backlog.remediation == ()
|
|
completion.assert_not_called()
|
|
|
|
|
|
def test_source_upsert_replaces_forward_and_reverse_edges(memory_root: Path) -> None:
|
|
"""Changing an import repairs forward and reverse graph evidence incrementally."""
|
|
|
|
manager = _manager(memory_root, {"a.py": "from b import b\n", "b.py": "pass\n", "c.py": "pass\n"}, {})
|
|
manager.repo_exec.write_text(manager.source_path("src/a.py"), "from c import c\n")
|
|
manager.upsert_src(_meta("src/a.py"), "goal")
|
|
graph = _maintainer(manager).graph
|
|
|
|
assert graph.source_dependencies["src/a.py"] == {"src/c.py"}
|
|
assert "src/a.py" not in graph.source_dependents.get("src/b.py", set())
|
|
assert graph.source_dependents["src/c.py"] == {"src/a.py"}
|
|
|
|
|
|
def test_test_upsert_replaces_source_test_mapping(memory_root: Path) -> None:
|
|
"""Changing a test import repairs both directions of source/test linkage."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "pass\n", "bar.py": "pass\n"}, {"test_subject.py": "from foo import subject\n"})
|
|
manager.repo_exec.write_text(manager.test_path("tests/test_subject.py"), "from bar import subject\n")
|
|
manager.upsert_test(_meta("tests/test_subject.py"), "goal")
|
|
graph = _maintainer(manager).graph
|
|
|
|
assert graph.sources_for_test["tests/test_subject.py"] == {"src/bar.py"}
|
|
assert graph.tests_for_source["src/bar.py"] == {"tests/test_subject.py"}
|
|
|
|
|
|
def test_new_source_refresh_is_visible_to_direct_test_lookup(memory_root: Path) -> None:
|
|
"""New source metadata enters the live graph before direct test lookup."""
|
|
|
|
manager = _manager(memory_root, {"old.py": "pass\n"}, {"test_new.py": "from new import subject\n"})
|
|
manager.repo_exec.write_text(manager.src_root / "new.py", "def subject():\n return 1\n")
|
|
new_source = manager.refresh_src("src/new.py", "new source")
|
|
|
|
assert [test.location for test in PythonAstDependencyStrategy(_maintainer(manager)).direct_tests_for([new_source])] == ["tests/test_new.py"]
|
|
|
|
|
|
def test_frozen_graph_captures_once_and_rejects_mutation(memory_root: Path) -> None:
|
|
"""The first frozen graph remains the durable topology reference."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "pass\n", "old.py": "from foo import subject\n", "new.py": "from foo import subject\n"}, {})
|
|
source = _source(manager, "src/foo.py")
|
|
old = _source(manager, "src/old.py")
|
|
new = _source(manager, "src/new.py")
|
|
graph = _maintainer(manager).graph
|
|
graph.source_dependents[source.location] = {old.location}
|
|
manager.capture_frozen_once()
|
|
graph.source_dependents[source.location] = {new.location}
|
|
frozen = manager.elsa
|
|
assert frozen is not None
|
|
|
|
assert frozen.source_dependents[source.location] == {old.location}
|
|
with pytest.raises(RuntimeError):
|
|
frozen.source_by_location = {}
|
|
|
|
|
|
def test_pristine_sources_share_startup_text_and_survive_checkpoint(memory_root: Path) -> None:
|
|
"""Child contexts and restored checkpoints retain startup edits despite subsequent source changes."""
|
|
|
|
import pickle
|
|
|
|
manager = _manager(memory_root, {"foo.py": "def original():\n return 1\n"}, {"test_foo.py": "pass\n"})
|
|
manager.capture_frozen_once()
|
|
child = FileManager.from_manager(manager)
|
|
manager.repo_exec.write_text(memory_root / "src/foo.py", "def replacement():\n return 2\n")
|
|
manager.capture_frozen_once()
|
|
restored = pickle.loads(pickle.dumps(manager))
|
|
|
|
assert manager.pristine_sources["src/foo.py"] == "def original():\n return 1\n"
|
|
assert child.pristine_sources == manager.pristine_sources == restored.pristine_sources
|
|
assert manager.pristine_sources["tests/test_foo.py"] == "pass\n"
|
|
with pytest.raises(TypeError):
|
|
manager.pristine_sources["src/foo.py"] = "changed" # type: ignore[index]
|
|
|
|
|
|
def test_stabilization_unions_current_and_frozen_downstream(memory_root: Path) -> None:
|
|
"""Topology expansion includes live and frozen downstream edges exactly once each."""
|
|
|
|
manager = _manager(memory_root, {"foo.py": "pass\n", "old.py": "pass\n", "new.py": "pass\n"}, {})
|
|
source = _source(manager, "src/foo.py")
|
|
old = _source(manager, "src/old.py")
|
|
new = _source(manager, "src/new.py")
|
|
manager._frozen = CodeDependencyGraph(source_by_location={source.location: source, old.location: old, new.location: new}, source_dependents={source.location: {old.location}})
|
|
live = CodeDependencyGraph(source_by_location={source.location: source, old.location: old, new.location: new}, source_dependents={source.location: {new.location}})
|
|
maintainer = PythonAstDependencyGraphMaintainer(repo_exec=manager.repo_exec, _graph=live)
|
|
manager.dependency_graph_maintainer = maintainer
|
|
manager.dependency_strategy = PythonAstDependencyStrategy(maintainer)
|
|
loop, _ = _loop(manager, source)
|
|
|
|
expanded = service_topology(loop, [source], opening_hashes={source.location: "opening"})
|
|
|
|
assert [item.location for item in expanded] == [new.location, old.location]
|
|
|
|
|
|
def test_graph_query_uses_built_evidence_without_reading_repository(memory_root: Path) -> None:
|
|
"""Once built, graph queries never need another repository content read."""
|
|
|
|
executor = _ReadBoundaryRepoExec(memory_root)
|
|
manager = _manager(memory_root, {"foo.py": "pass\n", "bar.py": "from foo import subject\n"}, {"test_foo.py": "from foo import subject\n"}, repo_exec=executor)
|
|
source = _source(manager, "src/foo.py")
|
|
bar = _source(manager, "src/bar.py")
|
|
test = _test(manager, "tests/test_foo.py")
|
|
executor.reads_allowed = False
|
|
strategy = PythonAstDependencyStrategy(_maintainer(manager))
|
|
|
|
assert strategy.direct_tests_for([source]) == [test]
|
|
assert strategy.downstream_sources_for([source]) == [bar]
|
|
|
|
|
|
def test_source_package_markers_are_created_through_repo_exec(memory_root: Path) -> None:
|
|
"""New nested source metadata creates required package markers in memory."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
source = _meta("src/pkg/new_module.py")
|
|
executor.write_text(memory_root / source.location, "pass\n")
|
|
|
|
PythonAstDependencyGraphMaintainer(repo_exec=executor).upsert(source, [source], [], memory_root, memory_root / "src", memory_root / "tests")
|
|
|
|
assert executor.is_file(memory_root / "src" / "pkg" / "__init__.py")
|
|
|
|
|
|
def test_test_package_markers_are_created_through_repo_exec(memory_root: Path) -> None:
|
|
"""New nested test metadata creates required package markers in memory."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
test = _meta("tests/pkg/test_new_module.py")
|
|
executor.write_text(memory_root / test.location, "pass\n")
|
|
|
|
PythonAstDependencyGraphMaintainer(repo_exec=executor).upsert(test, [], [test], memory_root, memory_root / "src", memory_root / "tests")
|
|
|
|
assert executor.is_file(memory_root / "tests" / "pkg" / "__init__.py")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("module_name", "import_line", "expected_dependency"),
|
|
[
|
|
("case_01", "", set()),
|
|
("case_02", "from target import subject", {"src/target.py"}),
|
|
("case_03", "import target", {"src/target.py"}),
|
|
("case_04", "from target import subject as alias", {"src/target.py"}),
|
|
("case_05", "from target import another", {"src/target.py"}),
|
|
("case_06", "from target import subject\nfrom sibling import helper", {"src/target.py", "src/sibling.py"}),
|
|
("case_07", "from sibling import helper", {"src/sibling.py"}),
|
|
("case_08", "from target import *", {"src/target.py"}),
|
|
("case_09", "import target as renamed", {"src/target.py"}),
|
|
("case_10", "from target import subject\n", {"src/target.py"}),
|
|
("case_11", "from target import subject\n# comment", {"src/target.py"}),
|
|
("case_12", "from target import subject\n\n", {"src/target.py"}),
|
|
("case_13", "from sibling import helper\nfrom target import subject", {"src/target.py", "src/sibling.py"}),
|
|
("case_14", "import sibling", {"src/sibling.py"}),
|
|
("case_15", "from target import subject as local", {"src/target.py"}),
|
|
("case_16", "from sibling import helper as local", {"src/sibling.py"}),
|
|
("case_17", "from target import subject\nfrom target import other", {"src/target.py"}),
|
|
("case_18", "from sibling import helper\nfrom sibling import other", {"src/sibling.py"}),
|
|
("case_19", "from target import subject\nvalue = subject", {"src/target.py"}),
|
|
("case_20", "from sibling import helper\nvalue = helper", {"src/sibling.py"}),
|
|
("case_21", "from target import subject\nfrom sibling import helper\nvalue = subject", {"src/target.py", "src/sibling.py"}),
|
|
("case_22", "import target\nimport sibling", {"src/target.py", "src/sibling.py"}),
|
|
("case_23", "from target import subject\nif True:\n value = subject", {"src/target.py"}),
|
|
("case_24", "from sibling import helper\nif True:\n value = helper", {"src/sibling.py"}),
|
|
("case_25", "from target import subject\nvalue = 1", {"src/target.py"}),
|
|
("case_26", "from sibling import helper\nvalue = 1", {"src/sibling.py"}),
|
|
],
|
|
)
|
|
def test_dependency_graph_preserves_explicit_static_import_edges(
|
|
memory_root: Path,
|
|
module_name: str,
|
|
import_line: str,
|
|
expected_dependency: set[str],
|
|
) -> None:
|
|
"""Each import shape produces exactly its explicit static graph dependencies."""
|
|
|
|
manager = _manager(
|
|
memory_root,
|
|
{
|
|
"target.py": "def subject():\n return 1\n",
|
|
"sibling.py": "def helper():\n return 1\n",
|
|
f"{module_name}.py": f"{import_line}\n\ndef caller():\n return 1\n",
|
|
},
|
|
{},
|
|
)
|
|
|
|
assert _maintainer(manager).graph.source_dependencies.get(f"src/{module_name}.py", set()) == expected_dependency
|
|
|
|
|
|
@pytest.mark.parametrize("trace", (
|
|
("tests/test_direct_code_editor.py:709: AssertionError",),
|
|
("external/framework:3",),
|
|
))
|
|
def test_assertion_only_failure_stays_in_beta_without_source_blame(
|
|
memory_root: Path, monkeypatch: pytest.MonkeyPatch, trace: tuple[str, ...],
|
|
) -> None:
|
|
"""The recorded assertion-only crash cannot enter FileBlame through a topology source."""
|
|
|
|
manager = _manager(
|
|
memory_root, {"editor.py": "def edit():\n return []\n"},
|
|
{"test_direct_code_editor.py": "def test_edit():\n assert False\n"},
|
|
)
|
|
test = _test(manager, "tests/test_direct_code_editor.py")
|
|
source = _source(manager, "src/editor.py")
|
|
failure = FailingTest(
|
|
test=test, stack_trace=list(trace), error_message="assert [] == [MemoryPath('/testbed/src/mod.py')]",
|
|
)
|
|
context = curate_failure_context_sources(file_manager=manager, failure=failure)
|
|
assert context == ([test] if trace[0].startswith("tests/") else [])
|
|
story = TargetedFailureStory(
|
|
testing_item=None, raw_output=failure.error_message, work_needed="Repair the failure",
|
|
candidate_sources=[source, *context], failing=failure, failure_phase="after_alpha",
|
|
focused_test=test, change_type=ChangeType.Update, failure_context_sources=context,
|
|
)
|
|
completion = Mock(side_effect=AssertionError("Mechanical routing must not call an LLM"))
|
|
monkeypatch.setattr(manager.client, "complete", completion)
|
|
backlog = BackLog(completion={})
|
|
|
|
routing = backlog.judgement_service.route_targeted_failure(file_manager=manager, story=story)
|
|
|
|
assert routing.outcome == "test"
|
|
assert routing.additional["failed_test"] == test.location
|
|
assert routing.additional["candidate_sources"] == list(map(lambda candidate: candidate.location, context))
|
|
assert routing.prompt == ""
|
|
assert backlog.remediation == ()
|
|
completion.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("via_qa", (False, True))
|
|
def test_confirmed_file_blame_test_owner_preserves_alpha_and_beta_obligation(
|
|
memory_root: Path, monkeypatch: pytest.MonkeyPatch, via_qa: bool,
|
|
) -> None:
|
|
"""Both postflight entry paths retain the original failed test after confirmed Beta attribution."""
|
|
|
|
manager = _manager(memory_root, {"owner.py": "def owner():\n return 2\n"},
|
|
{"test_owner.py": "def test_owner():\n assert owner() == 1\n"})
|
|
source, test = _source(manager, "src/owner.py"), _test(manager, "tests/test_owner.py")
|
|
failure = FailingTest(test=test, test_name="test_owner", error_message="assert 2 == 1",
|
|
stack_trace=["src/owner.py:2", "tests/test_owner.py:2"], raw_output="complete failure")
|
|
original = TestingBacklog(focused_spotlight=test, summary="Verify owner.",
|
|
downstream_classes=[source], original_cot="owner must return 2.",
|
|
message_id="original-beta-attempt")
|
|
story = TargetedFailureStory(
|
|
testing_item=original if via_qa else None, raw_output=failure.raw_output,
|
|
work_needed="Verify owner.", candidate_sources=[source], failing=failure,
|
|
failure_phase="after_beta" if via_qa else "after_alpha", focused_test=test,
|
|
original_cot=original.original_cot, change_type=ChangeType.Update,
|
|
)
|
|
routing = ReasonableVibeCheck(outcome="source", reason="Earlier production attribution.")
|
|
blame = Mock(outcome="test", reason="Highest review confirms obsolete test expectation.")
|
|
constructor = Mock(return_value=blame)
|
|
monkeypatch.setattr("band_aider.context.working_backlog.FileBlameDecision", constructor)
|
|
backlog = BackLog(completion={})
|
|
|
|
# Beta retains its existing attempt; Alpha creates ordinary new Beta work.
|
|
if via_qa:
|
|
qa = QABacklog(backlog, manager.client, manager, manager.parser)
|
|
[item] = qa.report_remediation(testing_item=original, summary="Observed failure.", failed_test=failure,
|
|
candidates=[source], queue=False, routing=routing)
|
|
else:
|
|
[item] = backlog.report_remediation(ai_client=manager.client, file_manager=manager, story=story,
|
|
candidates=[source], summary="Observed failure.", queue=False, routing=routing)
|
|
|
|
assert isinstance(item, TestingBacklog)
|
|
assert not isinstance(item, FailedQAItem)
|
|
assert item.focused_spotlight is test
|
|
assert item.message_id == (original.message_id if via_qa else "")
|
|
assert item.failure_error_message == failure.error_message
|
|
assert item.failure_test_name == failure.test_name
|
|
assert item.failure_stack_trace == failure.stack_trace
|
|
assert item.failure_raw_output == failure.raw_output
|
|
assert blame.reason in item.previous_beta_guidance
|
|
assert constructor.call_args.kwargs["routing"] is routing
|
|
assert backlog.remediation == ()
|
|
assert backlog.testing == ()
|
|
|
|
|
|
@pytest.mark.parametrize("via_qa", (False, True))
|
|
def test_chained_failure_remediation_retains_executed_test_and_method(
|
|
memory_root: Path, via_qa: bool,
|
|
) -> None:
|
|
"""Alpha postflight and QA retain the outer test despite a helper's earlier report."""
|
|
client = _Client(precedence_winner="src/owner.py")
|
|
manager = _manager(
|
|
memory_root,
|
|
{"owner.py": "def culprit_owner():\n return True\n"},
|
|
{"test_outer.py": "def test_outer():\n assert False\n",
|
|
"toolkit.py": "def read():\n raise KeyError('snapshot')\n"}, client=client,
|
|
)
|
|
owner = _source(manager, "src/owner.py")
|
|
outer = _test(manager, "tests/test_outer.py")
|
|
support = _test(manager, "tests/toolkit.py")
|
|
trace = [
|
|
"_____ test_outer _____",
|
|
"src/owner.py:1: in culprit_owner",
|
|
"tests/toolkit.py:2: in read",
|
|
"KeyError: snapshot",
|
|
"During handling of the above exception, another exception occurred:",
|
|
"tests/test_outer.py:2: in test_outer",
|
|
"src/owner.py:2: in culprit_owner",
|
|
"RuntimeError: rollback",
|
|
]
|
|
failure = FailingTest(test=outer, test_name="test_outer", error_message="culprit_owner failed",
|
|
stack_trace=trace, raw_output="\n".join(trace))
|
|
story = TargetedFailureStory(
|
|
testing_item=None, raw_output=failure.raw_output, work_needed="repair",
|
|
candidate_sources=[support, owner, outer], failing=failure, failure_phase="after_alpha",
|
|
focused_test=outer, change_type=ChangeType.Update,
|
|
)
|
|
backlog = BackLog(completion={})
|
|
# Both entry paths must receive the same normalized origin from the shared owner.
|
|
if via_qa:
|
|
qa = QABacklog(backlog, client, manager, manager.parser)
|
|
[item] = qa.report_remediation(
|
|
testing_item=TestingBacklog(summary="repair", focused_spotlight=outer, change_type=ChangeType.Update),
|
|
failed_test=failure, candidates=story.candidate_sources, summary="repair", queue=False,
|
|
)
|
|
else:
|
|
[item] = backlog.report_remediation(
|
|
ai_client=client, file_manager=manager, story=story,
|
|
candidates=story.candidate_sources, summary="repair", queue=False,
|
|
)
|
|
|
|
assert isinstance(item, FailedQAItem)
|
|
assert item.focused_spotlight == owner
|
|
assert item.failed_test is not None
|
|
assert item.failed_test.test == outer
|
|
assert item.failed_test.test_name == "test_outer"
|
|
assert item.failure_test_name == "test_outer"
|
|
assert item.failed_test.stack_trace == trace
|
|
assert item.failed_test.raw_output == failure.raw_output
|
|
assert set(map(lambda source: source.location, item.failure_context_sources)) == {
|
|
outer.location, owner.location, support.location,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("call_only", (False, True))
|
|
@pytest.mark.parametrize("via_qa", (False, True))
|
|
def test_remediation_recomputes_admission_from_original_pool_and_preserves_origin(
|
|
memory_root: Path, monkeypatch: pytest.MonkeyPatch, via_qa: bool, call_only: bool,
|
|
) -> None:
|
|
"""Alpha reuses discovery logic, not a prior decision's cached shortlist."""
|
|
|
|
client = _Client(precedence_winner="src/owner.py")
|
|
manager = _manager(
|
|
memory_root,
|
|
{"owner.py": "def culprit_owner():\n return True\n", "other.py": "pass\n"},
|
|
{"test_outer.py": "pass\n", "toolkit.py": "pass\n"}, client=client,
|
|
)
|
|
owner = _source(manager, "src/owner.py")
|
|
outer = _test(manager, "tests/test_outer.py")
|
|
support = _test(manager, "tests/toolkit.py")
|
|
failure = FailingTest(
|
|
test=outer, test_name="test_outer", error_message="culprit_owner failed",
|
|
stack_trace=["index = {'tests/toolkit.py': 0}", f"{outer.location}:2: in test_outer", support.location,
|
|
"src/owner.py: culprit_owner()", "src/other.py:8"],
|
|
raw_output="complete raw failure",
|
|
)
|
|
from band_aider.judgement.source_discovery_strategy import CallImport, CallTraceNode
|
|
calls = ()
|
|
# The same handoff must retain a production owner absent from runtime frames.
|
|
if call_only:
|
|
failure.stack_trace[:] = ["index = {'tests/toolkit.py': 0}", f"{outer.location}:2: in test_outer", support.location]
|
|
calls = (CallTraceNode(location=owner.location, name="culprit_owner", line=1,
|
|
caller_location=outer.location, call_line=3, call_column=4,
|
|
import_reference=CallImport(location=outer.location, line=1, text="from owner import culprit_owner", imported_name="owner.culprit_owner")),)
|
|
story = TargetedFailureStory(
|
|
testing_item=None, raw_output=failure.raw_output, work_needed="repair",
|
|
candidate_sources=[outer, support, owner], failing=failure, failure_phase="after_alpha",
|
|
focused_test=support, change_type=ChangeType.Update,
|
|
)
|
|
snapshot = (outer, support, owner)
|
|
discovery = Mock(wraps=failure_source_candidate_selection_strategy_factory.candidates_for_failure)
|
|
monkeypatch.setattr(failure_source_candidate_selection_strategy_factory, "candidates_for_failure", discovery)
|
|
monkeypatch.setattr(
|
|
failure_source_candidate_selection_strategy_factory, "calls_for_failure",
|
|
Mock(return_value=calls),
|
|
)
|
|
|
|
backlog = BackLog(completion={})
|
|
# Both originating workflows pass the original inputs into the same discovery owner.
|
|
if via_qa:
|
|
qa = QABacklog(backlog, client, manager, manager.parser)
|
|
[item] = qa.report_remediation(
|
|
testing_item=TestingBacklog(
|
|
summary="repair", focused_spotlight=support, change_type=ChangeType.Update,
|
|
),
|
|
failed_test=failure, candidates=snapshot, summary="repair", queue=False,
|
|
)
|
|
else:
|
|
[item] = backlog.report_remediation(
|
|
ai_client=client, file_manager=manager, story=story, candidates=snapshot, summary="repair", queue=False,
|
|
)
|
|
|
|
assert isinstance(item, FailedQAItem)
|
|
assert item.focused_spotlight == owner
|
|
assert item.failure_context_sources == list(snapshot)
|
|
assert item.failed_test is not None
|
|
assert item.failed_test.test == outer
|
|
assert item.failed_test.stack_trace == failure.stack_trace
|
|
assert item.failed_test.raw_output == failure.raw_output
|
|
assert failure.test == outer
|
|
assert item.failed_test.test_name == "test_outer"
|
|
assert discovery.called
|