band-aider-pub/tests/band_aider/test_bootstrap_mapping.py
2026-10-17 11:57:37 -07:00

3221 lines
117 KiB
Python

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Callable
from band_aider.judgement.source_discovery_strategy import CallTraceNode
import json
import logging
from dataclasses import dataclass
from io import StringIO
from pathlib import Path
from typing import Any, TextIO
import pytest
from band_aider.context.bootstrap_context import BootstrapContext
from band_aider.context.class_meta import ClassMeta, ContentType
from band_aider.context.context_factory import (
BootstrapContextReassembly,
LoadedCheckpoint,
context_factory,
)
from band_aider.context.global_context import GlobalContext, Intent, StoryCursor
from band_aider.context.cot_task_generator import CotTaskRecord
from band_aider.context.memento import Memento
from band_aider.judgement.code_dependency_graph import (
CodeDependencyGraph,
DependencyDelta,
DependencySelector,
)
from band_aider.judgement.code_dependency_graph_maintainer import (
CodeDependencyGraphMaintainer,
GraphSignature,
)
from band_aider.judgement.code_editor import CodeEditor, CodeEditorResult, EditorRecord
from band_aider.judgement.code_parser import CodeParser
from band_aider.judgement.dependency_discovery_strategy import (
DependencyDiscoveryStrategy,
DependencyTraversalCache,
)
from band_aider.judgement.cross_cutting_concern_check import (
CrossCuttingConcernCheck,
CrossCuttingConcernCheckFactory,
)
from band_aider.judgement.code_edit_or_leave_it import (
CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN,
CodeEditLeaveSummary,
)
from band_aider.judgement.edit_request import AiderEditRequest
from band_aider.judgement.file_manager import FileManager
from band_aider.judgement.intent_check import IntentCheckFactory
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.repository_idiom_check import (
RepositoryIdiomCheck,
RepositoryIdiomCheckFactory,
)
from band_aider.judgement.tighten_design_work_check import (
TightenDesignWorkCheck,
TightenDesignWorkCheckFactory,
)
from band_aider.judgement.vibe_check import ReasonableVibeCheck
from band_aider.judgement.source_discovery_strategy import (
SearchableBeatValue,
SourceDiscoveryStrategy,
)
from band_aider.loop.bootstrap_loop import BootstrapLoop
from band_aider.loop.alpha_loop import AlphaLoop
from band_aider.loop.supreme_loop import SupremeLoop
from band_aider.strategies import LanguageStrategySet
from band_aider.strategies.oop_signature_extract import (
OOpSignature,
OOpSignatureExtract,
OOpSignatureRender,
)
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
from band_aider.judgement.repo_exec import RepoExec
from tests.support.memory_repo_exec import MemoryRepoExec
pytestmark = pytest.mark.usefixtures("without_embedding_model")
class _Client(OpenAICompatibleClient):
"""_Client returns queued JSON responses and records prompts."""
def __init__(self, responses: list[object] | None = None) -> None:
"""Store queued completion responses."""
super().__init__(
model="test",
api_base="http://model.invalid",
api_key="test-key",
max_attempts=1,
)
self.responses = list(responses or [])
self.prompts: list[str] = []
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
"""Return the next queued response as strict JSON text."""
del logger
self.prompts.append(prompt)
if "Bootstrap source preflight input:" in prompt:
return json.dumps([])
if "ACCEPTANCE ROLE" in prompt and "ORIGINAL AIDER EDIT REQUEST" in prompt:
return json.dumps(
{
"outcome": "pass",
"reason": ["# wired"],
"instruction": "",
}
)
if not self.responses:
return json.dumps({"cdc": "metadata refreshed"})
response = self.responses.pop(0)
if isinstance(response, str):
return response
return json.dumps(response)
class _NoCards(OOpSignatureExtract):
"""Keep unrelated bootstrap fixtures explicit about empty card evidence."""
def extract(self, module: str, name: str):
del module, name
return None
def extract_references(self, sources_by_location):
del sources_by_location
return ()
def extract_source(self, location, source):
del location, source
return ()
class _NoCardRender(OOpSignatureRender):
"""Make bootstrap fixtures explicit about absent declaration rendering."""
def render(self, signature: OOpSignature, output) -> None:
"""Fail if an empty-card fixture unexpectedly receives a declaration."""
del signature, output
raise AssertionError("No reference cards should be rendered by this fixture.")
class _Parser(CodeParser):
"""_Parser returns prompt-sensitive source metadata for targeted refresh tests."""
def __init__(self, repo_root: Path) -> None:
"""Store the repository root and parse call history."""
self.repo_root = repo_root
self.calls: list[tuple[str, str]] = []
def parse_file(
self,
file_name: str,
prompt: str,
content_type: ContentType = ContentType.Source,
) -> ClassMeta:
"""Return one metadata record using the supplied prompt as relevance evidence."""
path = Path(file_name)
self.calls.append((path.name, prompt))
location = (
path.relative_to(self.repo_root).as_posix()
if path.is_absolute()
else path.as_posix()
)
return ClassMeta(
location=location,
summary=f"summary from {prompt} for {location}",
relevance=[f"relevance from {prompt}"],
relevant_symbols={"symbol": f"symbol from {prompt}"},
content_type=content_type,
)
def parse_stream(
self,
file_name: str,
stream: TextIO,
prompt: str,
content_type: ContentType = ContentType.Source,
) -> ClassMeta:
"""Parse a stream by delegating through the file-name path."""
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 multiple files with the same prompt."""
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
class _Discovery(SourceDiscoveryStrategy):
"""_Discovery supplies deterministic source files and searchable beats."""
def visit_call_sites(self, target: CallTraceNode, already_seen: set[str] | None = None, predicate: Callable[[CallTraceNode], bool] | None = None, *, target_args: dict[str, CallTraceNode] | None = None, pristine: Mapping[str, str] | None = None) -> None:
"""Supply no imported calls in this stationary discovery double."""
del target, already_seen, predicate
def get_call_tree(self, location: str, name: str | None = None, *, line: int | None = None, depth: int = 1, already_seen: set[str] | None = None, predicate: Callable[[CallTraceNode], bool] | None = None, pristine: Mapping[str, str] | None = None) -> CallTraceNode:
"""Reject origins because this double contains no callable definitions."""
raise LookupError(location)
def __init__(self, src_root: Path, test_root: Path, repo_exec: RepoExec) -> None:
"""Store roots for source and test discovery."""
self.src_root = src_root
self.test_root = test_root
self.repo_exec = repo_exec
def get_sources(self, recursive: bool = True, type: Any | None = None) -> list[Path]:
"""Return all source files under the configured source root."""
del type
return self.repo_exec.list_files(self.src_root, ("*.py",), recursive=recursive)
def get_tests(self, recursive: bool = True, type: Any | None = None) -> list[Path]:
"""Return all test files under the configured test root."""
del type
return self.repo_exec.list_files(self.test_root, ("*.py",), recursive=recursive)
def searchable_beats(self, file_name: str, source: str) -> list[SearchableBeatValue]:
"""Return non-empty stripped source lines as searchable beats."""
return [line.strip() for line in source.splitlines() if line.strip()]
class _GraphMaintainer(PythonAstDependencyGraphMaintainer):
"""_GraphMaintainer records graph refresh calls without building edges."""
def __init__(self, repo_exec: MemoryRepoExec) -> None:
"""Initialize graph state and call recording."""
super().__init__(repo_exec=repo_exec)
self.refresh_calls: list[str] = []
self.build_calls = 0
def build(
self,
source_files: list[ClassMeta],
test_files: list[ClassMeta],
repo_root: Path,
src_root: Path,
test_root: Path,
) -> CodeDependencyGraph:
"""Record a full graph build."""
self.build_calls += 1
return self.graph
def upsert(
self,
class_meta: ClassMeta,
source_files: list[ClassMeta],
test_files: list[ClassMeta],
repo_root: Path,
src_root: Path,
test_root: Path,
) -> CodeDependencyGraph:
"""Record one ClassMeta upsert."""
self.refresh_calls.append(class_meta.location)
if class_meta.location in {source.location for source in source_files}:
self._graph._apply_dependency_delta(
DependencyDelta(
source_added={class_meta.location: class_meta}
if class_meta.location not in self.graph.source_by_location
else {},
source_changed={class_meta.location: class_meta}
if class_meta.location in self.graph.source_by_location
else {},
)
)
else:
self.graph._apply_dependency_delta(
DependencyDelta(
test_added={class_meta.location: class_meta}
if class_meta.location not in self.graph.test_by_location
else {},
test_changed={class_meta.location: class_meta}
if class_meta.location in self.graph.test_by_location
else {},
)
)
return self.graph
def delete(
self,
class_meta: ClassMeta,
source_files: list[ClassMeta],
test_files: list[ClassMeta],
repo_root: Path,
src_root: Path,
test_root: Path,
) -> CodeDependencyGraph:
"""Record one ClassMeta deletion."""
self.refresh_calls.append(class_meta.location)
if class_meta.location in self.graph.source_by_location:
self.graph._apply_dependency_delta(
DependencyDelta(source_removed=frozenset({class_meta.location}))
)
else:
self.graph._apply_dependency_delta(
DependencyDelta(test_removed=frozenset({class_meta.location}))
)
return self.graph
def freeze(self) -> None:
"""Leave the fake graph mutable."""
class _DependencyStrategy(PythonAstDependencyStrategy):
"""_DependencyStrategy returns no dependency expansion for bootstrap tests."""
def __init__(self, maintainer: _GraphMaintainer) -> None:
"""Bind the production traversal implementation to the empty test graph."""
super().__init__(maintainer=maintainer)
class _Editor(CodeEditor):
"""_Editor records edit requests and mutates requested source files."""
def __init__(self) -> None:
"""Initialize request and callback recording."""
self.requests: list[AiderEditRequest] = []
self.file_manager: FileManager | None = None
self.changed_override: list[Path] | None = None
@property
def inner(self) -> CodeEditor | None:
"""Return None because this fake editor is concrete."""
return None
def edit(
self,
request: AiderEditRequest,
logger: logging.Logger | None = None,
) -> CodeEditorResult:
"""Edit scoped files and return changed paths."""
del logger
if self.file_manager is None:
raise AssertionError("Bootstrap editor requires its FileManager boundary.")
self.requests.append(request)
changed: list[Path] = []
for path in request.editable_files:
repo_exec = self.file_manager.repo_exec
repo_exec.write_text(path, repo_exec.read_text(path) + "\n# wired\n")
changed.append(path)
return CodeEditorResult(self.changed_override or changed, "editor summary", [])
class _ScenarioConcernCheck(CrossCuttingConcernCheck):
"""Return one explicit concern decision without invoking a completion endpoint."""
def __init__(
self,
*,
found: bool,
visible_files: list[ClassMeta],
beats_by_location: dict[str, list[str]],
) -> None:
ReasonableVibeCheck.__init__(self)
self._found = found
self._visible_files = list(visible_files)
self._beats_by_location = dict(beats_by_location)
def __call__(self, *args: object, **kwargs: object) -> bool:
"""Materialize the configured concern as the check's observable result."""
del args, kwargs
if not self._found:
self.outcome = "none"
self.additional = {"concerns": [], "selected_concern_index": None}
return False
locations = [file.location for file in self._visible_files[:2]]
self.outcome = "found"
self.additional = {
"concerns": [
{
"name": "shared responsibility",
"summary": "two files share a concrete responsibility",
"exhibits": [
{
"location": location,
"beat": (
self._beats_by_location.get(location)
or [f"exhibit {location}"]
)[0],
"symbol": "",
"reason": "participates",
}
for location in locations
],
}
],
"selected_concern_index": 0,
}
return True
class _ConcernFactory(CrossCuttingConcernCheckFactory):
"""Create explicit concern outcomes for a configured number of calls."""
def __init__(self, found_calls: int = 1) -> None:
super().__init__(
client=_Client(),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
)
self.found_calls = found_calls
self.visible_calls: list[list[str]] = []
def create(
self,
*,
original_prompt: str,
visible_files: list[ClassMeta],
beats_by_location: dict[str, list[str]],
) -> CrossCuttingConcernCheck:
"""Create one configured Prompt 1 result."""
del original_prompt
self.visible_calls.append([file.location for file in visible_files])
return _ScenarioConcernCheck(
found=len(self.visible_calls) <= self.found_calls,
visible_files=visible_files,
beats_by_location=beats_by_location,
)
class _ScenarioIdiomCheck(RepositoryIdiomCheck):
"""Return one bounded idiom catalog entry for the supplied sample."""
def __init__(self, sample_files: list[ClassMeta]) -> None:
ReasonableVibeCheck.__init__(self)
self._sample_files = list(sample_files)
def __call__(self, *args: object, **kwargs: object) -> bool:
"""Materialize the catalog entry owned by the sample."""
del args, kwargs
self.outcome = "cataloged"
self.additional = {
"entries": [
{
"category": "organization",
"observation": "nearby files keep focused responsibilities",
"evidence": [
{
"location": self._sample_files[0].location,
"symbol_or_excerpt": "def existing",
}
],
"design_implication": "keep the new construct focused",
}
]
}
return True
class _IdiomFactory(RepositoryIdiomCheckFactory):
"""Create a bounded Prompt 2 catalog for each requested sample."""
def __init__(self) -> None:
super().__init__(
client=_Client(),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
)
self.sample_sizes: list[int] = []
def create(
self,
*,
original_prompt: str,
selected_concern: dict[str, object],
sample_files: list[ClassMeta],
sources_by_location: dict[str, str],
reference_cards: str,
) -> RepositoryIdiomCheck:
"""Create one catalog result tied to the supplied sample."""
del original_prompt, selected_concern, sources_by_location, reference_cards
self.sample_sizes.append(len(sample_files))
return _ScenarioIdiomCheck(sample_files)
def _exhibit_beat(selected_concern: dict[str, object], index: int) -> str:
"""Return one validated design-exhibit beat from explicit test input."""
exhibits = selected_concern.get("exhibits")
if not isinstance(exhibits, list) or index >= len(exhibits):
raise AssertionError("Bootstrap design fixture requires the requested exhibit.")
exhibit = exhibits[index]
if not isinstance(exhibit, dict):
raise AssertionError("Bootstrap design fixture requires dictionary exhibits.")
beat = exhibit.get("beat")
if not isinstance(beat, str):
raise AssertionError("Bootstrap design fixture requires a string exhibit beat.")
return beat
class _DesignFactory(TightenDesignWorkCheckFactory):
"""_DesignFactory returns iteration-specific reusable design plans."""
def __init__(self, designed_calls: int = 1) -> None:
"""Store how many Prompt 3 calls should return designed."""
super().__init__(
client=_Client(),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
)
self.designed_calls = designed_calls
self.calls = 0
def create(
self,
*,
file_manager: FileManager,
original_prompt: str,
selected_concern: dict[str, object],
idiom_catalog: list[dict[str, object]],
visible_files: list[ClassMeta],
previous_construct_locations: list[str] | None = None,
) -> TightenDesignWorkCheck:
"""Create one fake Prompt 3 check."""
del file_manager, original_prompt, idiom_catalog, previous_construct_locations
self.calls += 1
call_index = self.calls
class _Check(TightenDesignWorkCheck):
"""_Check returns a concrete design while configured to do so."""
def __init__(self, designed: bool) -> None:
"""Store whether this design should pass."""
ReasonableVibeCheck.__init__(self)
self.designed = designed
def __call__(self, *args: object, **kwargs: object) -> bool:
"""Return the configured design decision."""
del args, kwargs
if not self.designed:
editable = [file.location for file in visible_files[:2]]
new_location = f"src/request_derived_construct_{call_index}.py"
self.additional = {
"new_files": [
{
"location": new_location,
"reason": "request-derived construct",
"constructs": [
{
"kind": "module",
"name": f"RequestDerivedConstruct{call_index}",
"responsibility": "implement request-derived bootstrap work",
"members": [],
}
],
}
],
"call_sites": [],
"construction_sites": [],
"editable_locations": editable,
"readonly_locations": [],
"expected_relationships": [],
"implementation_prompt": (
f"Create and populate {new_location}. "
"Preserve unrelated behavior. "
"No test edits during bootstrap. "
"No additional files outside the planned set."
),
}
return True
editable = [file.location for file in visible_files[:2]]
new_location = f"src/bootstrap_construct_{call_index}.py"
prompt = (
f"Create and populate {new_location}. "
f"Modify {editable[0]} change first site. "
f"Modify {editable[1]} change second site. "
"No test edits during bootstrap. No additional files outside the planned set."
)
self.additional = {
"new_files": [
{
"location": new_location,
"reason": "shared construct",
"constructs": [
{
"kind": "primary construct",
"name": f"BootstrapConstruct{call_index}",
"responsibility": "share responsibility",
"members": [
{"name": "apply", "purpose": "share behavior"}
],
}
],
}
],
"call_sites": [
{
"location": editable[0],
"symbol_or_beat": _exhibit_beat(selected_concern, 0),
"required_change": "change first site",
"usage": "use shared construct",
},
{
"location": editable[1],
"symbol_or_beat": _exhibit_beat(selected_concern, 1),
"required_change": "change second site",
"usage": "use shared construct",
},
],
"construction_sites": [],
"editable_locations": editable,
"readonly_locations": [],
"expected_relationships": [
{
"consumer": editable[0],
"dependency": new_location,
"reason": "consumer uses construct",
}
],
"implementation_prompt": prompt,
}
return True
return _Check(call_index <= self.designed_calls)
class _StandaloneDesignFactory(TightenDesignWorkCheckFactory):
"""_StandaloneDesignFactory returns a new-file-only bootstrap design."""
def __init__(self) -> None:
super().__init__(
client=_Client(),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
)
def create(
self,
*,
file_manager: FileManager,
original_prompt: str,
selected_concern: dict[str, object],
idiom_catalog: list[dict[str, object]],
visible_files: list[ClassMeta],
previous_construct_locations: list[str] | None = None,
) -> TightenDesignWorkCheck:
"""Create a fake Prompt 3 check for empty-source bootstrap."""
del (
file_manager,
original_prompt,
selected_concern,
idiom_catalog,
visible_files,
previous_construct_locations,
)
class _Check(TightenDesignWorkCheck):
def __init__(self) -> None:
ReasonableVibeCheck.__init__(self)
self.additional = {
"new_files": [
{
"location": "src/app.py",
"reason": "new application source",
"constructs": [
{
"kind": "module",
"name": "app",
"responsibility": "provide application behavior",
"members": [],
}
],
}
],
"call_sites": [],
"construction_sites": [],
"editable_locations": [],
"readonly_locations": [],
"expected_relationships": [],
"implementation_prompt": "Create and populate src/app.py.",
}
def __call__(self, *args: object, **kwargs: object) -> bool:
del args, kwargs
return True
return _Check()
def _file_manager(memory_root: Path, names: tuple[str, ...] = ("a.py", "b.py")) -> FileManager:
"""Create a FileManager backed solely by the in-memory repository boundary."""
repo_root = memory_root
src_root = repo_root / "src"
test_root = repo_root / "tests"
repo_exec = MemoryRepoExec(repo_root)
repo_exec.mkdir(src_root)
repo_exec.mkdir(test_root)
for name in names:
repo_exec.write_text(
src_root / name,
f"def existing_{name.replace('.', '_')}():\n return {name!r}\n",
)
parser = _Parser(repo_root)
client = _Client()
graph = _GraphMaintainer(repo_exec)
dependency = _DependencyStrategy(graph)
discovery = _Discovery(src_root, test_root, repo_exec)
def strategy_set_factory(
file_manager: FileManager,
**overrides: object,
) -> LanguageStrategySet:
del file_manager, overrides
return LanguageStrategySet(
{
SourceDiscoveryStrategy: discovery,
CodeDependencyGraphMaintainer: graph,
DependencyDiscoveryStrategy: dependency,
OOpSignatureExtract: _NoCards(),
OOpSignatureRender: _NoCardRender(),
}
)
LanguageStrategySet.set_instance(
strategy_set_factory
)
file_manager = FileManager(
parser=parser,
client=client,
repo_root=repo_root,
src_root=src_root,
test_root=test_root,
actor="tester",
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
prompt="original request",
repo_exec=repo_exec,
dependency_graph_maintainer=graph,
dependency_strategy=dependency,
)
return file_manager
def _strategy_set(file_manager: FileManager) -> LanguageStrategySet:
"""Return a bootstrap language strategy set for a FileManager."""
return LanguageStrategySet(
{
SourceDiscoveryStrategy: _Discovery(
file_manager.src_root,
file_manager.test_root,
file_manager.repo_exec,
),
CodeDependencyGraphMaintainer: file_manager.dependency_graph_maintainer,
}
)
def _current_source(file_manager: FileManager, location: str) -> ClassMeta:
"""Return one required source record from this test's explicit repository state."""
source = file_manager.current_src(location)
assert source is not None
return source
def _graph_maintainer(file_manager: FileManager) -> _GraphMaintainer:
"""Return the nominal in-memory graph maintainer installed by `_file_manager`."""
maintainer = file_manager.dependency_graph_maintainer
assert isinstance(maintainer, _GraphMaintainer)
return maintainer
def _read_source(file_manager: FileManager, name: str) -> str:
"""Read one source fixture through its owned in-memory repository boundary."""
return file_manager.repo_exec.read_text(file_manager.src_root / name)
def _write_source(file_manager: FileManager, name: str, content: str) -> None:
"""Write one source fixture through its owned in-memory repository boundary."""
file_manager.repo_exec.write_text(file_manager.src_root / name, content)
def _source_exists(file_manager: FileManager, name: str) -> bool:
"""Return whether one source fixture exists in the owned in-memory repository."""
return file_manager.repo_exec.exists(file_manager.src_root / name)
def _source_files(file_manager: FileManager, pattern: str = "*.py") -> list[Path]:
"""Return source fixture paths from the owned in-memory repository."""
return file_manager.repo_exec.list_files(
file_manager.src_root,
(pattern,),
recursive=False,
)
def _memento_paths(file_manager: FileManager) -> list[Path]:
"""Return any legacy snapshot paths stored in the in-memory source root."""
return [
path
for path in file_manager.repo_exec.list_files(
file_manager.src_root,
(),
recursive=False,
)
if path.name.startswith(".") and "~" in path.name
]
def test_prompt1_valid_concern_with_two_exact_exhibits_returns_true() -> None:
"""Verify Prompt 1 accepts two exact supplied exhibits."""
response = {
"outcome": "found",
"reason": "shared concern",
"additional": {
"concerns": [
{
"name": "shared",
"summary": "shared responsibility",
"exhibits": [
{"location": "src/a.py", "beat": "beat a", "symbol": "A", "reason": "uses it"},
{"location": "src/b.py", "beat": "beat b", "symbol": "B", "reason": "uses it"},
],
}
],
"selected_concern_index": 0,
},
}
check = CrossCuttingConcernCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
visible_files=[
ClassMeta("src/a.py", "a", ["rel"], {}),
ClassMeta("src/b.py", "b", ["rel"], {}),
],
beats_by_location={"src/a.py": ["beat a"], "src/b.py": ["beat b"]},
)
assert check() is True
assert check.additional["concerns"][0]["name"] == "shared"
def test_prompt1_valid_concern_with_one_visible_exhibit_returns_true() -> None:
"""Verify Prompt 1 accepts one visible integration-point exhibit."""
response = {
"outcome": "found",
"reason": "single integration point",
"additional": {
"concerns": [
{
"name": "screen integration",
"summary": "one file binds the requested new source file",
"exhibits": [
{
"location": "src/a.py",
"beat": "beat a",
"symbol": "A",
"reason": "binds the new feature",
}
],
}
],
"selected_concern_index": 0,
},
}
check = CrossCuttingConcernCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
visible_files=[ClassMeta("src/a.py", "a", ["rel"], {})],
beats_by_location={"src/a.py": ["beat a"]},
)
assert check() is True
exhibits = check.additional["concerns"][0]["exhibits"]
assert [exhibit["location"] for exhibit in exhibits] == ["src/a.py"]
def test_prompt1_comma_separated_symbols_do_not_become_expected_values() -> None:
"""Verify Prompt 1 does not preserve multiple symbols packed into one string."""
response = {
"outcome": "found",
"reason": "single integration point",
"additional": {
"concerns": [
{
"name": "screen integration",
"summary": "one file binds the requested new source file",
"exhibits": [
{
"location": "src/a.py",
"beat": "beat a",
"symbol": "TerminalUI, _handle_key",
"reason": "binds the new feature",
}
],
}
],
"selected_concern_index": 0,
},
}
check = CrossCuttingConcernCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
visible_files=[ClassMeta("src/a.py", "a", ["rel"], {})],
beats_by_location={"src/a.py": ["beat a"]},
)
assert check() is True
exhibit = check.additional["concerns"][0]["exhibits"][0]
assert exhibit["symbol"] == ""
@pytest.mark.parametrize(
"response",
[
{
"outcome": "found",
"reason": "invented location",
"additional": {
"concerns": [
{
"name": "shared",
"summary": "shared",
"exhibits": [
{"location": "src/a.py", "beat": "beat a", "symbol": "", "reason": "uses it"},
{"location": "src/missing.py", "beat": "beat b", "symbol": "", "reason": "uses it"},
],
}
],
"selected_concern_index": 0,
},
},
{
"outcome": "found",
"reason": "invalid field types",
"additional": {
"concerns": [
{
"name": "shared",
"summary": "shared",
"exhibits": [
{"location": "src/a.py", "beat": "beat a", "symbol": "", "reason": "uses it"},
{"location": "src/b.py", "beat": 2, "symbol": "", "reason": "uses it"},
],
}
],
"selected_concern_index": 0,
},
},
"",
],
)
def test_prompt1_invalid_outputs_retry_until_explicit_none(response: object) -> None:
"""Only a corrected explicit negative can close malformed concern selection."""
client = _Client([response, {"outcome": "none", "reason": "No supported concern.", "additional": {}}])
check = CrossCuttingConcernCheck(
client=client,
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
visible_files=[
ClassMeta("src/a.py", "a", ["rel"], {}),
ClassMeta("src/b.py", "b", ["rel"], {}),
],
beats_by_location={"src/a.py": ["beat a"], "src/b.py": ["beat b"]},
)
assert check() is False
assert check.additional["concerns"] == []
assert len(client.prompts) == 2
def test_prompt1_unknown_location_requires_corrected_judgement() -> None:
"""Verify Prompt 1 distinguishes invalid found output from a genuine none result."""
response = {
"outcome": "found",
"reason": "model used an inferred file",
"additional": {
"concerns": [
{
"name": "shared",
"summary": "shared responsibility",
"exhibits": [
{
"location": "src/a.py",
"beat": "visible explanatory beat",
"symbol": "A",
"reason": "uses it",
},
{
"location": "src/missing.py",
"beat": "unknown explanatory beat",
"symbol": "Missing",
"reason": "uses it",
},
],
}
],
"selected_concern_index": 0,
},
}
check = CrossCuttingConcernCheck(
client=_Client([response, {"outcome": "none", "reason": "No supported concern.", "additional": {}}]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
visible_files=[ClassMeta("src/a.py", "a", ["rel"], {})],
beats_by_location={"src/a.py": ["beat a"]},
)
assert check() is False
assert check.outcome == "none"
assert check.reason == "No supported concern."
def test_prompt1_accepts_visible_paraphrased_exhibit_beats() -> None:
"""Verify Prompt 1 treats returned exhibit beats as explanatory prose, not canonical keys."""
response = {
"outcome": "found",
"reason": "shared concern",
"additional": {
"concerns": [
{
"name": "shared",
"summary": "shared responsibility",
"exhibits": [
{
"location": "src/a.py",
"beat": "This site performs the reusable behavior.",
"symbol": "A",
"reason": "uses it",
},
{
"location": "src/b.py",
"beat": "This other site performs the same reusable behavior.",
"symbol": "B",
"reason": "uses it",
},
],
}
],
"selected_concern_index": 0,
},
}
check = CrossCuttingConcernCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
visible_files=[
ClassMeta("src/a.py", "a", ["rel"], {}),
ClassMeta("src/b.py", "b", ["rel"], {}),
],
beats_by_location={"src/a.py": ["beat a"], "src/b.py": ["beat b"]},
)
assert check() is True
exhibits = check.additional["concerns"][0]["exhibits"]
assert exhibits[0]["beat"] == "This site performs the reusable behavior."
assert exhibits[1]["beat"] == "This other site performs the same reusable behavior."
def test_bootstrap_derives_standalone_concern_after_corrected_none(memory_root: Path) -> None:
"""Verify invalid Prompt 1 output cannot stop bootstrap before design/editing."""
file_manager = _file_manager(memory_root)
response = {
"outcome": "found",
"reason": "model used an inferred file",
"additional": {
"concerns": [
{
"name": "shared",
"summary": "shared responsibility",
"exhibits": [
{
"location": "src/a.py",
"beat": "visible explanatory beat",
"symbol": "A",
"reason": "uses it",
},
{
"location": "src/missing.py",
"beat": "unknown explanatory beat",
"symbol": "Missing",
"reason": "uses it",
},
],
}
],
"selected_concern_index": 0,
},
}
class _InvalidFoundConcernFactory(CrossCuttingConcernCheckFactory):
def __init__(self) -> None:
super().__init__(
client=_Client(),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
)
def create(
self,
*,
original_prompt: str,
visible_files: list[ClassMeta],
beats_by_location: dict[str, list[str]],
) -> CrossCuttingConcernCheck:
return CrossCuttingConcernCheck(
client=_Client([response, {"outcome": "none", "reason": "No supported concern.", "additional": {}}]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt=original_prompt,
visible_files=visible_files,
beats_by_location=beats_by_location,
)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
stage="beats_extracted",
visible_locations=["src/a.py"],
beats_by_location={"src/a.py": ["beat a"]},
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_InvalidFoundConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_concern_check()
assert context.stage == "concern_selected"
assert context.completed is False
assert context.blocked_summary == ""
assert context.selected_concern is not None
assert context.selected_concern["name"] == "Standalone feature source"
def test_prompt2_restricts_evidence_to_current_pair() -> None:
"""Unsupported catalog evidence requires correction before an empty catalog is admitted."""
response = {
"outcome": "cataloged",
"reason": "cataloged",
"additional": {
"entries": [
{
"category": "naming",
"observation": "handlers use verb phrases",
"evidence": [
{"location": "src/missing.py", "symbol_or_excerpt": "def run"}
],
"design_implication": "use a verb phrase",
}
]
},
}
client = _Client([response, {"outcome": "cataloged", "reason": "No supported idioms.", "additional": {"entries": []}}])
check = RepositoryIdiomCheck(
client=client,
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
selected_concern={},
sample_files=[ClassMeta("src/a.py", "a", ["rel"], {})],
sources_by_location={"src/a.py": "def run(): pass"},
reference_cards="No reference cards resolved.",
)
assert check() is True
assert check.additional["entries"] == []
assert len(client.prompts) == 2
def test_prompt2_discards_timestamped_entries() -> None:
"""Verify Prompt 2 does not preserve timestamp-like catalog entries."""
response = {
"outcome": "cataloged",
"reason": "cataloged",
"additional": {
"entries": [
{
"category": "organization",
"observation": "2026-07-18 handlers live beside callers",
"evidence": [{"location": "src/a.py", "symbol_or_excerpt": "def run"}],
"design_implication": "place it nearby",
}
]
},
}
check = RepositoryIdiomCheck(
client=_Client([response, {"outcome": "cataloged", "reason": "No supported idioms.", "additional": {"entries": []}}]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
original_prompt="add behavior",
selected_concern={},
sample_files=[ClassMeta("src/a.py", "a", ["rel"], {})],
sources_by_location={"src/a.py": "def run(): pass"},
reference_cards="No reference cards resolved.",
)
check()
assert check.additional["entries"] == []
def test_prompt3_accepts_multiple_new_files_with_empty_readonly_scope(memory_root: Path) -> None:
"""Verify Prompt 3 accepts multiple source-root new files and empty readonly scope."""
file_manager = _file_manager(memory_root, ("a.py", "b.py", "context.py"))
response = {
"outcome": "designed",
"reason": "grounded",
"additional": {
"new_files": [
{
"location": "src/shared_one.py",
"reason": "first construct",
"constructs": [
{"kind": "construct", "name": "One", "responsibility": "one", "members": []}
],
},
{
"location": "src/shared_two.py",
"reason": "second construct",
"constructs": [
{"kind": "construct", "name": "Two", "responsibility": "two", "members": []}
],
},
],
"call_sites": [
{
"location": "src/a.py",
"symbol_or_beat": "beat a",
"required_change": "replace a",
"usage": "use shared",
},
{
"location": "src/b.py",
"symbol_or_beat": "beat b",
"required_change": "replace b",
"usage": "use shared",
},
],
"construction_sites": [],
"editable_locations": ["src/a.py", "src/b.py"],
"readonly_locations": [],
"expected_relationships": [
{"consumer": "src/a.py", "dependency": "src/shared_one.py", "reason": "uses it"}
],
"implementation_prompt": (
"Create src/shared_one.py and src/shared_two.py. "
"Modify src/a.py to replace a. Modify src/b.py to replace b."
),
},
}
check = TightenDesignWorkCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="add behavior",
selected_concern={
"exhibits": [
{"location": "src/a.py", "beat": "beat a", "symbol": ""},
{"location": "src/b.py", "beat": "beat b", "symbol": ""},
]
},
idiom_catalog=[],
visible_files=[
_current_source(file_manager, "src/a.py"),
_current_source(file_manager, "src/b.py"),
_current_source(file_manager, "src/context.py"),
],
)
assert check() is True
assert [item["location"] for item in check.additional["new_files"]] == [
"src/shared_one.py",
"src/shared_two.py",
]
assert check.additional["editable_locations"] == ["src/a.py", "src/b.py"]
assert check.additional["readonly_locations"] == []
def test_prompt3_accepts_new_file_with_one_existing_integration_point(memory_root: Path) -> None:
"""Verify Prompt 3 can design a new file from one visible call site."""
file_manager = _file_manager(memory_root, ("a.py",))
response = {
"outcome": "designed",
"reason": "grounded",
"additional": {
"new_files": [
{
"location": "src/report_ui.py",
"reason": "new report screen",
"constructs": [
{
"kind": "module",
"name": "report_ui",
"responsibility": "render report screen",
"members": [],
}
],
}
],
"call_sites": [
{
"location": "src/a.py",
"symbol_or_beat": "TerminalUI",
"required_change": "bind report screen",
"usage": "open report screen",
}
],
"construction_sites": [],
"editable_locations": ["src/a.py"],
"readonly_locations": [],
"expected_relationships": [
{"consumer": "src/a.py", "dependency": "src/report_ui.py", "reason": "opens it"}
],
"implementation_prompt": (
"Create src/report_ui.py. Modify src/a.py to bind report screen."
),
},
}
check = TightenDesignWorkCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="create report_ui.py",
selected_concern={
"exhibits": [
{"location": "src/a.py", "beat": "beat a", "symbol": "TerminalUI"}
]
},
idiom_catalog=[],
visible_files=[_current_source(file_manager, "src/a.py")],
)
assert check() is True
assert check.additional["editable_locations"] == ["src/a.py"]
assert check.additional["new_files"][0]["location"] == "src/report_ui.py"
def test_prompt3_expands_selected_symbols_to_full_source_definitions(memory_root: Path) -> None:
"""Verify Prompt 3 receives whole source definitions, not beat summaries."""
file_manager = _file_manager(memory_root, ("terminal_ui.py", "escalation_ui.py"))
_write_source(
file_manager,
"terminal_ui.py",
"class TerminalUI:\n"
" def _handle_key(self, key):\n"
" return key\n"
"\n"
"def helper():\n"
" return 'helper'\n"
)
_write_source(
file_manager,
"escalation_ui.py",
"class EscalationUIState:\n"
" focus_index = 0\n"
"\n"
"def handle_escalation_key(key, request, state):\n"
" return None, None\n"
)
client = _Client([
{
"outcome": "designed",
"reason": "ok",
"additional": {
"new_files": [{"location": "src/report_ui.py", "reason": "report screen", "constructs": [
{"kind": "module", "name": "report_ui", "responsibility": "Render reports.", "members": []},
]}],
"editable_locations": ["src/terminal_ui.py"],
"readonly_locations": [],
"implementation_prompt": "Create src/report_ui.py. Modify src/terminal_ui.py.",
},
}
])
check = TightenDesignWorkCheck(
client=client,
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="create report_ui.py",
selected_concern={
"name": "screen",
"summary": "screen boundary",
"exhibits": [
{
"location": "src/terminal_ui.py",
"beat": "lossy beat text",
"symbol": "TerminalUI, _handle_key",
"reason": "lossy reason",
},
{
"location": "src/escalation_ui.py",
"beat": "lossy beat text",
"symbol": "EscalationUIState, handle_escalation_key",
"reason": "lossy reason",
},
],
},
idiom_catalog=[
{
"category": "interface",
"observation": "standalone key handler",
"design_implication": "use same interface",
"evidence": [
{
"location": "src/escalation_ui.py",
"symbol_or_excerpt": "handle_escalation_key",
}
],
}
],
visible_files=[
_current_source(file_manager, "src/terminal_ui.py"),
_current_source(file_manager, "src/escalation_ui.py"),
],
)
assert check() is True
prompt = client.prompts[0]
assert "lossy beat text" not in prompt
assert "lossy reason" not in prompt
assert "class TerminalUI:" in prompt
assert "def _handle_key(self, key):" in prompt
assert "class EscalationUIState:" in prompt
assert "def handle_escalation_key(key, request, state):" in prompt
def test_prompt3_includes_step2_files_as_readonly_full_source_context(memory_root: Path) -> None:
"""Verify Prompt 3 receives whole Step 2 idiom evidence files as readonly context."""
file_manager = _file_manager(memory_root, ("terminal_ui.py", "idiom_context.py"))
_write_source(
file_manager,
"terminal_ui.py",
"class TerminalUI:\n"
" pass\n"
)
_write_source(
file_manager,
"idiom_context.py",
"class LocalFactory:\n"
" def build(self):\n"
" return 'repo idiom'\n"
)
client = _Client([
{
"outcome": "designed",
"reason": "ok",
"additional": {
"new_files": [{"location": "src/report_ui.py", "reason": "report screen", "constructs": [
{"kind": "module", "name": "report_ui", "responsibility": "Render reports.", "members": []},
]}],
"editable_locations": ["src/terminal_ui.py"],
"readonly_locations": [],
"implementation_prompt": "Create src/report_ui.py. Modify src/terminal_ui.py.",
},
}
])
check = TightenDesignWorkCheck(
client=client,
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="create report_ui.py",
selected_concern={
"name": "screen",
"summary": "screen boundary",
"exhibits": [
{
"location": "src/terminal_ui.py",
"beat": "terminal_ui",
"symbol": "TerminalUI",
"reason": "integration point",
}
],
},
idiom_catalog=[
{
"category": "construction",
"observation": "factory idiom",
"design_implication": "use local construction style",
"evidence": [
{
"location": "src/idiom_context.py",
"symbol_or_excerpt": "LocalFactory",
}
],
}
],
visible_files=[_current_source(file_manager, "src/terminal_ui.py")],
)
assert check() is True
prompt = client.prompts[0]
assert '"rendered_idiom_sources"' in prompt
assert '"location": "src/idiom_context.py"' in prompt
assert "class LocalFactory:" in prompt
assert "def build(self):" in prompt
assert check.additional["editable_locations"] == ["src/terminal_ui.py"]
assert check.additional["readonly_locations"] == []
def test_prompt3_accepts_visible_symbol_shorthand_and_sparse_construction_site(memory_root: Path) -> None:
"""Verify Prompt 3 does not drop a usable design for common grounded shorthand."""
file_manager = _file_manager(memory_root, ("terminal_ui.py", "escalation_ui.py"))
response = {
"outcome": "designed",
"reason": "grounded",
"additional": {
"new_files": [
{
"location": "src/report_ui.py",
"reason": "new report screen",
"constructs": [
{
"kind": "module",
"name": "report_ui",
"responsibility": "render report screen",
"members": [
{"name": "draw_report_screen", "purpose": "render report content"}
],
}
],
}
],
"call_sites": [
{
"location": "src/terminal_ui.py",
"symbol_or_beat": "_handle_key",
"required_change": "Delegate key events to report_ui.handle_report_key.",
"usage": "open report screen",
},
{
"location": "src/terminal_ui.py",
"symbol_or_beat": "_draw",
"required_change": "Delegate rendering to report_ui.draw_report_screen.",
"usage": "render report screen",
},
{
"location": "src/terminal_ui.py",
"symbol_or_beat": "_draw_completion",
"required_change": "Add a visual option for Open Report.",
"usage": "select report screen from completion",
},
],
"construction_sites": [
{
"location": "src/terminal_ui.py",
"required_change": "Initialize report UI state and data fields.",
}
],
"editable_locations": ["src/terminal_ui.py"],
"readonly_locations": [],
"expected_relationships": [
{
"consumer": "src/terminal_ui.py",
"dependency": "src/report_ui.py",
"reason": "Terminal UI opens the report screen.",
}
],
"implementation_prompt": (
"Create src/report_ui.py with real report UI code. "
"Modify src/terminal_ui.py to delegate key handling, rendering, "
"completion-menu selection, and state construction for report_ui."
),
},
}
terminal_ui = _current_source(file_manager, "src/terminal_ui.py")
terminal_ui.relevant_symbols = {
"TerminalUI": "main UI orchestrator",
"_handle_key": "handles key input",
"_draw": "draws active screen",
"_draw_completion": "draws completion menu",
}
escalation_ui = _current_source(file_manager, "src/escalation_ui.py")
escalation_ui.relevant_symbols = {
"EscalationUIState": "modal state",
"handle_escalation_key": "modal key handler",
}
check = TightenDesignWorkCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="create report_ui.py",
selected_concern={
"exhibits": [
{
"location": "src/terminal_ui.py",
"beat": "terminal_ui",
"symbol": "TerminalUI, _handle_key",
},
{
"location": "src/escalation_ui.py",
"beat": "escalation_ui",
"symbol": "EscalationUIState, handle_escalation_key",
},
]
},
idiom_catalog=[],
visible_files=[terminal_ui, escalation_ui],
)
assert check() is True
assert [site["symbol_or_beat"] for site in check.additional["call_sites"]] == [
"_handle_key",
"_draw",
"_draw_completion",
]
assert check.additional["construction_sites"][0]["symbol_or_beat"] == ""
def test_prompt3_accepts_new_file_with_no_existing_visible_files(memory_root: Path) -> None:
"""Verify Prompt 3 can design a standalone new source file under src_root."""
file_manager = _file_manager(memory_root, ())
response = {
"outcome": "designed",
"reason": "new application entrypoint",
"additional": {
"new_files": [
{
"location": "src/app.py",
"reason": "new application source",
"constructs": [
{
"kind": "module",
"name": "app",
"responsibility": "provide the application entrypoint",
"members": [],
}
],
}
],
"call_sites": [],
"construction_sites": [],
"editable_locations": [],
"readonly_locations": [],
"expected_relationships": [],
"implementation_prompt": "Create src/app.py with the application entrypoint.",
},
}
check = TightenDesignWorkCheck(
client=_Client([response]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="create a new application",
selected_concern={"exhibits": []},
idiom_catalog=[],
visible_files=[],
)
assert check() is True
assert check.additional["new_files"][0]["location"] == "src/app.py"
assert check.additional["editable_locations"] == []
@pytest.mark.parametrize(
"mutator",
[
lambda payload: payload["additional"]["editable_locations"].append("src/missing.py"),
lambda payload: payload["additional"]["new_files"].append(
{
"location": "outside.py",
"reason": "outside",
"constructs": [{"kind": "construct", "name": "Outside", "responsibility": "bad", "members": []}],
}
),
lambda payload: payload["additional"]["call_sites"][0].update({"location": "src/missing.py"}),
lambda payload: payload["additional"].update({"implementation_prompt": "too vague"}),
],
)
def test_prompt3_invalid_designs_retry_until_corrected_design(memory_root: Path, mutator) -> None:
"""Verify malformed Prompt 3 output cannot stop bootstrap design."""
file_manager = _file_manager(memory_root, ("a.py", "b.py"))
payload = {
"outcome": "designed",
"reason": "grounded",
"additional": {
"new_files": [
{
"location": "src/shared.py",
"reason": "shared",
"constructs": [
{"kind": "construct", "name": "Shared", "responsibility": "shared", "members": []}
],
}
],
"call_sites": [
{"location": "src/a.py", "symbol_or_beat": "beat a", "required_change": "replace a", "usage": "use it"},
{"location": "src/b.py", "symbol_or_beat": "beat b", "required_change": "replace b", "usage": "use it"},
],
"construction_sites": [],
"editable_locations": ["src/a.py", "src/b.py"],
"readonly_locations": [],
"expected_relationships": [],
"implementation_prompt": "Create src/shared.py. Modify src/a.py to replace a. Modify src/b.py to replace b.",
},
}
corrected = json.loads(json.dumps(payload))
mutator(payload)
check = TightenDesignWorkCheck(
client=_Client([payload, corrected]),
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
file_manager=file_manager,
original_prompt="add behavior",
selected_concern={
"exhibits": [
{"location": "src/a.py", "beat": "beat a", "symbol": ""},
{"location": "src/b.py", "beat": "beat b", "symbol": ""},
]
},
idiom_catalog=[],
visible_files=[
_current_source(file_manager, "src/a.py"),
_current_source(file_manager, "src/b.py"),
],
)
assert check() is True
assert check.outcome == "designed"
assert check.additional["new_files"][0]["location"] == "src/shared.py"
assert check.additional["editable_locations"] == ["src/a.py", "src/b.py"]
@pytest.mark.parametrize("legacy", (False, True))
def test_bootstrap_context_persistence_and_reassembly(memory_root: Path, legacy: bool) -> None:
"""Verify BootstrapContext stores durable fields and strips runtime collaborators."""
file_manager = _file_manager(memory_root)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
iteration=1,
stage="cataloging_idioms",
search_anchors=("AlphaAnchor", "beta.py"),
visible_locations=["src/a.py"],
beats_by_location={"src/a.py": ["beat"]},
pending_idiom_pairs=[("src/a.py", "src/b.py")],
editor_summaries=["summary"],
editor_records=[EditorRecord("retained bootstrap repair", "before", "after")],
)
state = context.__getstate__()
assert state["stage"] == "cataloging_idioms"
assert state["search_anchors"] == ("AlphaAnchor", "beta.py")
assert state["pending_idiom_pairs"] == [("src/a.py", "src/b.py")]
assert "upstream_file_manager" not in state
# Older checkpoint payloads predate optional Work Logs, but retain the same workflow state.
if legacy:
state.pop("editor_records")
thawed = BootstrapContext.__new__(BootstrapContext)
BootstrapContextReassembly().thaw(
thawed,
LoadedCheckpoint(2, state),
{"upstream_file_manager": file_manager},
)
assert thawed.upstream_file_manager is file_manager
assert thawed.search_anchors == ("AlphaAnchor", "beta.py")
assert thawed.pending_idiom_pairs == [("src/a.py", "src/b.py")]
assert thawed.editor_records == ([] if legacy else [EditorRecord("retained bootstrap repair", "before", "after")])
def test_context_factory_discards_checkpoint_with_stale_shape(memory_root: Path) -> None:
"""Verify hydration discards an old shape instead of migrating its missing fields."""
file_manager = _file_manager(memory_root)
thawed = BootstrapContext.__new__(BootstrapContext)
restored = context_factory._thaw(
thawed,
LoadedCheckpoint(
1,
{
"goal": "goal",
"classes": [],
"pending_legacy_stage_data": [],
},
),
{"upstream_file_manager": file_manager},
)
assert restored is None
def test_prompt2_pair_ordering_is_deterministic_and_bounded(memory_root: Path) -> None:
"""Verify Prompt 2 pairs are stable and never larger than two files."""
file_manager = _file_manager(memory_root, ("a.py", "b.py", "c.py", "d.py", "e.py"))
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=[source.location for source in file_manager.src_classes],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
first = loop._deterministic_pairs(context.visible_locations)
second = loop._deterministic_pairs(context.visible_locations)
assert first == second
assert all(len(pair) <= 2 for pair in first)
def test_prompt2_resume_preserves_remaining_pairs_and_catalog_dedup(memory_root: Path) -> None:
"""Verify pending Prompt 2 pairs are preserved and catalog merge is deduplicated."""
file_manager = _file_manager(memory_root)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=["src/a.py", "src/b.py"],
pending_idiom_pairs=[("src/b.py",), ("src/a.py",)],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
entry = {
"category": "naming",
"observation": "names are specific",
"evidence": [{"location": "src/a.py", "symbol_or_excerpt": "def existing"}],
"design_implication": "use a specific name",
}
loop._merge_idiom_entries([entry, dict(entry)])
assert context.pending_idiom_pairs == [("src/b.py",), ("src/a.py",)]
assert context.idiom_catalog == [entry]
def test_bootstrap_beat_extraction_keeps_every_available_beat(memory_root: Path) -> None:
"""Verify bootstrap carries every in-memory searchable beat into its mapping context."""
file_manager = _file_manager(memory_root, ("a.py",))
beats = [f"def behavior_{index}(): return {index}" for index in range(8)]
_write_source(file_manager, "a.py", "\n".join(beats))
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=["src/a.py"],
search_anchors=tuple(beats),
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
)
loop._extract_visible_beats()
assert context.beats_by_location == {"src/a.py": beats}
class _EmptyBeatDiscovery(_Discovery):
"""Represent a language strategy that found no relevant beat guidance."""
def searchable_beats(
self,
file_name: str,
source: str,
) -> list[SearchableBeatValue]:
del file_name, source
return []
def test_bootstrap_keeps_visible_files_when_no_beats_are_selected(memory_root: Path) -> None:
"""Verify empty selected guidance does not remove visible files or complete bootstrap early."""
file_manager = _file_manager(memory_root)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=[source.location for source in file_manager.src_classes],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=LanguageStrategySet(
{
SourceDiscoveryStrategy: _EmptyBeatDiscovery(
file_manager.src_root,
file_manager.test_root,
file_manager.repo_exec,
)
}
),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._extract_visible_beats()
assert context.visible_locations == ["src/a.py", "src/b.py"]
assert context.beats_by_location == {"src/a.py": [], "src/b.py": []}
assert context.completed is False
assert context.stage == "beats_extracted"
def test_bootstrap_keeps_visible_files_when_extraction_fails(memory_root: Path) -> None:
"""Verify beat extraction failure stores empty guidance without dropping the file."""
file_manager = _file_manager(memory_root)
class _FailingDiscovery(_Discovery):
def searchable_beats(
self,
file_name: str,
source: str,
) -> list[SearchableBeatValue]:
del file_name, source
raise RuntimeError("boom")
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=["src/a.py"],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=LanguageStrategySet(
{
SourceDiscoveryStrategy: _FailingDiscovery(
file_manager.src_root,
file_manager.test_root,
file_manager.repo_exec,
)
}
),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._extract_visible_beats()
assert context.visible_locations == ["src/a.py"]
assert context.beats_by_location == {"src/a.py": []}
assert context.stage == "beats_extracted"
class _BootstrapAcceptanceClient(_Client):
"""_BootstrapAcceptanceClient returns correction and sequenced acceptance decisions."""
def __init__(self, acceptance: list[dict[str, object]] | None = None) -> None:
"""Seed acceptance responses with evidence emitted by the stationary editor."""
super().__init__()
self.acceptance = list(
acceptance
or [{"outcome": "pass", "reason": ["# wired"], "instruction": ""}]
)
self.correction_prompts = 0
self.acceptance_prompts = 0
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
del logger
self.prompts.append(prompt)
if "Bootstrap source preflight input:" in prompt:
return json.dumps([])
if "Infer the missing Alpha source-edit step" in prompt:
self.correction_prompts += 1
return json.dumps(
{
"instruction": "Populate src/shared.py with the shared implementation.",
"reason": "The bootstrap source is still blank.",
"blank_files": ["src/shared.py"],
}
)
if "ACCEPTANCE ROLE" in prompt and "ORIGINAL AIDER EDIT REQUEST" in prompt:
self.acceptance_prompts += 1
response = self.acceptance.pop(0) if self.acceptance else {
"outcome": "pass",
"reason": ["# wired"],
"instruction": "",
}
return json.dumps(response)
return json.dumps({"cdc": "metadata refreshed"})
class _BlankThenFilledBootstrapEditor(_Editor):
"""_BlankThenFilledBootstrapEditor leaves the planned new file blank once, then fills it."""
def edit(
self,
request: AiderEditRequest,
logger: logging.Logger | None = None,
) -> CodeEditorResult:
del logger
if self.file_manager is None:
raise AssertionError("Bootstrap editor requires its FileManager boundary.")
self.requests.append(request)
attempt = len(self.requests)
changed: list[Path] = []
for path in request.editable_files:
if path.name == "shared.py":
if attempt == 1:
self.file_manager.repo_exec.write_text(path, "")
else:
self.file_manager.repo_exec.write_text(
path,
"def shared():\n return 'ok'\n",
)
else:
self.file_manager.repo_exec.append_text(
path,
f"\n# bootstrap attempt {attempt}\n",
)
changed.append(path)
return CodeEditorResult(changed, f"attempt {attempt}", [])
def _bootstrap_editor_context(file_manager: FileManager) -> BootstrapContext:
"""Return a context at the bootstrap editing stage."""
return BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
stage="editing",
visible_locations=[source.location for source in file_manager.src_classes],
selected_concern={
"exhibits": [
{"location": "src/a.py", "beat": "a", "symbol": ""},
{"location": "src/b.py", "beat": "b", "symbol": ""},
]
},
design_plan={
"new_files": [{"location": "src/shared.py", "reason": "shared"}],
"editable_locations": ["src/a.py", "src/b.py"],
"readonly_locations": [],
},
implementation_prompt="Create src/shared.py. Modify src/a.py and src/b.py.",
)
def _existing_bootstrap_editor_context(
file_manager: FileManager,
*,
max_iterations: int = 3,
) -> BootstrapContext:
"""Return an editing-stage context whose complete source scope already exists."""
return BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
stage="editing",
visible_locations=[source.location for source in file_manager.src_classes],
selected_concern={
"exhibits": [
{"location": "src/a.py", "beat": "a", "symbol": ""},
{"location": "src/b.py", "beat": "b", "symbol": ""},
]
},
design_plan={
"new_files": [],
"editable_locations": ["src/a.py", "src/b.py"],
"readonly_locations": [],
},
implementation_prompt="Modify src/a.py and src/b.py together.",
max_iterations=max_iterations,
)
def test_bootstrap_failed_attempt_continues_from_mutated_state(
memory_root: Path,
) -> None:
"""A rejected attempt leaves its edits for the next Bootstrap retry."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient(
[
{
"outcome": "fail",
"reason": ["attempt 1: a.py", "attempt 1: b.py"],
"instruction": "retry from the restored files",
},
{"outcome": "pass", "reason": ["attempt 2: a.py", "attempt 2: b.py"], "instruction": ""},
]
)
context = _existing_bootstrap_editor_context(file_manager)
class _RejectThenAcceptEditor(_Editor):
def edit(
self,
request: AiderEditRequest,
logger: logging.Logger | None = None,
) -> CodeEditorResult:
del logger
self.requests.append(request)
attempt = len(self.requests)
if attempt == 2:
assert _read_source(file_manager, "a.py") == "attempt 1: a.py\n"
assert _memento_paths(file_manager) == []
for path in request.editable_files:
file_manager.repo_exec.write_text(
path,
f"attempt {attempt}: {path.name}\n",
)
return CodeEditorResult(list(request.editable_files), f"attempt {attempt}", [])
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_RejectThenAcceptEditor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert _read_source(file_manager, "a.py") == "attempt 2: a.py\n"
def test_bootstrap_accepted_edit_retains_file_content(memory_root: Path) -> None:
"""Accepted Bootstrap edits remain on disk without Memento commits."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient()
context = _existing_bootstrap_editor_context(file_manager)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert "# wired" in _read_source(file_manager, "a.py")
assert _memento_paths(file_manager) == []
def test_bootstrap_has_no_snapshot_layers_between_iterations(memory_root: Path) -> None:
"""A retry leaves no snapshot layer in the owned repository state."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient(
[
{
"outcome": "fail",
"reason": ["# wired"],
"instruction": "try again",
},
{"outcome": "pass", "reason": ["# wired"], "instruction": ""},
]
)
context = _existing_bootstrap_editor_context(file_manager)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert context.iteration == 1
assert _memento_paths(file_manager) == []
def test_bootstrap_success_leaves_zero_mementos(memory_root: Path) -> None:
"""Complete Bootstrap success keeps edits and creates no snapshots."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient()
context = _existing_bootstrap_editor_context(file_manager)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert context.completed is True
assert context.stage == "completed"
assert "# wired" in _read_source(file_manager, "a.py")
assert _memento_paths(file_manager) == []
def test_bootstrap_retry_exhaustion_preserves_every_edited_file(memory_root: Path) -> None:
"""Blocked Bootstrap preserves cumulative editor changes for Stabilization."""
file_manager = _file_manager(memory_root)
originals = {
path: file_manager.repo_exec.read_bytes(path)
for path in _source_files(file_manager)
}
file_manager.client = _BootstrapAcceptanceClient(
[
{
"outcome": "fail",
"reason": ["# wired"],
"instruction": "retry",
}
]
)
context = _existing_bootstrap_editor_context(file_manager, max_iterations=1)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert context.stage == "blocked"
assert all(
file_manager.repo_exec.read_bytes(path) != original
for path, original in originals.items()
)
assert _memento_paths(file_manager) == []
def test_bootstrap_unexpected_exception_preserves_partial_edits(
memory_root: Path,
) -> None:
"""An editor exception leaves partial one-way Bootstrap edits on disk."""
file_manager = _file_manager(memory_root)
originals = {
path: file_manager.repo_exec.read_bytes(path)
for path in _source_files(file_manager)
}
context = _existing_bootstrap_editor_context(file_manager)
class _ExplodingEditor(_Editor):
def edit(
self,
request: AiderEditRequest,
logger: logging.Logger | None = None,
) -> CodeEditorResult:
del logger
if self.file_manager is None:
raise AssertionError("Bootstrap editor requires its FileManager boundary.")
for path in request.editable_files:
self.file_manager.repo_exec.write_text(
path,
"partial exceptional edit\n",
)
raise RuntimeError("editor exploded")
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_ExplodingEditor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
with pytest.raises(RuntimeError, match="editor exploded"):
loop._run_editor()
assert all(
file_manager.repo_exec.read_text(path) == "partial exceptional edit\n"
for path in originals
)
assert _memento_paths(file_manager) == []
def test_bootstrap_creates_no_memento_stack(memory_root: Path) -> None:
"""Bootstrap has no snapshot stack during retries."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient(
[
{
"outcome": "fail",
"reason": ["# wired"],
"instruction": "retry",
},
{"outcome": "pass", "reason": ["# wired"], "instruction": ""},
]
)
context = _existing_bootstrap_editor_context(file_manager)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert context.iteration == 1
assert _memento_paths(file_manager) == []
def test_alpha_start_after_bootstrap_cannot_revert_accepted_work(memory_root: Path) -> None:
"""Alpha's stale-state clear starts from accepted Bootstrap bytes, not its old baseline."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient()
context = _existing_bootstrap_editor_context(file_manager)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
source = _current_source(file_manager, "src/a.py")
accepted = file_manager.repo_exec.read_bytes(file_manager.src_root / "a.py")
source.set_or_get_memento(file_manager).clear()
assert file_manager.repo_exec.read_bytes(file_manager.src_root / "a.py") == accepted
assert not file_manager.repo_exec.exists(file_manager.src_root / ".a.py~")
def test_bootstrap_memento_preparation_helpers_are_removed() -> None:
"""Bootstrap exposes no helper that obscures direct memento ownership."""
assert not hasattr(BootstrapLoop, "_prepare_bootstrap_outer_mementos")
assert not hasattr(BootstrapLoop, "_prepare_bootstrap_attempt_mementos")
def test_bootstrap_editor_retries_until_blank_source_and_acceptance_checks_pass(memory_root: Path) -> None:
"""Verify bootstrap uses Alpha blank-source and edit-acceptance checks before leaving."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient([
{"outcome": "pass", "reason": ["def shared():", "return 'ok'"], "instruction": ""},
])
context = _bootstrap_editor_context(file_manager)
editor = _BlankThenFilledBootstrapEditor()
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert len(editor.requests) == 2
assert file_manager.client.correction_prompts == 1
assert file_manager.client.acceptance_prompts == 1
assert _read_source(file_manager, "shared.py") == "def shared():\n return 'ok'\n"
assert "BOOTSTRAP EDIT ACCEPTANCE FAILED" in editor.requests[1].backlog_guidance[-1]
assert not _source_exists(file_manager, ".shared.py~")
assert not _source_exists(file_manager, ".shared.py~1")
assert context.iteration == 1
assert context.stage == "completed"
assert context.completed is True
def test_bootstrap_preflights_planned_blank_files_before_edit_request(memory_root: Path) -> None:
"""Verify bootstrap creates planned new source files before invoking the editor."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient([
{"outcome": "pass", "reason": ["def shared():", "return 'ok'"], "instruction": ""},
])
context = _bootstrap_editor_context(file_manager)
class _PreflightAssertingEditor(_BlankThenFilledBootstrapEditor):
def edit(
self,
request: AiderEditRequest,
logger: logging.Logger | None = None,
) -> CodeEditorResult:
shared = file_manager.src_root / "shared.py"
assert shared in request.editable_files
assert _source_exists(file_manager, "shared.py")
assert _read_source(file_manager, "shared.py") == ""
return super().edit(request, logger=logger)
editor = _PreflightAssertingEditor()
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert _source_exists(file_manager, "shared.py")
assert file_manager.current_src("src/shared.py") is not None
def test_bootstrap_source_preflight_can_add_required_new_file(memory_root: Path) -> None:
"""Verify source-context preflight can add a structurally required new source file."""
file_manager = _file_manager(memory_root)
class _BootstrapPreflightClient(_Client):
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
del logger
self.prompts.append(prompt)
if "Bootstrap source preflight input:" in prompt:
return json.dumps(
[
{
"location": "src/extracted.py",
"reason": "required extraction target",
}
]
)
if "ACCEPTANCE ROLE" in prompt and "ORIGINAL AIDER EDIT REQUEST" in prompt:
return json.dumps({"outcome": "pass", "reason": ["# wired"], "instruction": ""})
return json.dumps({"cdc": "metadata refreshed"})
file_manager.client = _BootstrapPreflightClient()
context = _bootstrap_editor_context(file_manager)
editor = _Editor()
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
request = editor.requests[0]
assert file_manager.src_root / "extracted.py" in request.editable_files
assert _current_source(file_manager, "src/extracted.py").relevance == [
"relevance from goal",
]
assert _read_source(file_manager, "extracted.py") == "\n# wired\n"
assert any("BEGIN EDITABLE SOURCE FILES" in prompt for prompt in file_manager.client.prompts)
def test_bootstrap_editor_preserves_failed_acceptance_before_retry(memory_root: Path) -> None:
"""Verify acceptance failure leaves the current edit for the next retry."""
file_manager = _file_manager(memory_root)
file_manager.client = _BootstrapAcceptanceClient(
[
{
"outcome": "fail",
"reason": ["return 'bad'"],
"instruction": "Replace the rejected implementation with the accepted one.",
},
{"outcome": "pass", "reason": ["return 'good'"], "instruction": ""},
]
)
context = _bootstrap_editor_context(file_manager)
class _AcceptanceRetryEditor(_BlankThenFilledBootstrapEditor):
"""Retain each applied bootstrap block while acceptance requests another attempt."""
def edit(
self,
request: AiderEditRequest,
logger: logging.Logger | None = None,
) -> CodeEditorResult:
"""Expose the entry history before adding the current retained source edit."""
expected = [] if not self.requests else [EditorRecord("first attempt", "return 1", "return 'bad'")]
assert request.editor_records == expected
_edit_result = super().edit(request, logger=logger)
changed, summary = _edit_result.changed_paths, _edit_result.summary
_write_source(
file_manager,
"shared.py",
"def shared():\n return 'bad'\n"
if len(self.requests) == 1
else "def shared():\n return 'good'\n"
)
record = (EditorRecord("first attempt", "return 1", "return 'bad'") if len(self.requests) == 1
else EditorRecord("second attempt", "return 'bad'", "return 'good'"))
return CodeEditorResult(changed, summary, [*request.editor_records, record])
editor = _AcceptanceRetryEditor()
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert len(editor.requests) == 2
assert "Replace the rejected implementation" in editor.requests[1].backlog_guidance[-1]
assert editor.requests[1].editor_records == [EditorRecord("first attempt", "return 1", "return 'bad'"),
EditorRecord("second attempt", "return 'bad'", "return 'good'")]
assert context.completed and context.editor_records == []
assert _read_source(file_manager, "shared.py") == "def shared():\n return 'good'\n"
assert not _source_exists(file_manager, ".shared.py~")
def test_editor_request_is_one_coordinated_source_only_scope(memory_root: Path) -> None:
"""Verify bootstrap editor requests include all planned files in one source-only scope."""
file_manager = _file_manager(memory_root, ("a.py", "b.py", "context.py"))
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
design_plan={
"new_files": [{"location": "src/shared.py", "reason": "shared"}],
"editable_locations": ["src/a.py", "src/b.py"],
"readonly_locations": ["src/context.py"],
},
implementation_prompt="Create src/shared.py. Modify src/a.py and src/b.py.",
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
request = loop._edit_request()
assert request.editable_files == [
file_manager.src_root / "a.py",
file_manager.src_root / "b.py",
file_manager.src_root / "shared.py",
]
assert request.readonly_files == [file_manager.src_root / "context.py"]
assert all(file_manager.test_root not in path.parents for path in request.editable_files)
def test_design_stage_adds_upstream_sources_as_readonly_context(memory_root: Path) -> None:
"""Verify direct upstream source dependencies are available to the bootstrap editor."""
file_manager = _file_manager(memory_root, ("a.py", "b.py", "context.py"))
_graph_maintainer(file_manager)._graph = CodeDependencyGraph(
source_by_location={
source.location: source
for source in file_manager.src_classes
},
source_dependencies={"src/a.py": {"src/context.py"}},
)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
stage="designing",
visible_locations=["src/a.py", "src/b.py", "src/context.py"],
selected_concern={
"exhibits": [
{"location": "src/a.py", "beat": "a", "symbol": "", "reason": "a"},
{"location": "src/b.py", "beat": "b", "symbol": "", "reason": "b"},
]
},
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_design_check()
request = loop._edit_request()
assert file_manager.src_root / "context.py" in request.readonly_files
def test_editor_validation_rejects_blank_or_unplanned_changes(memory_root: Path) -> None:
"""Verify bootstrap rejects blank planned files and unplanned changed paths."""
file_manager = _file_manager(memory_root)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
design_plan={
"new_files": [{"location": "src/shared.py", "reason": "shared"}],
"editable_locations": ["src/a.py", "src/b.py"],
},
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
_write_source(file_manager, "shared.py", "")
with pytest.raises(ValueError, match="remained blank"):
loop._record_accepted_editor_result(
[file_manager.src_root / "shared.py"],
require_changes=True,
)
_write_source(file_manager, "shared.py", "# ok\n")
with pytest.raises(ValueError, match="outside planned"):
loop._record_accepted_editor_result(
[file_manager.src_root / "extra.py"],
require_changes=True,
)
def test_bootstrap_records_all_planned_editable_sources_as_involved(memory_root: Path) -> None:
"""Verify unchanged editables still seed the first stabilization surrogate set."""
file_manager = _file_manager(memory_root)
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
design_plan={
"new_files": [{"location": "src/shared.py", "reason": "shared"}],
"editable_locations": ["src/a.py", "src/b.py"],
},
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
_write_source(file_manager, "shared.py", "# ok\n")
loop._record_accepted_editor_result(
[
file_manager.src_root / "shared.py",
file_manager.src_root / "a.py",
],
require_changes=True,
)
assert context.involved_locations == ["src/shared.py", "src/a.py", "src/b.py"]
def test_preflight_registers_planned_blank_source_metadata_without_fabricating_graph_edges(memory_root: Path) -> None:
"""Verify bootstrap preflight creates metadata without inventing static relationships."""
file_manager = _file_manager(memory_root, ("a.py", "context.py"))
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
design_plan={
"new_files": [
{
"location": "src/shared.py",
"reason": "shared reason",
"constructs": [
{
"kind": "module",
"name": "SharedThing",
"responsibility": "provide shared behavior",
"members": [],
}
],
}
],
"editable_locations": ["src/a.py"],
"call_sites": [
{
"location": "src/a.py",
"symbol_or_beat": "A.use_shared",
"required_change": "wire A.use_shared to SharedThing",
"usage": "A calls the new shared behavior",
}
],
"expected_relationships": [
{
"consumer": "src/a.py",
"dependency": "src/shared.py",
"reason": "a uses shared",
}
],
},
idiom_catalog=[
{
"category": "construction",
"observation": "context idiom",
"design_implication": "follow context style",
"evidence": [
{
"location": "src/context.py",
"symbol_or_excerpt": "Context",
}
],
}
],
implementation_prompt="Create src/shared.py. Modify src/a.py.",
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._preflight_planned_blank_sources()
loop._register_preflight_participant_relevance()
assert _source_exists(file_manager, "shared.py")
assert _read_source(file_manager, "shared.py") == ""
assert _current_source(file_manager, "src/shared.py").summary == "shared reason"
assert any(
"New file reason: shared reason" in item
and "SharedThing: provide shared behavior" in item
for item in _current_source(file_manager, "src/shared.py").relevance
)
assert any(
"Call site: A.use_shared" in item
and "wire A.use_shared to SharedThing" in item
and "src/a.py depends on src/shared.py" in item
for item in _current_source(file_manager, "src/a.py").relevance
)
assert any(
"Step 2 idiom evidence: context idiom" in item
and "Design implication: follow context style" in item
for item in _current_source(file_manager, "src/context.py").relevance
)
graph = _graph_maintainer(file_manager).graph
assert graph.has_source("src/shared.py")
assert graph.source_dependencies.get("src/shared.py", set()) == set()
assert "src/shared.py" not in graph.source_dependencies.get("src/a.py", set())
assert any(
"Step 2 idiom evidence: context idiom" in item
for item in graph.source_by_location["src/context.py"].relevance
)
def test_preflight_registers_existing_planned_source_metadata_without_overriding(memory_root: Path) -> None:
"""Verify existing planned files get metadata only when ClassMeta is absent."""
file_manager = _file_manager(memory_root, ("a.py",))
_write_source(file_manager, "shared.py", "def shared():\n return 'already here'\n")
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
design_plan={
"new_files": [{"location": "src/shared.py", "reason": "planned shared"}],
"editable_locations": ["src/a.py"],
},
implementation_prompt="Use src/shared.py from src/a.py.",
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._preflight_planned_blank_sources()
current = file_manager.current_src("src/shared.py")
assert current is not None
assert current.summary == "planned shared"
assert "already here" in _read_source(file_manager, "shared.py")
assert any("registered planned source metadata" in entry[2] for entry in current.cdc)
current.summary = "existing summary wins"
loop._preflight_planned_blank_sources()
assert _current_source(file_manager, "src/shared.py").summary == "existing summary wins"
def test_refresh_src_relevance_replaces_stale_relevance_and_preserves_cdc(memory_root: Path) -> None:
"""Verify targeted refresh replaces relevance and preserves CDC history."""
file_manager = _file_manager(memory_root)
source = _current_source(file_manager, "src/a.py")
source.relevance = ["stale"]
source.relevant_symbols = {"old": "stale"}
source.cdc.append((source.cdc[0][0] if source.cdc else __import__("datetime").datetime.now(), "actor", "old cdc"))
parser = file_manager.parser
assert isinstance(parser, _Parser)
file_manager.client = _Client([{"cdc": "bootstrap cdc"}])
refreshed = file_manager.refresh_src_relevance(
"src/a.py",
relevance_prompt="original request",
cdc_prompt="frozen implementation instruction",
tailing_logs="editor summary",
)
assert refreshed.relevance == ["relevance from original request"]
assert refreshed.relevant_symbols == {"symbol": "symbol from original request"}
assert any(entry[2] == "old cdc" for entry in refreshed.cdc)
assert any(entry[2] == "bootstrap cdc" for entry in refreshed.cdc)
assert parser.calls[-1] == ("a.py", "original request")
def test_successful_bootstrap_edit_runs_one_design_cycle(memory_root: Path) -> None:
"""Verify an accepted bootstrap edit does not start a second design cycle."""
file_manager = _file_manager(memory_root, ("a.py", "b.py", "c.py"))
concern_factory = _ConcernFactory(found_calls=1)
editor = _Editor()
context = BootstrapContext(
goal="original request",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=[source.location for source in file_manager.src_classes],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=concern_factory,
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(designed_calls=1),
)
loop.run()
assert set(concern_factory.visible_calls[0]) == {"src/a.py", "src/b.py", "src/c.py"}
assert len(concern_factory.visible_calls) == 1
assert file_manager.current_src("src/bootstrap_construct_1.py") is not None
assert _current_source(file_manager, "src/a.py").relevance[0] == "relevance from original request"
assert all(
"Bootstrap preflight editable relevance" not in reason
for reason in _current_source(file_manager, "src/a.py").relevance
)
assert context.completed is True
def test_prompt1_none_uses_standalone_concern_and_runs_editor(memory_root: Path) -> None:
"""Verify a first Prompt 1 none result still proceeds through bootstrap editing."""
file_manager = _file_manager(memory_root)
editor = _Editor()
context = BootstrapContext(
goal="original request",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=[source.location for source in file_manager.src_classes],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(found_calls=0),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop.run()
assert context.completed is True
assert context.created_locations
assert editor.requests
def test_bootstrap_empty_visible_sources_creates_new_file_without_targeted_refresh(memory_root: Path) -> None:
"""Verify feature bootstrap seeds a new source while preserving pre-COT relevance."""
file_manager = _file_manager(memory_root, ())
editor = _Editor()
context = BootstrapContext(
goal="create a new application in app.py",
classes=[],
upstream_file_manager=file_manager,
visible_locations=[],
max_iterations=1,
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(found_calls=0),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_StandaloneDesignFactory(),
)
loop.run()
assert context.completed is True
assert context.created_locations == ["src/app.py"]
assert context.involved_locations == ["src/app.py"]
assert file_manager.current_src("src/app.py") is not None
assert _current_source(file_manager, "src/app.py").relevance
assert _graph_maintainer(file_manager).refresh_calls.count("src/app.py") >= 2
assert _read_source(file_manager, "app.py") == "\n# wired\n"
def test_prompt3_none_uses_request_derived_design_and_runs_editor(memory_root: Path) -> None:
"""Verify a Prompt 3 none result still proceeds through bootstrap editing."""
file_manager = _file_manager(memory_root)
editor = _Editor()
context = BootstrapContext(
goal="original request",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=[source.location for source in file_manager.src_classes],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(found_calls=1),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(designed_calls=0),
)
loop.run()
assert context.completed is True
assert editor.requests
assert context.created_locations
def test_successful_bootstrap_stops_without_spending_retry_budget(memory_root: Path) -> None:
"""Verify bootstrap does not spend retry attempts after an accepted edit."""
file_manager = _file_manager(memory_root)
concern_factory = _ConcernFactory(found_calls=10)
context = BootstrapContext(
goal="original request",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
visible_locations=[source.location for source in file_manager.src_classes],
)
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=_Editor(),
cross_cutting_concern_check_factory=concern_factory,
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(designed_calls=10),
)
loop.run()
assert context.iteration == 0
assert len(concern_factory.visible_calls) == 1
def test_bootstrap_leave_it_completes_without_refresh(memory_root: Path) -> None:
"""Verify LEAVE_IT is an authoritative bootstrap success with no refresh."""
file_manager = _file_manager(memory_root)
file_manager.client = _Client()
class _LeaveItEditor(_Editor):
def edit(self, request, logger=None):
self.requests.append(request)
return CodeEditorResult([], CodeEditLeaveSummary("\n".join(
[
"CODE EDIT OR LEAVE IT",
CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN,
"Judgement summary:",
"Already satisfies the request.",
]
)), [])
context = BootstrapContext(
goal="goal",
classes=list(file_manager.src_classes),
upstream_file_manager=file_manager,
stage="editing",
visible_locations=["src/a.py", "src/b.py"],
design_plan={
"new_files": [],
"editable_locations": ["src/a.py"],
"readonly_locations": [],
},
implementation_prompt="Ensure src/a.py already satisfies the request.",
)
editor = _LeaveItEditor()
loop = BootstrapLoop(
context=context,
language_strategy_set=_strategy_set(file_manager),
aider_worker=editor,
cross_cutting_concern_check_factory=_ConcernFactory(),
repository_idiom_check_factory=_IdiomFactory(),
tighten_design_work_check_factory=_DesignFactory(),
)
loop._run_editor()
assert context.completed is True
assert context.stage == "completed"
assert context.involved_locations == ["src/a.py"]
assert _graph_maintainer(file_manager).refresh_calls == []
assert all(CODE_EDIT_OR_LEAVE_IT_LEAVE_TOKEN not in summary for summary in context.editor_summaries)
def _global_context(
file_manager: FileManager,
*,
goal: str,
intent: Intent | None,
relevant_files: list[ClassMeta],
) -> GlobalContext:
"""Build a real GlobalContext over the test's in-memory repository state."""
context = GlobalContext(
goal=goal,
parser=file_manager.parser,
client=file_manager.client,
repo_root=file_manager.repo_root,
src_root=file_manager.src_root,
test_root=file_manager.test_root,
actor=file_manager.actor,
intent_check_factory=IntentCheckFactory(
file_manager.client,
file_manager.prompt_formatter_builder,
),
prompt_formatter_builder=file_manager.prompt_formatter_builder,
)
context._file_manager = file_manager
context.intent = intent
context.relevant_files = list(relevant_files)
return context
def test_begin_recursive_pass_does_not_create_old_blank_seed(memory_root: Path) -> None:
"""Verify pass preparation preserves supplied relevance without a retired seed workflow."""
file_manager = _file_manager(memory_root)
context = _global_context(
file_manager,
goal="goal",
intent=None,
relevant_files=list(file_manager.src_classes),
)
assert not hasattr(GlobalContext, "create_feature_seed")
assert context.begin_recursive_pass() is True
assert context.relevant_files == list(file_manager.src_classes)
assert context.story_cursor.outer_iteration == 1
def test_begin_recursive_pass_allows_feature_bootstrap_without_relevant_files(
memory_root: Path,
) -> None:
"""Verify empty relevance does not prevent new-file feature bootstrap."""
file_manager = _file_manager(memory_root, ())
context = _global_context(
file_manager,
goal="create a new application",
intent=Intent.FEATURE,
relevant_files=[],
)
assert context.begin_recursive_pass() is True
assert context.relevant_files == []
class _BlockedBootstrapSupremeLoop(SupremeLoop):
"""Exercise the blocked-bootstrap branch with real global context ownership."""
def __init__(self, context: GlobalContext) -> None:
super().__init__(
goal=context.goal,
parser=context.parser,
client=context.client,
repo_root=context.repo_root,
src_root=context.src_root,
test_root=context.test_root,
actor=context.actor,
intent_check_factory=context.intent_check_factory,
prompt_formatter_builder=context.prompt_formatter_builder,
context=context,
)
self.blocked_bootstrap_was_handled = False
self.cot_preparation_was_attempted = False
def _run_bootstrap_if_needed(self, *, thaw: bool = False) -> str:
"""Return a concrete blocked result from the owned bootstrap boundary."""
del thaw
self.blocked_bootstrap_was_handled = True
self.context.bootstrap_completed = False
return "Bootstrap mapping stopped: invalid found concern"
def _prepare_cot_after_bootstrap(self) -> bool:
"""Represent an empty post-bootstrap COT plan without running child loops."""
self.cot_preparation_was_attempted = True
return False
def _summarize_recursive_run(self, completed: bool) -> str:
"""Return a stationary final result for the empty COT scenario."""
assert completed is False
return "stabilization pending"
def test_supreme_loop_blocked_bootstrap_continues_to_stabilization(memory_root: Path) -> None:
"""Verify Bootstrap blockage is retained as evidence before COT preparation."""
file_manager = _file_manager(memory_root)
context = _global_context(
file_manager,
goal="goal",
intent=Intent.FEATURE,
relevant_files=list(file_manager.src_classes),
)
context.bootstrap_started = True
loop = _BlockedBootstrapSupremeLoop(context)
assert loop.run() == "stabilization pending"
assert loop.blocked_bootstrap_was_handled is True
assert loop.cot_preparation_was_attempted is True
assert context.story_cursor.run_summaries == [
"Bootstrap mapping stopped: invalid found concern"
]
assert context.supreme_result == "stabilization pending"
def test_intent_prompt_no_longer_mentions_seed_file() -> None:
"""Verify Prompt 0 describes bootstrap mapping rather than blank seed creation."""
client = _Client([{"intent": "new_feature", "new_feature": True, "reason": "new behavior"}])
check = IntentCheckFactory(
client,
MarkdownPromptFormatterBuilder(),
).create("add behavior")
assert check(ClassMeta("src/a.py", "a", ["rel"], {})) is True
assert "seed file" not in client.prompts[0].lower()
assert "bootstrap mapping" in client.prompts[0].lower()