933 lines
36 KiB
Python
933 lines
36 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import pickle
|
|
from pathlib import Path
|
|
from typing import TextIO
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.global_context import GlobalContext, StoryCursor
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.context.traversed_frontier_history import TraversedFrontierHistory
|
|
from band_aider.context.working_backlog import (
|
|
DeferredRemediationHandoff,
|
|
FailedQAItem,
|
|
)
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.file_parser import FileParser
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.intent_check import IntentCheckFactory
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.loop.supreme_loop import SupremeLoop
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
from tests.support.completion_clients import (
|
|
StaticCompletionClient,
|
|
StaticOpenAICompatibleClient,
|
|
)
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _FixtureParser(CodeParser):
|
|
"""Return explicit metadata without consulting a host-backed parser pipeline."""
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return one minimal record for an in-memory fixture location."""
|
|
|
|
del prompt
|
|
return ClassMeta(file_name, "fixture", [], {}, content_type=content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Ignore the stream because this fixture owns metadata directly."""
|
|
|
|
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]:
|
|
"""Return one explicit record for each requested fixture location."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _FixtureFileManager(FileManager):
|
|
"""Bind loop fixture state to an actual FileManager contract in memory."""
|
|
|
|
def __init__(self, sources: list[ClassMeta] | None = None) -> None:
|
|
"""Initialize source metadata and repository access without host IO."""
|
|
|
|
self._pristine_sources = {}
|
|
|
|
self.repo_root = Path("/testbed")
|
|
self.src_root = self.repo_root
|
|
self.test_root = self.repo_root / "tests"
|
|
self.repo_exec = MemoryRepoExec(self.repo_root)
|
|
self.parser = _FixtureParser()
|
|
self.client = StaticOpenAICompatibleClient('{"summary": "fixture"}')
|
|
self.actor = "test"
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
self.prompt = "fixture"
|
|
self.upstream_file_manager = None
|
|
self.src_dirty = {}
|
|
self.test_dirty = {}
|
|
self.src_classes = list(sources or [])
|
|
self.test_classes: list[ClassMeta] = []
|
|
self.src_index = {
|
|
source.location: index
|
|
for index, source in enumerate(self.src_classes)
|
|
}
|
|
self.test_index: dict[str, int] = {}
|
|
self.original_src_classes = list(self.src_classes)
|
|
self.original_test_classes: list[ClassMeta] = []
|
|
self.original_src_index = dict(self.src_index)
|
|
self.original_test_index: dict[str, int] = {}
|
|
self.dependency_graph_maintainer = None
|
|
self.dependency_strategy = None
|
|
self.parser_factory = None
|
|
self._frozen = None
|
|
for source in self.src_classes:
|
|
self.repo_exec.write_text(self.repo_root / source.location, "pass\n")
|
|
|
|
|
|
class _FakeContext(GlobalContext):
|
|
"""Provide the global context contract with only stationary fixture state."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the durable COT cursor and no-host-IO runtime collaborators."""
|
|
|
|
self.original_prompt = "original"
|
|
self.goal = "goal"
|
|
self.story_cursor = StoryCursor()
|
|
self.stories = ["story"]
|
|
self.story_cursor.task_cursor.current_cot_index = 0
|
|
self.relevant_files = [
|
|
ClassMeta("a.py", "a", ["rel"], {}),
|
|
ClassMeta("b.py", "b", ["rel"], {}),
|
|
]
|
|
self.story_cursor.cot_task_records = [
|
|
CotTaskRecord("task one", "a.py", root=self.relevant_files[0]),
|
|
CotTaskRecord("task two", "b.py", root=self.relevant_files[1]),
|
|
]
|
|
self.story_cursor.claimed_locations_this_recursive_pass = set()
|
|
self.story_cursor.task_cursor.visited_locations_by_cot = {}
|
|
self.story_cursor.task_cursor.stabilized_locations_by_cot = {}
|
|
self.story_cursor.task_cursor.active_root = None
|
|
self.story_cursor.task_cursor.active_stabilization_key = None
|
|
self.story_cursor.marginal_result = []
|
|
self.story_cursor._stabilization_started = False
|
|
self.stabilization_started = False
|
|
self.bootstrap_started = False
|
|
self.bootstrap_completed = False
|
|
self.intent = None
|
|
self.supreme_result = ""
|
|
self.supreme_completion_state = self.story_cursor.sovereign_phase
|
|
self._fixture_classes = list(self.relevant_files)
|
|
self._file_manager = None
|
|
self.repo_root = Path("/testbed")
|
|
self.src_root = self.repo_root
|
|
self.test_root = self.repo_root / "tests"
|
|
self.actor = "test"
|
|
self.parser = _FixtureParser()
|
|
self.client = StaticOpenAICompatibleClient('{"summary": "fixture"}')
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
self.intent_check_factory = IntentCheckFactory(
|
|
StaticCompletionClient('{"intent": "existing_files"}'),
|
|
self.prompt_formatter_builder,
|
|
)
|
|
self.search_anchors: tuple[str, ...] = ()
|
|
self.parser_factory = None
|
|
self.logger = logging.getLogger("test.supreme.context")
|
|
self.marginal_result = []
|
|
self.predecessor_root_location: str | None = None
|
|
self.handoff_should_succeed = True
|
|
self.received_handoffs: list[DeferredRemediationHandoff] = []
|
|
self.lifecycle_events: list[str] = []
|
|
self.stabilized_story_roots: list[list[ClassMeta]] = []
|
|
|
|
@property
|
|
def classes(self) -> list[ClassMeta]:
|
|
"""Expose the exact fixture classes requested by stabilization construction."""
|
|
|
|
return self._fixture_classes
|
|
|
|
@classes.setter
|
|
def classes(self, values: list[ClassMeta]) -> None:
|
|
"""Replace the fixture's visible current metadata explicitly."""
|
|
|
|
self._fixture_classes = list(values)
|
|
|
|
@property
|
|
def file_manager(self) -> FileManager:
|
|
"""Return the actual memory-backed manager owned by this fixture."""
|
|
|
|
if self._file_manager is None:
|
|
self._file_manager = _FixtureFileManager(self._fixture_classes)
|
|
return self._file_manager
|
|
|
|
@file_manager.setter
|
|
def file_manager(self, manager: FileManager | None) -> None:
|
|
"""Allow tests to install an explicit nominal in-memory manager."""
|
|
|
|
self._file_manager = manager
|
|
|
|
@property
|
|
def current_cot_task_record(self) -> CotTaskRecord | None:
|
|
"""Mirror the production cursor boundary for an empty planned COT list."""
|
|
|
|
records = self.story_cursor.cot_task_records
|
|
index = self.story_cursor.task_cursor.current_cot_index
|
|
if not records or index >= len(records):
|
|
return None
|
|
return records[index]
|
|
|
|
def current_cot_key(self) -> str:
|
|
return str(self.story_cursor.task_cursor.current_cot_index)
|
|
|
|
def set_current_cot_task(self, index: int) -> None:
|
|
self.story_cursor.task_cursor.current_cot_index = index
|
|
|
|
def begin_recursive_pass(self) -> bool:
|
|
return True
|
|
|
|
def visited_locations_for_current_cot(self) -> set[str]:
|
|
return self.story_cursor.task_cursor.visited_locations_by_cot.setdefault(self.current_cot_key(), set())
|
|
|
|
def stabilized_locations_for_current_cot(self) -> set[str]:
|
|
return self.story_cursor.task_cursor.stabilized_locations_by_cot.setdefault(self.current_cot_key(), set())
|
|
|
|
def previous_cot_root_location_for_current_cot(self) -> str | None:
|
|
"""Return the explicit predecessor selected for this fixture COT."""
|
|
|
|
return self.predecessor_root_location
|
|
|
|
def future_cot_records_for_current_cot(self) -> list[CotTaskRecord]:
|
|
"""Return planner-owned later records without calculating a graph closure in the fake."""
|
|
|
|
index = self.story_cursor.task_cursor.current_cot_index
|
|
return list(self.story_cursor.cot_task_records[index + 1:])
|
|
|
|
def inherited_deferred_remediation_for_current_cot(
|
|
self,
|
|
) -> dict[str, list[FailedQAItem]]:
|
|
"""Return no inherited remediation for this focused root-selection fixture."""
|
|
|
|
return {}
|
|
|
|
def handoff_deferred_remediation_to_next_cot(
|
|
self,
|
|
handoff: DeferredRemediationHandoff,
|
|
) -> bool:
|
|
"""Record the exact parent-owned relay decision for one child result."""
|
|
|
|
self.received_handoffs.append(handoff)
|
|
self.lifecycle_events.append("handoff")
|
|
return self.handoff_should_succeed
|
|
|
|
def mark_root_visited(self, root: ClassMeta) -> None:
|
|
self.visited_locations_for_current_cot().add(root.location)
|
|
self.story_cursor.claimed_locations_this_recursive_pass.add(root.location)
|
|
self.story_cursor.task_cursor.active_root = root
|
|
self.story_cursor.task_cursor.active_stabilization_key = f"{self.current_cot_key()}:{root.location}"
|
|
|
|
def mark_root_stabilized(
|
|
self,
|
|
root: ClassMeta,
|
|
summary: str,
|
|
story_roots: list[ClassMeta] | None = None,
|
|
) -> None:
|
|
self.lifecycle_events.append("stabilized")
|
|
self.stabilized_story_roots.append(list(story_roots or []))
|
|
self.stabilized_locations_for_current_cot().add(root.location)
|
|
self.story_cursor.task_cursor.active_root = None
|
|
self.story_cursor.task_cursor.active_stabilization_key = None
|
|
|
|
def persist(self, keep: int | None = None) -> Path:
|
|
"""Provide the production persistence seam for focused Supreme fixtures."""
|
|
|
|
del keep
|
|
return self.repo_root / "fixture-checkpoint.pkl"
|
|
|
|
|
|
def _failed_remediation(source: ClassMeta) -> FailedQAItem:
|
|
"""Return a nominal deferred source-remediation record for one fixture root."""
|
|
|
|
return FailedQAItem(
|
|
focused_spotlight=source,
|
|
summary="Fixture remediation",
|
|
downstream_classes=[source],
|
|
)
|
|
|
|
|
|
class _StationaryStabilizationLoop(StabilizationLoop):
|
|
"""Represent one controlled child outcome through the nominal loop contract."""
|
|
|
|
def __init__(
|
|
self,
|
|
context: StabilizationContext,
|
|
*,
|
|
accepted_after_run: bool | None = None,
|
|
result: str = "",
|
|
) -> None:
|
|
"""Bind a real stabilization context without starting worker or system work."""
|
|
|
|
self.context = context
|
|
self._accepted_after_run = accepted_after_run
|
|
self._result = result
|
|
self.run_count = 0
|
|
self._logger_stack = []
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Return the explicit child outcome without starting a worker."""
|
|
|
|
del logger_name
|
|
self.run_count += 1
|
|
if self._accepted_after_run is not None:
|
|
self.context.accepted = self._accepted_after_run
|
|
return self._result
|
|
|
|
|
|
class _SupremeHarness(SupremeLoop):
|
|
"""Exercise SupremeLoop ownership with explicit nominal child loops."""
|
|
|
|
def __init__(
|
|
self,
|
|
context: _FakeContext,
|
|
children_by_location: dict[str, _StationaryStabilizationLoop] | None = None,
|
|
*,
|
|
summary: str = "done",
|
|
) -> None:
|
|
"""Build the real parent loop around a memory-backed supplied context."""
|
|
|
|
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.current_cot_completed_check_factory = None
|
|
self._children_by_location = dict(children_by_location or {})
|
|
self._summary = summary
|
|
self.created_roots: list[str] = []
|
|
self.reported_results: list[str] = []
|
|
|
|
def _create_stabilization_loop(self, root: ClassMeta) -> StabilizationLoop:
|
|
"""Return the child explicitly assigned to this COT-owned source root."""
|
|
|
|
self.created_roots.append(root.location)
|
|
return self._children_by_location[root.location]
|
|
|
|
def _summarize_recursive_run(self, completed: bool) -> str:
|
|
"""Expose completion input while keeping the parent result stationary."""
|
|
|
|
del completed
|
|
return self._summary
|
|
|
|
def _post_run(self) -> str:
|
|
"""Return the local incomplete-root summary configured by the test."""
|
|
|
|
return self._summary
|
|
|
|
def _write_change_report(self, result: str) -> Path:
|
|
"""Record report intent without allowing this loop fixture to access disk."""
|
|
|
|
self.reported_results.append(result)
|
|
return self.context.repo_root / "change-report.md"
|
|
|
|
|
|
def _stabilization_context(
|
|
context: _FakeContext,
|
|
root: ClassMeta,
|
|
*,
|
|
accepted: bool,
|
|
remediation: FailedQAItem | None = None,
|
|
surrogate_roots: list[ClassMeta] | None = None,
|
|
frontiers: list[list[ClassMeta]] | None = None,
|
|
) -> StabilizationContext:
|
|
"""Create one actual child context with explicit durable outcome state."""
|
|
|
|
child = StabilizationContext(
|
|
goal="fixture COT",
|
|
classes=context.classes,
|
|
upstream_file_manager=context.file_manager,
|
|
root_change=root,
|
|
cot_index=context.story_cursor.task_cursor.current_cot_index,
|
|
surrogate_roots=list(surrogate_roots or []),
|
|
topology=TopologicalLiveliness(
|
|
frontiers=list(frontiers or []),
|
|
visited={
|
|
source.location
|
|
for frontier in [[root, *(surrogate_roots or [])], *(frontiers or [])]
|
|
for source in frontier
|
|
},
|
|
),
|
|
)
|
|
child.accepted = accepted
|
|
if remediation is not None:
|
|
child.deferred_remediation_backlog[root.location] = [remediation]
|
|
return child
|
|
|
|
|
|
def test_rejected_hydration_reenters_fresh_sync_path() -> None:
|
|
"""An explicitly supplied memory context avoids implicit host-backed hydration."""
|
|
|
|
context = _FakeContext()
|
|
loop = _SupremeHarness(context)
|
|
|
|
assert loop.context is context
|
|
assert isinstance(context.file_manager.repo_exec, MemoryRepoExec)
|
|
assert context.file_manager.repo_exec.preserves_host_environment is False
|
|
|
|
|
|
def test_supreme_loop_processes_each_rooted_cot_task_once() -> None:
|
|
"""Each planned COT root runs exactly once through the owned child loop."""
|
|
|
|
context = _FakeContext()
|
|
first, second = context.relevant_files
|
|
loop = _SupremeHarness(
|
|
context,
|
|
{
|
|
first.location: _StationaryStabilizationLoop(
|
|
_stabilization_context(context, first, accepted=True),
|
|
),
|
|
second.location: _StationaryStabilizationLoop(
|
|
_stabilization_context(context, second, accepted=True),
|
|
),
|
|
},
|
|
)
|
|
|
|
assert loop.run() == "done"
|
|
assert loop.created_roots == ["a.py", "b.py"]
|
|
assert context.story_cursor.claimed_locations_this_recursive_pass == {"a.py", "b.py"}
|
|
|
|
|
|
def test_supreme_loop_ends_batch_when_claimable_files_are_exhausted() -> None:
|
|
"""A completed two-root batch leaves no active root or unfinished COT state."""
|
|
|
|
context = _FakeContext()
|
|
first, second = context.relevant_files
|
|
loop = _SupremeHarness(
|
|
context,
|
|
{
|
|
first.location: _StationaryStabilizationLoop(
|
|
_stabilization_context(context, first, accepted=True),
|
|
),
|
|
second.location: _StationaryStabilizationLoop(
|
|
_stabilization_context(context, second, accepted=True),
|
|
),
|
|
},
|
|
)
|
|
|
|
assert loop.run() == "done"
|
|
assert context.story_cursor.task_cursor.active_root is None
|
|
assert context.stabilized_locations_for_current_cot() == {"b.py"}
|
|
|
|
|
|
def test_incomplete_current_cot_does_not_finish_the_whole_story() -> None:
|
|
"""A local ownership block retains the active COT instead of finishing the story."""
|
|
|
|
context = _FakeContext()
|
|
root = context.relevant_files[0]
|
|
child = _stabilization_context(context, root, accepted=False)
|
|
child.blocked_summary = "current-owned remediation remains"
|
|
loop = _SupremeHarness(
|
|
context,
|
|
{
|
|
root.location: _StationaryStabilizationLoop(
|
|
child,
|
|
accepted_after_run=False,
|
|
result="child incomplete",
|
|
),
|
|
},
|
|
summary="current COT remains incomplete",
|
|
)
|
|
|
|
assert loop.run() == "current COT remains incomplete"
|
|
assert context.supreme_result == ""
|
|
assert context.story_cursor.task_cursor.current_cot_index == 0
|
|
assert context.stabilized_locations_for_current_cot() == set()
|
|
|
|
|
|
def test_supreme_transfers_context_deferred_remediation_before_stabilizing() -> None:
|
|
"""An accepted child hands off its source remediation before root advancement."""
|
|
|
|
context = _FakeContext()
|
|
root = context.relevant_files[0]
|
|
remediation = _failed_remediation(root)
|
|
loop = _SupremeHarness(
|
|
context,
|
|
{
|
|
root.location: _StationaryStabilizationLoop(
|
|
_stabilization_context(
|
|
context,
|
|
root,
|
|
accepted=True,
|
|
remediation=remediation,
|
|
),
|
|
),
|
|
},
|
|
summary="summary",
|
|
)
|
|
|
|
assert loop._stabilize_root(root)
|
|
assert context.lifecycle_events == ["handoff", "stabilized"]
|
|
assert context.received_handoffs == [
|
|
DeferredRemediationHandoff(deferred_items=[remediation]),
|
|
]
|
|
|
|
|
|
def test_supreme_captures_completed_cot_frontiers_as_durable_linked_lineage() -> None:
|
|
"""An accepted COT contributes only its local traversed frontiers to the cursor."""
|
|
|
|
context = _FakeContext()
|
|
loop = _SupremeHarness(context)
|
|
root, downstream = context.relevant_files
|
|
completed = _stabilization_context(
|
|
context,
|
|
root,
|
|
accepted=True,
|
|
frontiers=[[downstream]],
|
|
)
|
|
|
|
loop._capture_traversed_frontier_history(completed)
|
|
|
|
history = context.story_cursor.task_cursor.prior_traversed_frontiers
|
|
assert history is not None
|
|
assert history.cot_index == 0
|
|
assert history.root_location == root.location
|
|
assert history.frontiers == ((root.location,), (downstream.location,))
|
|
assert history.previous is None
|
|
restored = pickle.loads(pickle.dumps(context.story_cursor.task_cursor))
|
|
assert restored.prior_traversed_frontiers == history
|
|
|
|
|
|
def test_supreme_injects_prior_lineage_without_marking_the_new_cot_visited() -> None:
|
|
"""Prior traversal is read-only evidence and cannot suppress current-COT inspection."""
|
|
|
|
context = _FakeContext()
|
|
loop = _SupremeHarness(context)
|
|
root, revisitable = context.relevant_files
|
|
history = TraversedFrontierHistory(
|
|
cot_index=0,
|
|
root_location=revisitable.location,
|
|
frontiers=((revisitable.location,),),
|
|
)
|
|
context.story_cursor.task_cursor.current_cot_index = 1
|
|
context.story_cursor.task_cursor.prior_traversed_frontiers = history
|
|
context.classes = [root, revisitable]
|
|
context.file_manager = _FixtureFileManager([root, revisitable])
|
|
|
|
child = loop._create_stabilization_context(root)
|
|
child.append_frontier([revisitable])
|
|
|
|
assert child.prior_traversed_frontiers is history
|
|
assert revisitable.location not in child.topology.visited
|
|
assert child.traversed_frontier_snapshot() == ((), ())
|
|
|
|
|
|
def test_supreme_retains_an_accepted_root_when_deferred_handoff_fails() -> None:
|
|
"""A failed parent relay keeps the accepted child root active for retry."""
|
|
|
|
context = _FakeContext()
|
|
context.handoff_should_succeed = False
|
|
root = context.relevant_files[0]
|
|
remediation = _failed_remediation(root)
|
|
loop = _SupremeHarness(
|
|
context,
|
|
{
|
|
root.location: _StationaryStabilizationLoop(
|
|
_stabilization_context(
|
|
context,
|
|
root,
|
|
accepted=True,
|
|
remediation=remediation,
|
|
),
|
|
),
|
|
},
|
|
)
|
|
|
|
assert not loop._stabilize_root(root)
|
|
assert context.received_handoffs == [
|
|
DeferredRemediationHandoff(deferred_items=[remediation]),
|
|
]
|
|
assert context.lifecycle_events == ["handoff"]
|
|
assert context.story_cursor.task_cursor.active_root is root
|
|
|
|
|
|
def test_finish_run_reports_nonempty_cot_tasks() -> None:
|
|
"""The report projection preserves each nonblank task without accessing disk."""
|
|
|
|
context = _FakeContext()
|
|
context.story_cursor.cot_task_records.append(
|
|
CotTaskRecord("", "unused.py", ClassMeta("unused.py", "", [], {})),
|
|
)
|
|
loop = _SupremeHarness(context)
|
|
|
|
assert loop._report_cot_tasks() == ["task one", "task two"]
|
|
|
|
|
|
def test_supreme_loop_keeps_cot_records_without_claiming_roots() -> None:
|
|
"""Reading planned COT records does not mutate root-claim bookkeeping."""
|
|
|
|
context = _FakeContext()
|
|
loop = _SupremeHarness(context)
|
|
|
|
assert [record.location for record in context.story_cursor.cot_task_records] == [
|
|
"a.py",
|
|
"b.py",
|
|
]
|
|
assert loop.context.relevant_files == context.relevant_files
|
|
assert context.story_cursor.claimed_locations_this_recursive_pass == set()
|
|
assert context.story_cursor.task_cursor.visited_locations_by_cot == {}
|
|
|
|
|
|
def test_current_workspace_generate_cot_debug() -> None:
|
|
"""The real COT planner orders an in-memory upstream dependency first."""
|
|
|
|
memento_location = "band_aider/context/memento.py"
|
|
class_meta_location = "band_aider/context/class_meta.py"
|
|
memento_task = (
|
|
"Implement the Memento unit-of-work lifecycle in the Memento class. "
|
|
"Implement __enter__ to initialize private, per-instance lifecycle state, "
|
|
"activate exactly one layer, and capture current source bytes for the selected "
|
|
"layer if a snapshot does not exist. Implement __exit__ to guarantee cleanup: "
|
|
"if an unresolved Memento exits its with block, it must automatically roll back "
|
|
"without suppressing exceptions; if already committed or rolled back, __exit__ "
|
|
"must not resolve it again. Refactor commit and rollback to require an active, "
|
|
"unresolved Memento."
|
|
)
|
|
class_meta_task = (
|
|
"Refactor the method ClassMeta.set_or_get_memento to remove immediate activation "
|
|
"of Memento instances by __enter__. Instead, have set_or_get_memento return "
|
|
"unentered Memento objects constructed by calling Memento.first_available, "
|
|
"Memento.current, or the Memento constructor with source_path and repo_exec "
|
|
"parameters. The method must resolve source_path using ClassMeta.resolved_path "
|
|
"and use file_manager.repo_exec as the repo_exec argument. This confines "
|
|
"lifecycle management (activation via __enter__) to the caller, ensuring "
|
|
"lifecycle compliance and removing the pattern of constructing and immediately "
|
|
"entering Mementos."
|
|
)
|
|
source_locations = [class_meta_location, memento_location]
|
|
|
|
class _CotClient(OpenAICompatibleClient):
|
|
"""Return stationary broad and fixed-location refinement COT responses."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
del logger
|
|
if "BEGIN DEPENDENCY REFERENCE SELECTION" in prompt:
|
|
return '{"references": []}'
|
|
if "FILE-OWNED CONTRACT" in prompt:
|
|
if f"FIXED LOCATION:\n{memento_location}" in prompt:
|
|
return json.dumps({
|
|
"task": memento_task,
|
|
"location": memento_location,
|
|
"symbols": ["Memento"],
|
|
})
|
|
return json.dumps({
|
|
"task": class_meta_task,
|
|
"location": class_meta_location,
|
|
"symbols": ["ClassMeta", "ClassMeta.set_or_get_memento"],
|
|
})
|
|
return json.dumps({"cot_tasks": [
|
|
{"task": "Plan ClassMeta.set_or_get_memento", "location": class_meta_location},
|
|
{"task": "Plan Memento lifecycle", "location": memento_location},
|
|
]})
|
|
|
|
repo_root = Path("/testbed")
|
|
repo_exec = MemoryRepoExec(repo_root)
|
|
repo_exec.write_text(
|
|
repo_root / memento_location,
|
|
"class Memento:\n"
|
|
" def __enter__(self) -> Memento:\n"
|
|
" return self\n\n"
|
|
" def __exit__(self, *arguments: object) -> None:\n"
|
|
" pass\n\n"
|
|
" def commit(self) -> None:\n"
|
|
" pass\n\n"
|
|
" def rollback(self) -> None:\n"
|
|
" pass\n",
|
|
)
|
|
repo_exec.write_text(
|
|
repo_root / class_meta_location,
|
|
"from band_aider.context.memento import Memento\n\n"
|
|
"class ClassMeta:\n"
|
|
" def set_or_get_memento(self) -> Memento:\n"
|
|
" return Memento()\n",
|
|
)
|
|
parser_client = StaticOpenAICompatibleClient(json.dumps({
|
|
"location": "",
|
|
"summary": "source",
|
|
"relevance": ["source"],
|
|
"irrelevance": [],
|
|
"relevant_symbols": {"Memento": "Lifecycle owner", "ClassMeta": "Caller"},
|
|
}))
|
|
file_manager = FileManager(
|
|
parser=FileParser(
|
|
parser_client,
|
|
MarkdownPromptFormatterBuilder(),
|
|
repo_exec,
|
|
),
|
|
client=_CotClient(),
|
|
repo_root=repo_root,
|
|
src_root=repo_root,
|
|
test_root=repo_root / "tests",
|
|
actor="regression-test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
repo_exec=repo_exec,
|
|
)
|
|
candidates = [file_manager.current_src(location) for location in source_locations]
|
|
candidates = [candidate for candidate in candidates if candidate is not None]
|
|
assert len(candidates) == 2
|
|
|
|
strategy = file_manager.dependency_strategy
|
|
assert strategy is not None
|
|
upstream = strategy.upstream_sources_for(
|
|
[candidates[0]],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
file_manager.test_root,
|
|
recursive=True,
|
|
)
|
|
upstream_locations = [source.location for source in upstream]
|
|
assert upstream_locations == [memento_location]
|
|
|
|
context = _FakeContext()
|
|
context.client = _CotClient()
|
|
context.goal = "Implement the requested lifecycle refactor"
|
|
context.story_cursor = StoryCursor()
|
|
context.relevant_files = candidates
|
|
context.file_manager = file_manager
|
|
context.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
context.generate_cot_tasks()
|
|
generated_records = list(context.story_cursor.cot_task_records)
|
|
refined_records = list(generated_records)
|
|
ordered_records = list(context.story_cursor.cot_task_records)
|
|
ordered_locations = [record.location for record in ordered_records]
|
|
|
|
assert ordered_locations == [memento_location, class_meta_location]
|
|
|
|
def test_refresh_cot_tasks_renders_dependency_order() -> None:
|
|
"""Refreshing COT work preserves planner order at the parent-loop boundary."""
|
|
|
|
memento_location = "band_aider/context/memento.py"
|
|
class_meta_location = "band_aider/context/class_meta.py"
|
|
class_meta_task = "Refactor the method ClassMeta.set_or_get_memento to remove immediate activation of Memento instances by __enter__. Instead, have set_or_get_memento return unentered Memento objects constructed by calling Memento.first_available, Memento.current, or the Memento constructor with source_path and repo_exec parameters. The method must resolve source_path using ClassMeta.resolved_path and use file_manager.repo_exec as the repo_exec argument. This confines lifecycle management (activation via __enter__) to the caller, ensuring lifecycle compliance and removing the pattern of constructing and immediately entering Mementos."
|
|
memento_task = "Implement the Memento unit-of-work lifecycle in the Memento class. Implement __enter__ to initialize private, per-instance lifecycle state, activate exactly one layer, and capture current source bytes for the selected layer if a snapshot does not exist. Implement __exit__ to guarantee cleanup: if an unresolved Memento exits its with block, it must automatically roll back without suppressing exceptions; if already committed or rolled back, __exit__ must not resolve it again. Refactor commit and rollback to require an active, unresolved Memento."
|
|
|
|
class _Client(OpenAICompatibleClient):
|
|
"""Return fixed COT records by the exact stationary prompt category."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
del logger
|
|
if "BEGIN DEPENDENCY REFERENCE SELECTION" in prompt:
|
|
return '{"references": []}'
|
|
if "FILE-OWNED CONTRACT" in prompt:
|
|
if f"FIXED LOCATION:\n{memento_location}" in prompt:
|
|
return json.dumps({
|
|
"task": memento_task,
|
|
"location": memento_location,
|
|
"symbols": ["Memento"],
|
|
})
|
|
return json.dumps({
|
|
"task": class_meta_task,
|
|
"location": class_meta_location,
|
|
"symbols": ["ClassMeta", "ClassMeta.set_or_get_memento"],
|
|
})
|
|
return json.dumps({"cot_tasks": [
|
|
{"task": "Update ClassMeta lifecycle callers.", "location": class_meta_location},
|
|
{"task": "Update Memento lifecycle.", "location": memento_location},
|
|
]})
|
|
|
|
root = Path("/testbed")
|
|
repo_exec = MemoryRepoExec(root)
|
|
repo_exec.write_text(
|
|
root / memento_location,
|
|
"class Memento:\n"
|
|
" def __enter__(self) -> Memento:\n"
|
|
" return self\n\n"
|
|
" def __exit__(self, *arguments: object) -> None:\n"
|
|
" pass\n\n"
|
|
" def commit(self) -> None:\n"
|
|
" pass\n\n"
|
|
" def rollback(self) -> None:\n"
|
|
" pass\n",
|
|
)
|
|
repo_exec.write_text(
|
|
root / class_meta_location,
|
|
"from band_aider.context.memento import Memento\n\n"
|
|
"class ClassMeta:\n"
|
|
" def set_or_get_memento(self) -> Memento:\n"
|
|
" return Memento()\n",
|
|
)
|
|
parser_client = StaticOpenAICompatibleClient(json.dumps({
|
|
"location": "",
|
|
"summary": "source",
|
|
"relevance": ["source"],
|
|
"irrelevance": [],
|
|
"relevant_symbols": {"set_or_get_memento": "Caller", "Memento": "Lifecycle owner"},
|
|
}))
|
|
manager = FileManager(
|
|
parser=FileParser(
|
|
parser_client,
|
|
MarkdownPromptFormatterBuilder(),
|
|
repo_exec,
|
|
),
|
|
client=_Client(),
|
|
repo_root=root,
|
|
src_root=root,
|
|
test_root=root / "tests",
|
|
actor="test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
repo_exec=repo_exec,
|
|
)
|
|
candidates = [manager.current_src(memento_location), manager.current_src(class_meta_location)]
|
|
candidates = [item for item in candidates if item is not None]
|
|
assert len(candidates) == 2
|
|
graph = manager.dependency_strategy
|
|
assert graph is not None
|
|
upstream = [item.location for item in graph.upstream_sources_for(
|
|
[candidates[1]],
|
|
manager.src_classes,
|
|
manager.src_root,
|
|
manager.test_root,
|
|
recursive=True,
|
|
)]
|
|
assert upstream == [memento_location]
|
|
|
|
context = _FakeContext()
|
|
context.goal = "goal"
|
|
context.client = _Client()
|
|
context.file_manager = manager
|
|
context.logger = logging.getLogger("cot-order-regression")
|
|
context.relevant_files = candidates
|
|
context.story_cursor.cot_task_records = []
|
|
context.story_cursor.outer_iteration = 0
|
|
loop = _SupremeHarness(context)
|
|
assert loop._prepare_cot_after_bootstrap()
|
|
first_active_record = context.current_cot_task_record
|
|
rendered_first_screen = [record.task for record in context.story_cursor.cot_task_records]
|
|
context.story_cursor.task_cursor.current_cot_index = 1
|
|
second_active_record = context.current_cot_task_record
|
|
rendered_second_screen = [record.task for record in context.story_cursor.cot_task_records]
|
|
|
|
assert first_active_record is not None
|
|
assert second_active_record is not None
|
|
assert first_active_record.location == memento_location
|
|
assert second_active_record.location == class_meta_location
|
|
assert rendered_first_screen[0].startswith(memento_task)
|
|
assert rendered_second_screen[1].startswith(class_meta_task)
|
|
|
|
|
|
def test_supreme_loop_bootstrap_involved_root_seeds_surrogate_siblings() -> None:
|
|
"""Verify first bootstrap-involved stabilization story receives sibling surrogate roots."""
|
|
|
|
context = _FakeContext()
|
|
a, b = context.relevant_files
|
|
context.classes = [a, b]
|
|
context.story_cursor.bootstrap_involved_locations = {"a.py", "b.py", "missing.py"}
|
|
context.file_manager = _FixtureFileManager([a, b])
|
|
loop = _SupremeHarness(context)
|
|
|
|
stabilization_context = loop._create_stabilization_context(a)
|
|
|
|
assert stabilization_context.root_change == a
|
|
assert stabilization_context.surrogate_roots == [b]
|
|
assert not hasattr(stabilization_context, "reserved_future_cot_locations")
|
|
assert stabilization_context.inherited_remediation_backlog == {}
|
|
assert not hasattr(stabilization_context, "deferred_remediation_handoff_callback")
|
|
|
|
|
|
def test_supreme_loop_rederives_predecessor_readonly_file_when_creating_child() -> None:
|
|
"""A child receives predecessor evidence from the parent COT ledger."""
|
|
|
|
context = _FakeContext()
|
|
root = context.relevant_files[1]
|
|
context.predecessor_root_location = "a.py"
|
|
context.file_manager = _FixtureFileManager(context.relevant_files)
|
|
context.logger = logging.getLogger("test.supreme.resume")
|
|
loop = _SupremeHarness(context)
|
|
|
|
child = SupremeLoop._create_stabilization_loop(loop, root)
|
|
|
|
assert child.previous_cot_root_readonly_file == Path("/testbed/a.py")
|
|
assert child.provenance is loop.provenance
|
|
assert child.future_cot_records == [
|
|
context.story_cursor.cot_task_records[1],
|
|
]
|
|
|
|
|
|
def test_supreme_loop_bootstrap_surrogates_only_on_first_involved_story() -> None:
|
|
"""Verify later or unrelated stabilization roots do not receive bootstrap surrogates."""
|
|
|
|
context = _FakeContext()
|
|
a, b = context.relevant_files
|
|
context.story_cursor.bootstrap_involved_locations = {"a.py", "b.py"}
|
|
context.file_manager = _FixtureFileManager([a, b])
|
|
loop = _SupremeHarness(context)
|
|
|
|
context.story_cursor.claimed_locations_this_recursive_pass = {"already.py"}
|
|
assert loop._bootstrap_surrogate_roots_for(a) == []
|
|
|
|
context.story_cursor.claimed_locations_this_recursive_pass = set()
|
|
unrelated = ClassMeta("c.py", "c", ["rel"], {})
|
|
assert loop._bootstrap_surrogate_roots_for(unrelated) == []
|
|
|
|
|
|
def test_supreme_loop_marks_only_root_topology_after_surrogate_stabilization() -> None:
|
|
"""Verify downstream frontiers are not treated as bootstrap surrogate roots."""
|
|
|
|
context = _FakeContext()
|
|
a, b = context.relevant_files
|
|
downstream = ClassMeta("c.py", "c", ["rel"], {})
|
|
child = _stabilization_context(
|
|
context,
|
|
a,
|
|
accepted=True,
|
|
surrogate_roots=[b],
|
|
frontiers=[[downstream]],
|
|
)
|
|
loop = _SupremeHarness(
|
|
context,
|
|
{a.location: _StationaryStabilizationLoop(child)},
|
|
summary="summary",
|
|
)
|
|
|
|
loop._stabilize_root(a)
|
|
|
|
assert [source.location for source in context.stabilized_story_roots[0]] == [
|
|
"a.py",
|
|
"b.py",
|
|
]
|