3393 lines
119 KiB
Python
3393 lines
119 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 collections.abc import Callable
|
|
from itertools import chain
|
|
from unittest.mock import Mock
|
|
import pytest
|
|
|
|
from band_aider.context.context_factory import (
|
|
StabilizationContextReassembly,
|
|
)
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.provenance import Phase, Provenance
|
|
from band_aider.context.work_completion import WorkCompletion
|
|
from band_aider.context.working_backlog import (
|
|
BackLog as DurableBackLog,
|
|
BacklogLane,
|
|
StagedInboundItem,
|
|
)
|
|
from band_aider.context.resume_state import configure_resume_settings, resume_settings
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.loop.stabilization_dtos import FrontierBoundary
|
|
from band_aider.loop.mechanical_loop_gate import PredicateCheck
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import (
|
|
BiDependencyDiscoveryStrategy,
|
|
)
|
|
from band_aider.judgement.code_dependency_graph_maintainer import (
|
|
CodeDependencyGraphMaintainer,
|
|
)
|
|
from band_aider.judgement.code_editor import CodeEditor
|
|
from band_aider.judgement.code_dependency_graph import (
|
|
CodeDependencyGraph,
|
|
DependencySelector,
|
|
)
|
|
from band_aider.judgement.dependency_discovery_strategy import (
|
|
DependencyDirection,
|
|
DependencyTraversalCache,
|
|
)
|
|
from band_aider.judgement.python_ast_dependency_strategy import (
|
|
PythonAstDependencyStrategy,
|
|
)
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.test_runner_strategy import TestRunnerStrategy
|
|
from band_aider.strategies import LanguageStrategySet
|
|
from band_aider.strategies import language_strategy_set_factory
|
|
|
|
from tests.band_aider_loop_impl import *
|
|
from tests.topology_test_support import boundary_for, service_topology
|
|
|
|
|
|
def _loop_file_manager(loop: StabilizationLoop) -> FileManager:
|
|
"""Return the runtime manager after the stabilization constructor has bound it."""
|
|
|
|
file_manager = loop.context.file_manager
|
|
if file_manager is None:
|
|
raise AssertionError("The stabilization scenario did not bind its FileManager.")
|
|
return file_manager
|
|
|
|
|
|
def _scenario_file_manager(loop: StabilizationLoop) -> TestFileManager:
|
|
"""Return the explicit in-memory FileManager used by a pure scenario."""
|
|
|
|
file_manager = _loop_file_manager(loop)
|
|
if not isinstance(file_manager, TestFileManager):
|
|
raise AssertionError("The scenario did not use the in-memory FileManager.")
|
|
return file_manager
|
|
|
|
|
|
def _loop_backlog(loop: StabilizationLoop) -> DurableBackLog:
|
|
"""Return the durable backlog that every constructed stabilization scenario owns."""
|
|
|
|
backlog = loop.context.backlog
|
|
if backlog is None:
|
|
raise AssertionError("The stabilization scenario did not bind its BackLog.")
|
|
return backlog
|
|
|
|
|
|
def _context_backlog(context: StabilizationContext) -> DurableBackLog:
|
|
"""Return the durable backlog that a constructed stabilization context owns."""
|
|
|
|
backlog = context.backlog
|
|
if backlog is None:
|
|
raise AssertionError("The stabilization context did not initialize its BackLog.")
|
|
return backlog
|
|
|
|
|
|
def _source_metadata(file_manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return required source metadata from a scenario's fully synchronized manager."""
|
|
|
|
source = file_manager.current_src(location)
|
|
if source is None:
|
|
raise AssertionError(f"The scenario did not synchronize source metadata for {location}.")
|
|
return source
|
|
|
|
|
|
def _test_metadata(file_manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return required test metadata from a scenario's fully synchronized manager."""
|
|
|
|
test = file_manager.current_test(location)
|
|
if test is None:
|
|
raise AssertionError(f"The scenario did not synchronize test metadata for {location}.")
|
|
return test
|
|
|
|
|
|
def _focused_metadata(
|
|
item: FailedQAItem | InspectionBacklog | TestingBacklog,
|
|
) -> ClassMeta:
|
|
"""Return required focus metadata from one populated backlog test scenario."""
|
|
|
|
focus = item.focused_spotlight
|
|
if focus is None:
|
|
raise AssertionError("The scenario expected every backlog item to have a focus.")
|
|
return focus
|
|
|
|
|
|
class RuntimeStabilizationContext(StabilizationContext):
|
|
"""Stabilization context exposing only its runtime-only checkpoint collaborators."""
|
|
|
|
new_files_callback: Callable[..., bool]
|
|
topology_summary: str
|
|
test_runner_strategy: TestRunnerStrategy
|
|
|
|
|
|
class _ScenarioAlphaLoop(AlphaLoop):
|
|
"""A real AlphaLoop whose source pass is deliberately controlled by one scenario."""
|
|
|
|
def __init__(
|
|
self,
|
|
template: AlphaLoop,
|
|
bfs_depth: int,
|
|
source_runner: Callable[[int, ContributorBacklog], str],
|
|
) -> None:
|
|
"""Retain the production-created Alpha context and gate for one controlled pass."""
|
|
|
|
super().__init__(
|
|
gate=template.gate,
|
|
context=template.context,
|
|
aider_worker=template.aider_worker,
|
|
frontier_provider=template.frontier_provider,
|
|
previous_cot_root_readonly_file=template.previous_cot_root_readonly_file,
|
|
logger=template.logger,
|
|
)
|
|
self._bfs_depth = bfs_depth
|
|
self._source_runner = source_runner
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Execute the scenario-owned source outcome through a nominal AlphaLoop."""
|
|
|
|
del logger_name
|
|
return self._source_runner(self._bfs_depth, self.context.backlog)
|
|
|
|
|
|
class ScenarioStabilizationLoop(StabilizationLoop):
|
|
"""StabilizationLoop with explicit nominal child outcomes for scheduler keyframes."""
|
|
|
|
source_runner: Callable[[int, ContributorBacklog], str] | None
|
|
testing_runner: Callable[[], None] | None
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Initialize optional scenario outcomes before production loop setup."""
|
|
|
|
self.source_runner = None
|
|
self.testing_runner = None
|
|
super().__post_init__()
|
|
|
|
def set_source_runner(
|
|
self,
|
|
source_runner: Callable[[int, ContributorBacklog], str],
|
|
) -> None:
|
|
"""Install the one declared Alpha outcome needed by this scenario."""
|
|
|
|
self.source_runner = source_runner
|
|
|
|
def set_testing_runner(self, testing_runner: Callable[[], None]) -> None:
|
|
"""Install the one declared Beta outcome needed by this scenario."""
|
|
|
|
self.testing_runner = testing_runner
|
|
|
|
def _create_alpha_loop(self, bfs_depth: int) -> AlphaLoop:
|
|
"""Return a production AlphaLoop or the scenario's nominal controlled child."""
|
|
|
|
template = super()._create_alpha_loop(bfs_depth)
|
|
if self.source_runner is None:
|
|
return template
|
|
return _ScenarioAlphaLoop(template, bfs_depth, self.source_runner)
|
|
|
|
def _run_testing_pass(self) -> None:
|
|
"""Run the declared Beta outcome, otherwise retain production behavior."""
|
|
|
|
if self.testing_runner is None:
|
|
super()._run_testing_pass()
|
|
return
|
|
self.testing_runner()
|
|
|
|
|
|
def _scenario_stabilization_loop(
|
|
file_manager: TestFileManager,
|
|
backlog: DurableBackLog,
|
|
coverage: FakeCoverageStrategy,
|
|
*,
|
|
worker: CodeEditor | None = None,
|
|
cot_index: int = 0,
|
|
total_cot: int = 2,
|
|
) -> ScenarioStabilizationLoop:
|
|
"""Build a nominal stabilization scenario with controllable child outcomes."""
|
|
|
|
root = file_manager.require_src("src/foo.py")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
cot_index=cot_index,
|
|
total_cot=total_cot,
|
|
completion=backlog.completion,
|
|
backlog=backlog,
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
return ScenarioStabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=coverage,
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
aider_worker=worker,
|
|
)
|
|
|
|
|
|
def test_frontier_zero_seeds_share_the_selected_provenance_parent(tmp_path: Path) -> None:
|
|
"""Root admission links all seeds before queueing, and replay preserves their identities."""
|
|
|
|
manager = _file_manager(tmp_path)
|
|
manager.repo_exec.write_text(manager.src_root / "surrogate.py", "value = 1\n")
|
|
manager.resync()
|
|
backlog = DurableBackLog(completion={})
|
|
loop = _scenario_stabilization_loop(manager, backlog, FakeCoverageStrategy(1.0), cot_index=1)
|
|
surrogate = manager.require_src("src/surrogate.py")
|
|
loop.context.surrogate_roots = [surrogate]
|
|
graph = loop.provenance
|
|
assert graph is not None
|
|
alpha = graph.start(Phase.ALPHA, InspectionBacklog(manager.require_src("src/foo.py"), "Earlier COT."))
|
|
beta = graph.track_provenance(alpha, Phase.BETA, TestingBacklog(manager.require_test("tests/test_foo.py"), "Earlier tests."))
|
|
strategy = Mock(spec=BiDependencyDiscoveryStrategy)
|
|
strategy.has_path.side_effect = lambda start, target: start == "src/surrogate.py" and target == "src/foo.py"
|
|
strategy.upstream_sources_for.return_value = []
|
|
loop.bi_dependency_strategy = strategy
|
|
|
|
loop._enqueue_current_frontier_to_backlog()
|
|
|
|
seeds = list(backlog.inspection)
|
|
assert len(seeds) == 2
|
|
ids = tuple(item.message_id for item in seeds)
|
|
assert all(graph.messages[message_id].parent_id == beta for message_id in ids)
|
|
assert set(map(lambda item: _focused_metadata(item).location, seeds)) == {"src/foo.py", "src/surrogate.py"}
|
|
messages = tuple(graph.messages)
|
|
loop._enqueue_current_frontier_to_backlog()
|
|
assert tuple(graph.messages) == messages
|
|
assert tuple(item.message_id for item in backlog.inspection) == ids
|
|
|
|
|
|
def _failed_item(
|
|
source: ClassMeta,
|
|
test: ClassMeta,
|
|
*,
|
|
summary: str = "repair source",
|
|
upstream_sources: list[ClassMeta] | None = None,
|
|
) -> FailedQAItem:
|
|
"""Create one topology-shaped remediation item."""
|
|
|
|
return FailedQAItem(
|
|
focused_spotlight=source,
|
|
summary=summary,
|
|
upstream_classes=list(upstream_sources or []),
|
|
downstream_classes=[source],
|
|
failed_test=FailingTest(
|
|
test=test,
|
|
stack_trace=["failed"],
|
|
error_message="failed",
|
|
),
|
|
)
|
|
|
|
|
|
def _freeze_dependency_edges(
|
|
file_manager: FileManager,
|
|
edges: list[tuple[str, str]],
|
|
) -> None:
|
|
"""Install a deterministic frozen graph for forward-remediation examples."""
|
|
|
|
dependencies: dict[str, set[str]] = {}
|
|
dependents: dict[str, set[str]] = {}
|
|
for upstream, downstream in edges:
|
|
dependents.setdefault(upstream, set()).add(downstream)
|
|
dependencies.setdefault(downstream, set()).add(upstream)
|
|
graph = CodeDependencyGraph(
|
|
source_by_location={source.location: source for source in file_manager.src_classes},
|
|
source_dependencies=dependencies,
|
|
source_dependents=dependents,
|
|
)
|
|
graph.freeze()
|
|
file_manager._frozen = graph
|
|
|
|
|
|
class _PerimeterTopologyStrategy(PythonAstDependencyStrategy):
|
|
"""Provide a small graph for deferred-topology perimeter keyframes.
|
|
|
|
Example: ``_PerimeterTopologyStrategy(sources, [(root, child)])`` exposes
|
|
the direct indexed edge used by the cadence tests.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
sources_by_location: dict[str, ClassMeta],
|
|
edges: list[tuple[str, str]],
|
|
) -> None:
|
|
"""Store direct graph edges and record bounded upstream path checks."""
|
|
|
|
super().__init__()
|
|
self.sources_by_location = sources_by_location
|
|
self.dependencies: dict[str, set[str]] = {}
|
|
self.dependents: dict[str, set[str]] = {}
|
|
for upstream, downstream in edges:
|
|
self.dependencies.setdefault(downstream, set()).add(upstream)
|
|
self.dependents.setdefault(upstream, set()).add(downstream)
|
|
self.downstream_seed_locations: list[str] = []
|
|
self.downstream_calls: list[tuple[list[str], bool]] = []
|
|
self.has_path_calls: list[tuple[str, str, frozenset[str]]] = []
|
|
|
|
def source_dependencies_of(
|
|
self,
|
|
source_location: str,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> set[str]:
|
|
"""Return direct prerequisites from the synthetic graph index."""
|
|
|
|
del selector
|
|
return set(self.dependencies.get(source_location, set()))
|
|
|
|
def source_dependents_of(
|
|
self,
|
|
source_location: str,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> set[str]:
|
|
"""Return direct dependents from the synthetic graph index."""
|
|
|
|
del selector
|
|
return set(self.dependents.get(source_location, set()))
|
|
|
|
def direct_tests_for(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
all_test_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
test_root: Path | None = None,
|
|
) -> list[ClassMeta]:
|
|
"""Return no tests because these examples isolate topology admission."""
|
|
|
|
del source_files, all_test_files, src_root, test_root
|
|
return []
|
|
|
|
def downstream_sources_for(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
recursive: bool = False,
|
|
dependency_cache: DependencyTraversalCache | None = None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
"""Return deterministic direct or recursive downstream source metadata."""
|
|
|
|
del all_source_files, src_root, dependency_cache, selector
|
|
self.downstream_seed_locations = sorted(
|
|
source.location
|
|
for source in source_files
|
|
)
|
|
self.downstream_calls.append((self.downstream_seed_locations, recursive))
|
|
locations = {
|
|
dependent
|
|
for source in source_files
|
|
for dependent in self.dependents.get(source.location, set())
|
|
}
|
|
if recursive:
|
|
queue = sorted(locations)
|
|
while queue:
|
|
location = queue.pop(0)
|
|
for dependent in sorted(self.dependents.get(location, set())):
|
|
if dependent in locations:
|
|
continue
|
|
locations.add(dependent)
|
|
queue.append(dependent)
|
|
return list(map(
|
|
lambda location: self.sources_by_location[location],
|
|
filter(
|
|
lambda location: location in self.sources_by_location,
|
|
sorted(locations),
|
|
),
|
|
))
|
|
|
|
def upstream_sources_for(
|
|
self,
|
|
files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
test_root: Path | None = None,
|
|
recursive: bool = False,
|
|
dependency_cache: DependencyTraversalCache | None = None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
"""Return direct upstream requirements for the supplied candidates."""
|
|
|
|
del all_source_files, src_root, test_root, recursive, dependency_cache, selector
|
|
locations = set(chain.from_iterable(map(
|
|
lambda source: self.dependencies.get(source.location, set()),
|
|
files,
|
|
)))
|
|
return list(map(
|
|
lambda location: self.sources_by_location[location],
|
|
filter(
|
|
lambda location: location in self.sources_by_location,
|
|
sorted(locations),
|
|
),
|
|
))
|
|
|
|
def has_path(
|
|
self,
|
|
start: str,
|
|
target: str,
|
|
direction: DependencyDirection = DependencyDirection.UPSTREAM,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
excluded_locations: frozenset[str] = frozenset(),
|
|
) -> bool:
|
|
"""Mirror the production path contract while recording bounded lookback queries."""
|
|
|
|
del selector
|
|
self.has_path_calls.append((start, target, excluded_locations))
|
|
if start in excluded_locations:
|
|
return False
|
|
if start == target:
|
|
return True
|
|
neighbors = (
|
|
self.dependencies
|
|
if direction is DependencyDirection.UPSTREAM
|
|
else self.dependents
|
|
)
|
|
queue = [start]
|
|
seen = {start}
|
|
while queue:
|
|
location = queue.pop(0)
|
|
for neighbor in sorted(neighbors.get(location, set())):
|
|
if neighbor == target:
|
|
return True
|
|
if neighbor in excluded_locations or neighbor in seen:
|
|
continue
|
|
seen.add(neighbor)
|
|
queue.append(neighbor)
|
|
return False
|
|
|
|
def joint_beam_sources_for(
|
|
self,
|
|
current_frontier: list[ClassMeta],
|
|
deferred_blames: list[ClassMeta],
|
|
all_source_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
) -> list[ClassMeta]:
|
|
"""Return no remediation whitelist evidence for topology-only examples."""
|
|
|
|
del current_frontier, deferred_blames, all_source_files, src_root
|
|
return []
|
|
|
|
|
|
def _boundary(
|
|
sources: dict[str, ClassMeta],
|
|
*current_locations: str,
|
|
unchanged_locations: tuple[str, ...] = (),
|
|
) -> FrontierBoundary:
|
|
"""Build one raw-BFS boundary with the scenario's unchanged-source algebra."""
|
|
|
|
current = frozenset(current_locations)
|
|
unchanged = frozenset(unchanged_locations)
|
|
return FrontierBoundary(
|
|
current_locations=current,
|
|
unchanged_locations=unchanged,
|
|
live_seed_locations=current - unchanged,
|
|
source_by_location=sources,
|
|
)
|
|
|
|
|
|
def _drain_topology_source_work(
|
|
_: int,
|
|
source_backlog: ContributorBacklog,
|
|
) -> str:
|
|
"""Commit all admitted source work without simulating an editor response."""
|
|
|
|
claimed = source_backlog.pop_next_source_work()
|
|
while claimed is not None:
|
|
with claimed as scope:
|
|
scope.commit()
|
|
claimed = source_backlog.pop_next_source_work()
|
|
return "drained topology work"
|
|
|
|
|
|
def _perimeter_topology_loop(
|
|
tmp_path: Path,
|
|
names: list[str],
|
|
edges: list[tuple[str, str]],
|
|
*,
|
|
frozen_edges: list[tuple[str, str]] | None = None,
|
|
future_cot_records: list[CotTaskRecord] | None = None,
|
|
logger: logging.Logger | None = None,
|
|
) -> tuple[StabilizationLoop, dict[str, ClassMeta], _PerimeterTopologyStrategy]:
|
|
"""Build one frozen-COT synthetic graph with a separately controlled live strategy."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
for name in names:
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.src_root / name,
|
|
f"def {Path(name).stem}():\n return 1\n"
|
|
)
|
|
file_manager.resync()
|
|
sources = {
|
|
source.location: source
|
|
for source in file_manager.src_classes
|
|
}
|
|
_freeze_dependency_edges(file_manager, frozen_edges or edges)
|
|
root = sources["src/foo.py"]
|
|
strategy = _PerimeterTopologyStrategy(sources, edges)
|
|
context = StabilizationContext(
|
|
goal="perimeter topology",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
language_strategy_set = _language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
)
|
|
language_strategy_set.bindings[BiDependencyDiscoveryStrategy] = strategy
|
|
loop = ScenarioStabilizationLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: True),
|
|
context=context,
|
|
language_strategy_set=language_strategy_set,
|
|
future_cot_records=future_cot_records,
|
|
logger=logger,
|
|
)
|
|
loop.set_source_runner(_drain_topology_source_work)
|
|
return loop, sources, strategy
|
|
|
|
|
|
def _stage_unchanged_perimeter_join(
|
|
loop: StabilizationLoop,
|
|
sources: dict[str, ClassMeta],
|
|
) -> tuple[ClassMeta, dict[str, str]]:
|
|
"""Place X after Alpha no-change handling while retaining the root whitelist seed."""
|
|
|
|
root = sources["src/foo.py"]
|
|
perimeter = sources["src/x.py"]
|
|
loop.context.append_frontier([perimeter])
|
|
loop.context.topology.visited.update({
|
|
root.location,
|
|
perimeter.location,
|
|
})
|
|
loop.context.topology.whitelist.add(root.location)
|
|
return perimeter, loop._current_frontier_hashes()
|
|
|
|
|
|
def test_topological_liveliness_enqueues_bfs_slice_then_dequeues_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The context-owned scheduler advances only through its explicit dequeue/enqueue keyframes."""
|
|
|
|
root_location = "src/foo.py"
|
|
child_location = "src/x.py"
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py"],
|
|
[(root_location, child_location)],
|
|
)
|
|
topology = loop.context.topology
|
|
topology.visited.add(root_location)
|
|
|
|
topology.enqueue_bfs_slice(
|
|
[sources[child_location]],
|
|
FrontierBoundary(
|
|
current_locations=frozenset({root_location}),
|
|
unchanged_locations=frozenset(),
|
|
live_seed_locations=frozenset({root_location}),
|
|
source_by_location=sources,
|
|
),
|
|
)
|
|
next_frontier = topology.dequeue_topologically()
|
|
|
|
assert [source.location for source in next_frontier] == [child_location]
|
|
assert topology.bfs_depth == 0
|
|
assert not topology.has_deferred_frontier
|
|
|
|
topology.append_frontier(next_frontier)
|
|
frontier, bfs_depth = topology.current_frontier(loop.context.t0)
|
|
assert [source.location for source in frontier] == [child_location]
|
|
assert bfs_depth == 1
|
|
|
|
|
|
def test_stabilization_populates_bfs_before_the_independent_topology_dequeue(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""One raw BiGraph pull is supplied before independent topology admission."""
|
|
|
|
root_location = "src/foo.py"
|
|
child_location = "src/x.py"
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py"],
|
|
[(root_location, child_location)],
|
|
)
|
|
root = sources[root_location]
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
boundary = FrontierBoundary(
|
|
current_locations=frozenset({root.location}),
|
|
unchanged_locations=frozenset(),
|
|
live_seed_locations=frozenset({root.location}),
|
|
source_by_location=sources,
|
|
)
|
|
assert strategy.downstream_calls == []
|
|
bfs_slice = strategy.downstream_sources_for(
|
|
[sources[location] for location in sorted(boundary.live_seed_locations)],
|
|
_loop_file_manager(loop).src_classes,
|
|
_loop_file_manager(loop).src_root,
|
|
)
|
|
loop.context.topology.enqueue_bfs_slice(bfs_slice, boundary)
|
|
populated_calls = list(strategy.downstream_calls)
|
|
|
|
assert [source.location for source in loop.context.topology.dequeue_topologically()] == [
|
|
child_location,
|
|
]
|
|
assert strategy.downstream_calls == populated_calls
|
|
|
|
|
|
|
|
|
|
def _stage_pending_candidate(
|
|
loop: StabilizationLoop,
|
|
sources: dict[str, ClassMeta],
|
|
candidate_location: str,
|
|
*current_locations: str,
|
|
) -> None:
|
|
"""Stage a discovered candidate that has not yet passed topology admission."""
|
|
|
|
boundary = _boundary(sources, *current_locations)
|
|
loop.context.topology.enqueue_bfs_slice(
|
|
[sources[candidate_location]],
|
|
boundary,
|
|
)
|
|
loop.context.topology.deferred_topology_locations = {candidate_location}
|
|
|
|
|
|
def test_pending_deferred_topology_waits_for_active_bigraph_pull(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A pending candidate waits for the active pull that can close its requirement."""
|
|
|
|
root_location = "src/foo.py"
|
|
unchanged_location = "src/x.py"
|
|
changed_location = "src/a.py"
|
|
requirement_location = "src/b.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, unchanged_location),
|
|
(root_location, changed_location),
|
|
(changed_location, requirement_location),
|
|
(requirement_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "a.py", "b.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.total_cot = 2
|
|
loop.context.topology.frontiers = [
|
|
[sources[unchanged_location]],
|
|
[sources[changed_location]],
|
|
]
|
|
loop.context.topology.bfs_depth = 2
|
|
loop.context.topology.visited.update({
|
|
root_location,
|
|
unchanged_location,
|
|
changed_location,
|
|
})
|
|
loop.context.frontier_open_hashes[2] = {
|
|
changed_location: "opening-before-alpha",
|
|
}
|
|
_stage_pending_candidate(
|
|
loop,
|
|
sources,
|
|
candidate_location,
|
|
root_location,
|
|
)
|
|
assert not _loop_backlog(loop).inspection
|
|
assert not _loop_backlog(loop).remediation
|
|
assert not _loop_backlog(loop).testing
|
|
assert loop.context.topology.deferred_topology_locations == {
|
|
candidate_location
|
|
}
|
|
|
|
loop.run()
|
|
|
|
assert loop.context.blocked_summary == ""
|
|
assert loop.context.accepted
|
|
assert not loop.context.topology.deferred_topology_locations
|
|
assert list(map(
|
|
lambda frontier: tuple(map(lambda source: source.location, frontier)),
|
|
loop.context.topology.frontiers[-2:],
|
|
)) == [(requirement_location,), (candidate_location,)]
|
|
assert requirement_location in loop.context.topology.visited
|
|
assert candidate_location in loop.context.topology.visited
|
|
|
|
|
|
def test_unstaged_active_frontier_is_remaining_downstream(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A completed active frontier remains drawable before its boundary is staged."""
|
|
|
|
root_location = "src/foo.py"
|
|
active_location = "src/a.py"
|
|
requirement_location = "src/b.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, active_location),
|
|
(active_location, requirement_location),
|
|
(requirement_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["a.py", "b.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.total_cot = 2
|
|
loop.context.topology.frontiers = [[sources[active_location]]]
|
|
loop.context.topology.bfs_depth = 1
|
|
loop.context.topology.visited.update({root_location, active_location})
|
|
_stage_pending_candidate(
|
|
loop,
|
|
sources,
|
|
candidate_location,
|
|
root_location,
|
|
)
|
|
loop.context.frontier_open_hashes[1] = {
|
|
active_location: "opening-before-alpha"
|
|
}
|
|
loop.run()
|
|
|
|
assert loop.context.blocked_summary == ""
|
|
assert loop.context.accepted
|
|
assert requirement_location in loop.context.topology.visited
|
|
assert candidate_location in loop.context.topology.visited
|
|
|
|
|
|
def test_empty_downstream_still_plays_ready_topology(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Pushing an empty raw slice still permits an already-ready pending candidate."""
|
|
|
|
root_location = "src/foo.py"
|
|
active_location = "src/a.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, active_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["a.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.total_cot = 2
|
|
loop.context.topology.frontiers = [[sources[active_location]]]
|
|
loop.context.topology.bfs_depth = 1
|
|
loop.context.topology.visited.update({root_location, active_location})
|
|
_stage_pending_candidate(
|
|
loop,
|
|
sources,
|
|
candidate_location,
|
|
root_location,
|
|
)
|
|
loop.run()
|
|
|
|
assert loop.context.blocked_summary == ""
|
|
assert loop.context.accepted
|
|
assert candidate_location in loop.context.topology.visited
|
|
|
|
|
|
def test_stalled_topology_draws_next_boundary_instead_of_blocking(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A no-ready hand consumes only one boundary before the outer loop draws again."""
|
|
|
|
root_location = "src/foo.py"
|
|
empty_location = "src/x.py"
|
|
live_location = "src/a.py"
|
|
requirement_location = "src/b.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, empty_location),
|
|
(root_location, live_location),
|
|
(live_location, requirement_location),
|
|
(requirement_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "a.py", "b.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.total_cot = 2
|
|
loop.context.topology.frontiers = [
|
|
[sources[empty_location]],
|
|
[sources[live_location]],
|
|
]
|
|
loop.context.topology.bfs_depth = 2
|
|
loop.context.topology.visited.update({
|
|
root_location,
|
|
empty_location,
|
|
live_location,
|
|
})
|
|
loop.context.frontier_open_hashes[2] = {
|
|
live_location: "opening-before-alpha",
|
|
}
|
|
_stage_pending_candidate(loop, sources, candidate_location, root_location)
|
|
loop.run()
|
|
|
|
assert requirement_location in loop.context.topology.visited
|
|
assert loop.context.blocked_summary == ""
|
|
|
|
|
|
def test_reverse_discovery_queues_alpha_in_topological_order(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A lexically earlier dependent waits while Alpha receives its prerequisite first."""
|
|
|
|
root_location = "src/foo.py"
|
|
candidate_location = "src/a_candidate.py"
|
|
prerequisite_location = "src/z_prerequisite.py"
|
|
loop, _, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["a_candidate.py", "z_prerequisite.py"],
|
|
[
|
|
(root_location, candidate_location),
|
|
(root_location, prerequisite_location),
|
|
(prerequisite_location, candidate_location),
|
|
],
|
|
)
|
|
loop.context.total_cot = 2
|
|
# Existing whitelist policy keeps the unchanged root eligible as the raw BFS seed.
|
|
loop.context.topology.whitelist.add(root_location)
|
|
loop.run()
|
|
|
|
assert loop.context.blocked_summary == ""
|
|
assert loop.context.accepted
|
|
assert list(map(
|
|
lambda frontier: tuple(map(lambda source: source.location, frontier)),
|
|
loop.context.topology.frontiers,
|
|
)) == [(prerequisite_location,), (candidate_location,)]
|
|
assert loop.context.topology.visited == {
|
|
root_location,
|
|
prerequisite_location,
|
|
candidate_location,
|
|
}
|
|
assert not loop.context.topology.deferred_topology_locations
|
|
|
|
|
|
def test_block_requires_quiescence_and_current_owned_deferral(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Only a quiescent scheduler with current-COT deferral reaches the tripwire."""
|
|
|
|
root_location = "src/foo.py"
|
|
requirement_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, requirement_location),
|
|
(requirement_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["y.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.total_cot = 2
|
|
loop.context.topology.frontiers = [[]]
|
|
loop.context.topology.bfs_depth = 1
|
|
loop.context.topology.visited.add(root_location)
|
|
_stage_pending_candidate(loop, sources, candidate_location, root_location)
|
|
loop.run()
|
|
|
|
assert not loop.context.accepted
|
|
assert loop.context.blocked_summary == "Story-local backlog remains unresolved."
|
|
assert loop.context.topology.deferred_topology_locations == {
|
|
candidate_location
|
|
}
|
|
|
|
|
|
def test_exhaustion_requires_no_ready_no_downstream_no_deferral(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A quiescent non-final COT exhausts only after every topology lane is empty."""
|
|
|
|
root_location = "src/foo.py"
|
|
loop, sources, strategy = _perimeter_topology_loop(tmp_path, [], [])
|
|
loop.context.total_cot = 2
|
|
loop.context.topology.frontiers = [[]]
|
|
loop.context.topology.bfs_depth = 1
|
|
loop.context.topology.visited.add(root_location)
|
|
loop.run()
|
|
|
|
assert loop.context.accepted
|
|
assert loop.context.fully_explored
|
|
assert loop.context.blocked_summary == ""
|
|
assert not loop.context.topology.deferred_topology_locations
|
|
|
|
|
|
def test_stabilization_discovers_a_live_edge_added_after_the_prior_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A live edge added after Alpha's frontier work enters the next just-in-time BFS slice."""
|
|
|
|
root_location = "src/foo.py"
|
|
first_location = "src/a.py"
|
|
later_location = "src/b.py"
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["a.py", "b.py"],
|
|
[(root_location, first_location)],
|
|
frozen_edges=[(root_location, first_location)],
|
|
)
|
|
root = sources[root_location]
|
|
first = sources[first_location]
|
|
later = sources[later_location]
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
# The initial just-in-time query discovers only the frozen/live A edge.
|
|
assert [source.location for source in service_topology(loop, [root], {})] == [
|
|
first.location,
|
|
]
|
|
loop.context.append_frontier([first])
|
|
loop.context.topology.visited.add(first.location)
|
|
|
|
# Model the dependency graph update caused by Alpha's edit to A.
|
|
strategy.dependencies.setdefault(later.location, set()).add(first.location)
|
|
strategy.dependents.setdefault(first.location, set()).add(later.location)
|
|
|
|
# The next frontier queries live BFS after that mutation; B was not precomputed from root.
|
|
assert [source.location for source in service_topology(loop, [first], {})] == [
|
|
later.location,
|
|
]
|
|
assert strategy.downstream_calls == [
|
|
([root.location], False),
|
|
([first.location], False),
|
|
]
|
|
beam = loop.context.topology._beam
|
|
assert beam is not None
|
|
assert later.location in beam
|
|
|
|
|
|
def test_stabilization_draws_then_plays_the_complete_ready_topology_layer(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""One draw precedes admission and every sibling ready at its snapshot shares a layer."""
|
|
|
|
root_location = "src/foo.py"
|
|
first_location = "src/a.py"
|
|
join_location = "src/c.py"
|
|
later_location = "src/b.py"
|
|
initial_edges = [
|
|
(root_location, first_location),
|
|
(first_location, join_location),
|
|
(root_location, join_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["a.py", "b.py", "c.py"],
|
|
initial_edges,
|
|
frozen_edges=initial_edges,
|
|
)
|
|
root = sources[root_location]
|
|
first = sources[first_location]
|
|
join = sources[join_location]
|
|
later = sources[later_location]
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
# Root discovers A and the blocked join C; only A is ready for Alpha.
|
|
assert [source.location for source in service_topology(loop, [root], {})] == [
|
|
first.location,
|
|
]
|
|
assert loop.context.topology.deferred_topology_locations == {join.location}
|
|
loop.context.append_frontier([first])
|
|
loop.context.topology.visited.add(first.location)
|
|
|
|
# Alpha changes A's live graph. The next cadence draws A, then classifies
|
|
# both the newly discovered B and already-pending C against one snapshot.
|
|
strategy.dependencies.setdefault(later.location, set()).add(first.location)
|
|
strategy.dependents.setdefault(first.location, set()).add(later.location)
|
|
assert [source.location for source in service_topology(loop, [first], {})] == [
|
|
later.location,
|
|
join.location,
|
|
]
|
|
assert strategy.downstream_calls == [
|
|
([root.location], False),
|
|
([first.location], False),
|
|
]
|
|
|
|
|
|
def _final_exam_item(marker: ClassMeta) -> TestingBacklog:
|
|
"""Create the topology-shaped final exam sentinel."""
|
|
|
|
return TestingBacklog(
|
|
focused_spotlight=marker,
|
|
summary="last exam",
|
|
downstream_classes=[marker],
|
|
change_type=ChangeType.Update,
|
|
last_exam=True,
|
|
)
|
|
|
|
|
|
def test_stabilization_expands_changed_memento_to_deferred_class_meta(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A changed Memento frontier admits its dependent ClassMeta in memory."""
|
|
|
|
loop, sources, _ = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["memento.py", "class_meta.py"],
|
|
[
|
|
("src/foo.py", "src/memento.py"),
|
|
("src/memento.py", "src/class_meta.py"),
|
|
],
|
|
)
|
|
file_manager = _loop_file_manager(loop)
|
|
root = sources["src/foo.py"]
|
|
memento = sources["src/memento.py"]
|
|
class_meta = sources["src/class_meta.py"]
|
|
|
|
loop.context.topology.visited.add(root.location)
|
|
first_frontier = service_topology(loop, [root], {})
|
|
assert first_frontier == [memento]
|
|
loop.context.append_frontier(first_frontier)
|
|
loop.context.topology.visited.add(memento.location)
|
|
opening_hashes = loop._current_frontier_hashes()
|
|
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.source_path(memento.location),
|
|
"def memento():\n return 2\n",
|
|
)
|
|
file_manager.resync()
|
|
|
|
next_frontier = service_topology(loop, [memento], opening_hashes)
|
|
|
|
assert next_frontier == [class_meta]
|
|
|
|
|
|
|
|
|
|
def test_stabilization_passes_the_previous_root_only_into_alpha_loop(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Only Alpha receives the predecessor source; Beta keeps its ordinary evidence flow."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "prior.py", "def prior():\n return 1\n")
|
|
file_manager.resync()
|
|
loop = _scenario_stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
)
|
|
expected = file_manager.source_path("src/prior.py")
|
|
loop.previous_cot_root_readonly_file = expected
|
|
|
|
alpha = loop._create_alpha_loop(0)
|
|
beta_context = loop._create_beta_loop().context
|
|
|
|
assert alpha.previous_cot_root_readonly_file == expected
|
|
assert not hasattr(alpha.context, "previous_cot_root_readonly_file")
|
|
assert not hasattr(beta_context, "previous_cot_root_readonly_file")
|
|
assert alpha.context.pinned_context == []
|
|
assert beta_context.pinned_context == []
|
|
|
|
def test_stabilization_binds_scoped_file_manager_on_aider_worker(tmp_path: Path) -> None:
|
|
"""Verify stabilization scope, not Alpha/Beta, binds the worker FileManager."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
worker = EditingSourceWorker()
|
|
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
worker=worker,
|
|
)
|
|
|
|
assert worker.file_manager is _loop_file_manager(loop)
|
|
|
|
def test_stabilization_preserves_injected_gate(tmp_path: Path) -> None:
|
|
"""Verify StabilizationLoop construction does not replace the injected parent gate."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.src_classes[0]
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=source,
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: False,
|
|
)
|
|
|
|
loop = StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
|
|
assert loop.gate is gate
|
|
|
|
def test_stabilization_child_gates_receive_retry_checks_without_domain_calls(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify Alpha and Beta child gates receive retry checks that domain loops do not call."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _scenario_stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
)
|
|
source_frontier, source_index = loop.context.frontier
|
|
|
|
alpha = loop._create_alpha_loop(source_index)
|
|
beta = loop._create_beta_loop()
|
|
assert isinstance(alpha.gate.should_retry_check, PredicateCheck)
|
|
assert isinstance(beta.gate.should_retry_check, PredicateCheck)
|
|
alpha_retry = alpha.gate.should_retry_check.predicate
|
|
beta_retry = beta.gate.should_retry_check.predicate
|
|
|
|
assert isinstance(alpha_retry, RetryVibeCheck)
|
|
assert isinstance(beta_retry, RetryVibeCheck)
|
|
|
|
alpha.run()
|
|
beta.run()
|
|
|
|
assert alpha_retry.changes == []
|
|
assert beta_retry.changes == []
|
|
|
|
|
|
def test_stabilization_creates_empty_child_inboxes_without_resume(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify a fresh stabilization pass creates child contexts without resumed inbox work."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(file_manager, BackLog(), FakeCoverageStrategy(1.0))
|
|
|
|
alpha = loop._create_alpha_loop(0).context
|
|
beta = loop._create_beta_loop().context
|
|
|
|
assert alpha.inbound_queue == []
|
|
assert beta.inbound_queue == []
|
|
|
|
|
|
def test_alpha_and_beta_persist_strip_runtime_queue_depth_callbacks(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify alpha/beta checkpoints omit stabilization-owned queue-depth callbacks."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(file_manager, BackLog(), FakeCoverageStrategy(1.0))
|
|
|
|
alpha_context = loop._create_alpha_loop(0).context
|
|
beta_context = loop._create_beta_loop().context
|
|
|
|
assert alpha_context.alpha_postflight_callback is None
|
|
assert not hasattr(beta_context, "alpha_postflight_callback")
|
|
assert not hasattr(alpha_context, "previous_cot_root_readonly_file")
|
|
|
|
alpha_context.persist()
|
|
beta_context.persist()
|
|
|
|
|
|
def test_stabilization_persist_strips_runtime_callbacks_and_helpers(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify stabilization checkpoints omit injected runtime-only callbacks and helpers."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
context = RuntimeStabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=file_manager.require_src("src/foo.py"),
|
|
backlog=BackLog(),
|
|
)
|
|
StabilizationLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: False),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
),
|
|
)
|
|
context.new_files_callback = lambda: False
|
|
context.topology_summary = "Topology 0 source backlog pass."
|
|
context.test_runner_strategy = PassingRunner()
|
|
assert file_manager.dependency_graph_maintainer is not None
|
|
context.dependency_graph_maintainer = file_manager.dependency_graph_maintainer
|
|
|
|
checkpoint = context.persist()
|
|
persisted = pickle.loads(file_manager.repo_exec.read_bytes(checkpoint))
|
|
|
|
payload = persisted["payload"]
|
|
assert payload["goal"] == context.goal
|
|
assert not hasattr(context, "previous_cot_root_location")
|
|
assert "new_files_callback" not in payload
|
|
assert "topology_summary" not in payload
|
|
assert "test_runner_strategy" not in payload
|
|
assert "dependency_graph_maintainer" not in payload
|
|
|
|
|
|
def test_thawed_alpha_and_beta_restore_runtime_queue_depth_callbacks(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify resumed child loops publish their restored queue depths to Stabilization."""
|
|
|
|
previous_settings = resume_settings()
|
|
try:
|
|
file_manager = _file_manager(tmp_path)
|
|
configure_resume_settings(
|
|
hydrate=True,
|
|
checkpoint_dir=file_manager.repo_root / "checkpoints",
|
|
keep=previous_settings.keep,
|
|
repo_exec=file_manager.repo_exec,
|
|
)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(file_manager, BackLog(), FakeCoverageStrategy(1.0))
|
|
|
|
fresh_alpha = loop._create_alpha_loop(0).context
|
|
fresh_beta = loop._create_beta_loop().context
|
|
fresh_alpha.persist()
|
|
fresh_beta.persist()
|
|
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
fresh_alpha.move_inbound(
|
|
StagedInboundItem(
|
|
lane=BacklogLane.Inspection,
|
|
value=InspectionBacklog(source, "resume source inspection"),
|
|
inbound_id="resume-alpha",
|
|
)
|
|
)
|
|
fresh_beta.move_inbound(
|
|
StagedInboundItem(
|
|
lane=BacklogLane.Testing,
|
|
value=_testing_item(source, test),
|
|
inbound_id="resume-beta",
|
|
)
|
|
)
|
|
fresh_alpha.persist()
|
|
fresh_beta.persist()
|
|
|
|
resumed_alpha = loop._create_alpha_loop(0)
|
|
resumed_beta = loop._create_beta_loop()
|
|
resumed_alpha._publish_queue_depth()
|
|
assert loop.context.alpha_inbound_queue_depth == 1
|
|
assert loop.context.beta_inbound_queue_depth == 0
|
|
|
|
resumed_beta._publish_queue_depth()
|
|
assert loop.context.alpha_inbound_queue_depth == 0
|
|
assert loop.context.beta_inbound_queue_depth == 1
|
|
finally:
|
|
configure_resume_settings(
|
|
hydrate=previous_settings.hydrate,
|
|
checkpoint_dir=previous_settings.checkpoint_dir,
|
|
keep=previous_settings.keep,
|
|
repo_exec=previous_settings.repo_exec,
|
|
)
|
|
|
|
|
|
def test_stabilization_restores_the_graph_selected_for_resume(tmp_path: Path) -> None:
|
|
"""A resumed stabilization loop restores its durable graph outside hydration."""
|
|
|
|
previous_settings = resume_settings()
|
|
try:
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
graph = Provenance.open(
|
|
repo_root=file_manager.repo_root,
|
|
repo_exec=file_manager.repo_exec,
|
|
)
|
|
message_id = graph.start(
|
|
Phase.ALPHA,
|
|
InspectionBacklog(file_manager.require_src("src/foo.py"), "resume source"),
|
|
)
|
|
configure_resume_settings(
|
|
hydrate=True,
|
|
checkpoint_dir=file_manager.repo_root / "checkpoints",
|
|
keep=previous_settings.keep,
|
|
repo_exec=file_manager.repo_exec,
|
|
)
|
|
|
|
loop = _stabilization_loop(file_manager, BackLog(), FakeCoverageStrategy(1.0))
|
|
|
|
assert loop.provenance is not None
|
|
assert loop.provenance.graph_id == graph.graph_id
|
|
assert message_id in loop.provenance.messages
|
|
finally:
|
|
configure_resume_settings(
|
|
hydrate=previous_settings.hydrate,
|
|
checkpoint_dir=previous_settings.checkpoint_dir,
|
|
keep=previous_settings.keep,
|
|
repo_exec=previous_settings.repo_exec,
|
|
)
|
|
|
|
|
|
def test_standalone_stabilization_starts_a_new_graph_without_resume(tmp_path: Path) -> None:
|
|
"""An independently constructed stabilization loop cannot read a prior graph."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
first = _stabilization_loop(file_manager, BackLog(), FakeCoverageStrategy(1.0))
|
|
assert first.provenance is not None
|
|
message_id = first.provenance.start(
|
|
Phase.ALPHA,
|
|
InspectionBacklog(file_manager.require_src("src/foo.py"), "first task source"),
|
|
)
|
|
|
|
fresh = _stabilization_loop(file_manager, BackLog(), FakeCoverageStrategy(1.0))
|
|
|
|
assert fresh.provenance is not None
|
|
assert fresh.provenance.graph_id != first.provenance.graph_id
|
|
assert message_id not in fresh.provenance.messages
|
|
|
|
|
|
def test_stabilization_does_not_seed_alpha_completed_source_into_inspection(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify Alpha-completed sources do not re-enter inspection."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
completion = {
|
|
source.location: WorkCompletion(
|
|
location=source.location,
|
|
completed=True,
|
|
tests={
|
|
test.location: WorkCompletion(
|
|
location=test.location,
|
|
completed=True,
|
|
)
|
|
},
|
|
)
|
|
}
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=source,
|
|
completion=completion,
|
|
)
|
|
|
|
assert not _context_backlog(context).inspection
|
|
|
|
|
|
def test_stabilization_run_seeds_initial_root_inspection_after_graph_setup(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify the first root inspection seed happens from StabilizationLoop.run."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _scenario_stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
)
|
|
source = file_manager.require_src("src/foo.py")
|
|
|
|
assert not _loop_backlog(loop).inspection
|
|
|
|
events: list[str] = []
|
|
|
|
def run_source(index: int, source_backlog: ContributorBacklog) -> str:
|
|
"""Commit the seeded root inspection through the declared Alpha outcome."""
|
|
|
|
popped = source_backlog.pop_next_source_work()
|
|
assert popped is not None
|
|
with popped as scope:
|
|
item = scope.value
|
|
assert isinstance(item, InspectionBacklog)
|
|
events.append(_focused_metadata(item).location)
|
|
scope.commit()
|
|
return f"recorded topology {index}"
|
|
|
|
loop.set_source_runner(run_source)
|
|
|
|
loop.run()
|
|
|
|
assert events == [source.location]
|
|
|
|
|
|
def test_stabilization_remediation_clears_whole_source_completion_tree(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify remediation invalidates Alpha and every Beta completion under the blamed source."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
failing_test = file_manager.require_test("tests/test_foo.py")
|
|
completion = {
|
|
source.location: WorkCompletion(
|
|
location=source.location,
|
|
completed=True,
|
|
tests={
|
|
failing_test.location: WorkCompletion(
|
|
location=failing_test.location,
|
|
completed=True,
|
|
),
|
|
"tests/test_other.py": WorkCompletion(
|
|
location="tests/test_other.py",
|
|
completed=True,
|
|
),
|
|
},
|
|
)
|
|
}
|
|
backlog = BackLog(completion=completion)
|
|
|
|
backlog.queue_remediation(
|
|
_failed_item(source, failing_test, summary="repair foo")
|
|
)
|
|
|
|
assert source.location not in backlog.completion
|
|
|
|
|
|
def test_stabilization_dispatches_remediation_before_testing(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify existing scheduling sends source remediation to Alpha before Beta testing."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(source, test, summary="repair source first")
|
|
testing = _testing_item(source, test)
|
|
backlog = BackLog(remediation=[remediation], testing=[testing])
|
|
loop = _scenario_stabilization_loop(
|
|
file_manager,
|
|
backlog,
|
|
FakeCoverageStrategy(1.0),
|
|
)
|
|
events: list[str] = []
|
|
|
|
def run_source(_: int, source_backlog: ContributorBacklog) -> str:
|
|
"""Record and commit the first source item selected by Stabilization."""
|
|
|
|
events.append("alpha")
|
|
claimed = source_backlog.pop_next_source_work()
|
|
assert claimed is not None
|
|
assert claimed.value is remediation
|
|
with claimed as scope:
|
|
scope.commit()
|
|
return "Alpha consumed remediation."
|
|
|
|
def run_testing_pass() -> None:
|
|
"""Record Beta only after Alpha has cleared the source-side lane."""
|
|
|
|
events.append("beta")
|
|
claimed = backlog.pop_testing()
|
|
assert claimed is not None
|
|
assert claimed.value is testing
|
|
with claimed as scope:
|
|
scope.commit()
|
|
loop.context.fully_explored = True
|
|
|
|
loop.set_source_runner(run_source)
|
|
loop.set_testing_runner(run_testing_pass)
|
|
|
|
loop.run()
|
|
|
|
assert events == ["alpha", "beta"]
|
|
|
|
def test_stabilization_inspection_noop_consumes_without_replay(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify claimed inspection work is popped before Alpha even when later runs reseed root inspection."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
backlog = BackLog()
|
|
inspection = InspectionBacklog(source, "inspect foo")
|
|
backlog.queue_inspection(inspection)
|
|
worker = NoChangeWorker()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
backlog,
|
|
FakeCoverageStrategy(1.0),
|
|
worker=worker,
|
|
)
|
|
|
|
loop.run()
|
|
|
|
assert not backlog.inspection
|
|
assert len(worker.requests) == 1
|
|
|
|
loop.run()
|
|
|
|
assert len(worker.requests) == 2
|
|
|
|
def test_topological_admission_skips_visited_raw_sources(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A raw BFS hand never readmits a source already visited by this COT."""
|
|
|
|
loop, sources, _ = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["bar.py"],
|
|
[],
|
|
)
|
|
foo = sources["src/foo.py"]
|
|
bar = sources["src/bar.py"]
|
|
loop.context.topology.visited.add(foo.location)
|
|
loop.context.topology.enqueue_bfs_slice(
|
|
[foo, bar],
|
|
_boundary(sources, bar.location),
|
|
)
|
|
|
|
assert [source.location for source in loop.context.topology.dequeue_topologically()] == [
|
|
bar.location,
|
|
]
|
|
|
|
def test_class_meta_hash_reflects_current_source_content(tmp_path: Path) -> None:
|
|
"""Verify ClassMeta hash changes when source content changes."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
before = source.get_hash(file_manager)
|
|
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.source_path(source.location),
|
|
"def foo():\n return 2\n",
|
|
)
|
|
|
|
assert source.get_hash(file_manager) != before
|
|
|
|
def test_stabilization_prunes_unchanged_frontier_from_generic_expansion(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify unchanged current frontier nodes do not seed generic downstream expansion."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "bar.py",
|
|
"from foo import foo\n\ndef bar():\n return foo()\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.capture_frozen_once()
|
|
root = file_manager.require_src("src/foo.py")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
frontier_open_hashes = loop._current_frontier_hashes()
|
|
|
|
assert not service_topology(loop, [root], frontier_open_hashes)
|
|
assert loop.context.topology.unchanged_perimeter == set()
|
|
|
|
|
|
def test_topological_readiness_uses_only_the_explicit_unchanged_perimeter(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A fresh C closes through X while X stays out of ordinary BFS seeds."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "y.py", "c.py"],
|
|
edges,
|
|
)
|
|
perimeter, perimeter_hashes = _stage_unchanged_perimeter_join(loop, sources)
|
|
candidate = sources[candidate_location]
|
|
|
|
assert [
|
|
source.location
|
|
for source in service_topology(loop, [perimeter], perimeter_hashes)
|
|
] == [candidate.location]
|
|
assert strategy.downstream_calls == [
|
|
([root_location], False),
|
|
]
|
|
assert loop.context.topology.deferred_topology_locations == set()
|
|
assert loop.context.topology.unchanged_perimeter == {
|
|
perimeter.location,
|
|
}
|
|
assert loop.context.topology.visited == {
|
|
root_location,
|
|
perimeter.location,
|
|
candidate.location,
|
|
}
|
|
assert loop.context.topology._virtual_visited == {y_location}
|
|
assert y_location not in {
|
|
source.location
|
|
for frontier in loop.context.full_topologies
|
|
for source in frontier
|
|
}
|
|
assert y_location not in loop.context.topology.visited
|
|
|
|
|
|
def test_skipped_no_change_requirement_allows_nonfinal_cot_completion(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A pending join closes and permits completion when its only skipped leg ends at X."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, _ = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "y.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.total_cot = 2
|
|
root = sources[root_location]
|
|
perimeter = sources[x_location]
|
|
candidate = sources[candidate_location]
|
|
|
|
# Model Alpha completing a changed root, so ordinary BFS discovers X and C.
|
|
loop.context.topology.visited.add(root.location)
|
|
assert [
|
|
source.location
|
|
for source in service_topology(loop, [root], {root.location: "before-root"})
|
|
] == [perimeter.location]
|
|
assert loop.context.topology.deferred_topology_locations == {
|
|
candidate.location,
|
|
}
|
|
|
|
# Model Alpha completing X unchanged: Y is not traversed or visited.
|
|
loop.context.append_frontier([perimeter])
|
|
perimeter_hashes = loop._current_frontier_hashes()
|
|
loop.context.topology.visited.add(perimeter.location)
|
|
assert [
|
|
source.location
|
|
for source in service_topology(loop, [perimeter], perimeter_hashes)
|
|
] == [candidate.location]
|
|
assert loop.context.topology.unchanged_perimeter == {perimeter.location}
|
|
assert loop.context.topology._virtual_visited == {y_location}
|
|
assert y_location not in loop.context.topology.visited
|
|
|
|
# Once C completes, no deferred topology remains to block non-final COT completion.
|
|
loop.context.append_frontier([candidate])
|
|
loop.context.topology.visited.add(candidate.location)
|
|
loop.run()
|
|
assert loop.context.accepted
|
|
assert not loop.context.blocked_summary
|
|
assert loop.context.topology.deferred_topology_locations == set()
|
|
|
|
|
|
def test_deferred_topology_ignores_requirements_outside_the_current_cot_beam(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""An external direct dependency neither blocks topology admission nor triggers lookback."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
outside_location = "src/outside.py"
|
|
frozen_edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "y.py", "c.py", "outside.py"],
|
|
[*frozen_edges, (outside_location, candidate_location)],
|
|
frozen_edges=frozen_edges,
|
|
)
|
|
perimeter, perimeter_hashes = _stage_unchanged_perimeter_join(loop, sources)
|
|
|
|
assert [
|
|
source.location
|
|
for source in service_topology(loop, [perimeter], perimeter_hashes)
|
|
] == [candidate_location]
|
|
assert not any(
|
|
outside_location in {start, target}
|
|
for start, target, _ in strategy.has_path_calls
|
|
)
|
|
assert outside_location not in loop.context.topology._virtual_visited
|
|
|
|
|
|
def test_deferred_topology_rejects_unchanged_perimeter_closure_with_an_alternate_live_path(
|
|
tmp_path: Path,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""A live root-to-Y route leaves C deferred despite an unchanged X route."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
z_location = "src/z.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(root_location, z_location),
|
|
(z_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
logger = logging.getLogger("deferred-topology-alternate-path")
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "z.py", "y.py", "c.py"],
|
|
edges,
|
|
logger=logger,
|
|
)
|
|
perimeter, perimeter_hashes = _stage_unchanged_perimeter_join(loop, sources)
|
|
caplog.set_level(logging.INFO, logger=logger.name)
|
|
|
|
assert [
|
|
source.location
|
|
for source in service_topology(loop, [perimeter], perimeter_hashes)
|
|
] == [z_location]
|
|
assert loop.context.topology.deferred_topology_locations == {candidate_location}
|
|
assert y_location not in loop.context.topology.visited
|
|
assert y_location not in loop.context.topology._virtual_visited
|
|
# The indexed reverse walk proves the live route without a repeated whole-path query.
|
|
assert strategy.has_path_calls == []
|
|
messages = [
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if record.getMessage().startswith("Deferred topology reconciliation ")
|
|
]
|
|
assert len(messages) == 1
|
|
payload = json.loads(messages[0].split(
|
|
"Deferred topology reconciliation ",
|
|
1,
|
|
)[1])
|
|
candidate = payload["candidates"][candidate_location]
|
|
assert candidate["decision"] == "deferred"
|
|
assert candidate["direct_in_beam_requirements"] == [root_location, y_location]
|
|
assert candidate["visited_requirements"] == [root_location]
|
|
assert candidate["unmet_requirements"] == [y_location]
|
|
assert candidate["skip_witness_by_requirement"] == {y_location: [x_location]}
|
|
assert candidate["alternate_live_path_by_requirement"] == {y_location: True}
|
|
assert candidate["rejection_reason"] == "alternate_live_path"
|
|
|
|
|
|
def test_topological_virtual_visits_reuse_a_proven_requirement_across_reconciliations(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""One successful Y proof is a COT-local virtual closure for a later BFS slice."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
mid_location = "src/mid.py"
|
|
y_location = "src/y.py"
|
|
first_candidate_location = "src/c.py"
|
|
second_candidate_location = "src/d.py"
|
|
frozen_edges = [
|
|
(root_location, x_location),
|
|
(x_location, mid_location),
|
|
(mid_location, y_location),
|
|
(y_location, first_candidate_location),
|
|
(y_location, second_candidate_location),
|
|
(root_location, first_candidate_location),
|
|
(root_location, second_candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "mid.py", "y.py", "c.py", "d.py"],
|
|
[
|
|
edge
|
|
for edge in frozen_edges
|
|
if edge != (root_location, second_candidate_location)
|
|
],
|
|
frozen_edges=frozen_edges,
|
|
)
|
|
perimeter, perimeter_hashes = _stage_unchanged_perimeter_join(loop, sources)
|
|
|
|
assert [
|
|
source.location
|
|
for source in service_topology(loop, [perimeter], perimeter_hashes)
|
|
] == [first_candidate_location]
|
|
assert loop.context.topology.deferred_topology_locations == set()
|
|
assert loop.context.topology.unchanged_perimeter == {x_location}
|
|
assert loop.context.topology._virtual_visited == {
|
|
mid_location,
|
|
y_location,
|
|
}
|
|
assert not ({mid_location, y_location} & loop.context.topology.visited)
|
|
assert not ({mid_location, y_location} & {
|
|
source.location
|
|
for frontier in loop.context.topology.frontiers
|
|
for source in frontier
|
|
})
|
|
|
|
loop.context.append_frontier([sources[first_candidate_location]])
|
|
strategy.dependencies.setdefault(second_candidate_location, set()).add(
|
|
first_candidate_location
|
|
)
|
|
strategy.dependents.setdefault(first_candidate_location, set()).add(
|
|
second_candidate_location
|
|
)
|
|
strategy.has_path_calls.clear()
|
|
|
|
assert [
|
|
source.location
|
|
for source in service_topology(
|
|
loop,
|
|
[sources[first_candidate_location]],
|
|
{},
|
|
)
|
|
] == [second_candidate_location]
|
|
assert strategy.has_path_calls == []
|
|
|
|
|
|
def test_deferred_topology_excludes_future_cot_perimeter_locations(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A later-COT source cannot become an unchanged perimeter for current-COT closure."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
future_location = "src/future.py"
|
|
frozen_edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, strategy = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "y.py", "c.py", "future.py"],
|
|
[*frozen_edges, (future_location, y_location)],
|
|
frozen_edges=frozen_edges,
|
|
future_cot_records=[
|
|
CotTaskRecord(
|
|
"future COT",
|
|
future_location,
|
|
ClassMeta(future_location, "future", [], {}),
|
|
),
|
|
],
|
|
)
|
|
loop.context.topology.visited.add(root_location)
|
|
loop.context.topology.deferred_topology_locations = {candidate_location}
|
|
loop.context.topology.unchanged_perimeter = {future_location}
|
|
strategy.dependents[root_location].discard(x_location)
|
|
|
|
assert service_topology(loop, [sources[root_location]], {}) == []
|
|
assert loop.context.topology.deferred_topology_locations == {candidate_location}
|
|
assert loop.context.topology.unchanged_perimeter == set()
|
|
|
|
|
|
def test_terminal_deferred_topology_snapshot_reports_the_unresolved_alternate_path(
|
|
tmp_path: Path,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""The terminal tripwire identifies C, Y, X, and the live Z route that blocks closure."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
z_location = "src/z.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(root_location, z_location),
|
|
(z_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
logger = logging.getLogger("deferred-topology-terminal")
|
|
loop, sources, _ = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "z.py", "y.py", "c.py"],
|
|
edges,
|
|
logger=logger,
|
|
)
|
|
loop.context.topology.frontiers = [[]]
|
|
loop.context.topology.bfs_depth = 1
|
|
loop.context.fully_explored = True
|
|
loop.context.topology.visited.update({root_location, x_location})
|
|
loop.context.topology.unchanged_perimeter = {x_location}
|
|
loop.context.topology.deferred_topology_locations = {candidate_location}
|
|
caplog.set_level(logging.INFO, logger=logger.name)
|
|
|
|
loop.run()
|
|
|
|
messages = [
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if record.getMessage().startswith("Deferred topology unresolved snapshot ")
|
|
]
|
|
assert len(messages) == 1
|
|
payload = json.loads(messages[0].split(
|
|
"Deferred topology unresolved snapshot ",
|
|
1,
|
|
)[1])
|
|
candidate = payload["deferred_candidates"][candidate_location]
|
|
assert candidate["unmet_requirements"] == [y_location]
|
|
assert candidate["skip_witness_by_requirement"] == {y_location: [x_location]}
|
|
assert candidate["alternate_live_path_by_requirement"] == {y_location: True}
|
|
assert candidate["rejection_reason"] == "alternate_live_path"
|
|
assert payload["queues"] == {
|
|
"inspection": [],
|
|
"runnable_remediation": [],
|
|
"testing": [],
|
|
"inherited_remediation": [],
|
|
"deferred_remediation": [],
|
|
}
|
|
assert loop.context.blocked_summary == "Story-local backlog remains unresolved."
|
|
|
|
|
|
def test_unexplained_pending_requirement_keeps_stabilization_incomplete(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A pending join with a live alternate path retains the incomplete completion state."""
|
|
|
|
root_location = "src/foo.py"
|
|
x_location = "src/x.py"
|
|
z_location = "src/z.py"
|
|
y_location = "src/y.py"
|
|
candidate_location = "src/c.py"
|
|
edges = [
|
|
(root_location, x_location),
|
|
(x_location, y_location),
|
|
(root_location, z_location),
|
|
(z_location, y_location),
|
|
(y_location, candidate_location),
|
|
(root_location, candidate_location),
|
|
]
|
|
loop, sources, _ = _perimeter_topology_loop(
|
|
tmp_path,
|
|
["x.py", "z.py", "y.py", "c.py"],
|
|
edges,
|
|
)
|
|
loop.context.topology.frontiers = [[]]
|
|
loop.context.topology.bfs_depth = 1
|
|
loop.context.fully_explored = True
|
|
loop.context.topology.visited.update({root_location, x_location})
|
|
loop.context.topology.unchanged_perimeter = {x_location}
|
|
loop.context.topology.deferred_topology_locations = {candidate_location}
|
|
summary = loop.run()
|
|
|
|
assert not loop.context.accepted
|
|
assert loop.context.blocked_summary == "Story-local backlog remains unresolved."
|
|
assert "Stabilization incomplete; upstream squash was not performed." in summary
|
|
assert _scenario_file_manager(loop).squash_summaries == []
|
|
|
|
|
|
def test_stabilization_admits_diamond_frontiers_and_ignores_external_requirements(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify admission separates a shared prerequisite from an external source."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "y.py",
|
|
"from foo import foo\n\ndef y():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "a.py", "def a():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "z.py",
|
|
"from foo import foo\nfrom y import y\nfrom a import a\n\ndef z():\n return foo(), y(), a()\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.capture_frozen_once()
|
|
root = file_manager.require_src("src/foo.py")
|
|
y = file_manager.require_src("src/y.py")
|
|
z = file_manager.require_src("src/z.py")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: True),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
root_hashes = loop._current_frontier_hashes()
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.source_path(root.location),
|
|
"def foo():\n return 2\n",
|
|
)
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
assert [source.location for source in service_topology(loop, [root], root_hashes)] == [
|
|
y.location,
|
|
]
|
|
assert context.topology.deferred_topology_locations == {z.location}
|
|
|
|
context.append_frontier([y])
|
|
loop.context.topology.visited.add(y.location)
|
|
y_hashes = {y.location: "before-y"}
|
|
|
|
assert [source.location for source in service_topology(loop, [y], y_hashes)] == [
|
|
z.location,
|
|
]
|
|
assert not context.topology.deferred_topology_locations
|
|
|
|
|
|
def test_stabilization_admission_releases_hidden_long_blocker_one_frontier_at_a_time(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify a transitive prerequisite becomes eligible only after each relay closes."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "a.py",
|
|
"from foo import foo\n\ndef a():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "b.py",
|
|
"from a import a\n\ndef b():\n return a()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "z.py",
|
|
"from foo import foo\nfrom b import b\n\ndef z():\n return foo(), b()\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.capture_frozen_once()
|
|
root = file_manager.require_src("src/foo.py")
|
|
a = file_manager.require_src("src/a.py")
|
|
b = file_manager.require_src("src/b.py")
|
|
z = file_manager.require_src("src/z.py")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: True),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
root_hashes = loop._current_frontier_hashes()
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.source_path(root.location),
|
|
"def foo():\n return 2\n",
|
|
)
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
assert [source.location for source in service_topology(loop, [root], root_hashes)] == [
|
|
a.location,
|
|
]
|
|
assert context.topology.deferred_topology_locations == {z.location}
|
|
|
|
context.append_frontier([a])
|
|
loop.context.topology.visited.add(a.location)
|
|
assert [source.location for source in service_topology(loop, [a], {a.location: "before-a"})] == [
|
|
b.location,
|
|
]
|
|
assert context.topology.deferred_topology_locations == {z.location}
|
|
|
|
loop.context.topology.visited.add(a.location)
|
|
context.append_frontier([b])
|
|
loop.context.topology.visited.add(b.location)
|
|
assert [source.location for source in service_topology(loop, [b], {b.location: "before-b"})] == [
|
|
z.location,
|
|
]
|
|
|
|
def test_deferred_whitelist_overrides_unchanged_frontier_pruning(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify deferred remediation can reopen a path pruned by unchanged hashes."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "mid.py",
|
|
"from foo import foo\n\ndef mid():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "target.py",
|
|
"from mid import mid\n\ndef target():\n return mid()\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.capture_frozen_once()
|
|
root = file_manager.require_src("src/foo.py")
|
|
target = file_manager.require_src("src/target.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(target, test, summary="repair target")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
deferred_remediation_backlog={target.location: [remediation]},
|
|
)
|
|
_context_backlog(context).replace_inspection([])
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
frontier_open_hashes = loop._current_frontier_hashes()
|
|
|
|
loop.context.topology.visited.add(root.location)
|
|
loop._grow_deferred_frontier_whitelist_locations([root])
|
|
assert loop.context.topology.whitelist == {
|
|
"src/foo.py",
|
|
"src/mid.py",
|
|
"src/target.py",
|
|
}
|
|
assert [source.location for source in service_topology(loop,
|
|
[root],
|
|
frontier_open_hashes,
|
|
)] == ["src/mid.py"]
|
|
|
|
def test_deferred_whitelist_readds_traversed_seed_for_direct_beam(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify whitelist evidence seeds only direct downstream expansion."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "mid.py",
|
|
"from foo import foo\n\ndef mid():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "next.py",
|
|
"from mid import mid\n\ndef next_value():\n return mid()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "target.py",
|
|
"from next import next_value\n\ndef target():\n return next_value()\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.capture_frozen_once()
|
|
root = file_manager.require_src("src/foo.py")
|
|
mid = file_manager.require_src("src/mid.py")
|
|
target = file_manager.require_src("src/target.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(target, test, summary="repair target")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
topology=TopologicalLiveliness(
|
|
frontiers=[[mid]],
|
|
bfs_depth=1,
|
|
whitelist={
|
|
"src/mid.py",
|
|
"src/next.py",
|
|
"src/target.py",
|
|
},
|
|
),
|
|
deferred_remediation_backlog={target.location: [remediation]},
|
|
)
|
|
_context_backlog(context).replace_inspection([])
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
loop.context.topology.visited.add(mid.location)
|
|
frontier_open_hashes = loop._current_frontier_hashes()
|
|
|
|
assert [source.location for source in service_topology(loop, [mid], frontier_open_hashes)] == [
|
|
"src/next.py",
|
|
]
|
|
|
|
|
|
def test_deferred_whitelist_preserves_ordered_topology_for_unchanged_joint_beam(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify deferred remediation advances unchanged beam sources in topology order."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "b.py",
|
|
"from foo import foo\n\ndef b():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "c.py",
|
|
"from b import b\n\ndef c():\n return b()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "d.py",
|
|
"from c import c\nfrom foo import foo\n\ndef d():\n return c(), foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "x.py", "def x():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "y.py",
|
|
"from x import x\n\ndef y():\n return x()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "z.py",
|
|
"from y import y\n\ndef z():\n return y()\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.capture_frozen_once()
|
|
root = file_manager.require_src("src/foo.py")
|
|
b = file_manager.require_src("src/b.py")
|
|
c = file_manager.require_src("src/c.py")
|
|
target = file_manager.require_src("src/d.py")
|
|
x = file_manager.require_src("src/x.py")
|
|
y = file_manager.require_src("src/y.py")
|
|
z = file_manager.require_src("src/z.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(target, test, summary="repair target")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
surrogate_roots=[x],
|
|
deferred_remediation_backlog={target.location: [remediation]},
|
|
)
|
|
_context_backlog(context).replace_inspection([])
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
|
|
root_hashes = loop._current_frontier_hashes()
|
|
loop.context.topology.visited.update({root.location, x.location})
|
|
loop._grow_deferred_frontier_whitelist_locations([root, x])
|
|
|
|
assert loop.context.topology.whitelist == {
|
|
root.location,
|
|
b.location,
|
|
c.location,
|
|
target.location,
|
|
}
|
|
assert [source.location for source in service_topology(loop,
|
|
[root, x],
|
|
root_hashes,
|
|
)] == [b.location]
|
|
assert loop.context.topology.deferred_topology_locations == {target.location}
|
|
|
|
context.append_frontier([b])
|
|
loop.context.topology.visited.add(b.location)
|
|
b_hashes = loop._current_frontier_hashes()
|
|
loop._grow_deferred_frontier_whitelist_locations([b])
|
|
|
|
assert [source.location for source in service_topology(loop, [b], b_hashes)] == [
|
|
c.location,
|
|
]
|
|
assert loop.context.topology.deferred_topology_locations == {target.location}
|
|
|
|
context.append_frontier([c])
|
|
loop.context.topology.visited.add(c.location)
|
|
c_hashes = loop._current_frontier_hashes()
|
|
loop._grow_deferred_frontier_whitelist_locations([c])
|
|
|
|
assert [source.location for source in service_topology(loop, [c], c_hashes)] == [
|
|
target.location,
|
|
]
|
|
assert not loop.context.topology.deferred_topology_locations
|
|
context.append_frontier([target])
|
|
assert [
|
|
[source.location for source in frontier]
|
|
for frontier in context.full_topologies
|
|
] == [
|
|
[root.location, x.location],
|
|
[b.location],
|
|
[c.location],
|
|
[target.location],
|
|
]
|
|
assert {y.location, z.location}.isdisjoint(
|
|
source.location
|
|
for frontier in context.full_topologies
|
|
for source in frontier
|
|
)
|
|
|
|
def test_stabilization_remediation_noop_consumes_without_replay(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify claimed remediation work is popped before Alpha even when later runs reseed root inspection."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
backlog = BackLog()
|
|
remediation = _failed_item(source, test, summary="repair source")
|
|
backlog.queue_remediation(remediation)
|
|
worker = NoChangeWorker()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
backlog,
|
|
FakeCoverageStrategy(1.0),
|
|
worker=worker,
|
|
)
|
|
|
|
loop.run()
|
|
|
|
assert not backlog.remediation
|
|
assert len(worker.requests) == 1
|
|
|
|
loop.run()
|
|
|
|
assert len(worker.requests) == 2
|
|
|
|
def test_stabilization_inspection_claimed_change_creates_testing_backlog(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify changed inspection sources still create testing after source claim."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
backlog = BackLog()
|
|
inspection = InspectionBacklog(source, "inspect foo")
|
|
backlog.queue_inspection(inspection)
|
|
worker = EditingSourceWorker()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
backlog,
|
|
FakeCoverageStrategy(1.0),
|
|
worker=worker,
|
|
)
|
|
|
|
_, source_index = loop.context.frontier
|
|
loop.context.active_topology = source_index
|
|
loop.context.persist()
|
|
alpha = loop._create_alpha_loop(source_index)
|
|
loop._run(alpha)
|
|
source_frontier, _ = loop.context.frontier
|
|
loop.context.topology.visited.update(
|
|
source.location
|
|
for source in source_frontier
|
|
)
|
|
loop.context.active_topology = None
|
|
loop.context.persist()
|
|
|
|
assert not backlog.inspection
|
|
assert [_focused_metadata(item).location for item in backlog.testing] == ["tests/test_foo.py"]
|
|
|
|
def test_stabilization_remediation_claimed_change_creates_testing_backlog(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify changed remediation sources still create testing after source claim."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
backlog = BackLog()
|
|
remediation = _failed_item(source, test, summary="repair source")
|
|
backlog.queue_remediation(remediation)
|
|
worker = EditingSourceWorker()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
backlog,
|
|
FakeCoverageStrategy(1.0),
|
|
worker=worker,
|
|
)
|
|
|
|
_, source_index = loop.context.frontier
|
|
loop.context.active_topology = source_index
|
|
loop.context.persist()
|
|
alpha = loop._create_alpha_loop(source_index)
|
|
loop._run(alpha)
|
|
source_frontier, _ = loop.context.frontier
|
|
loop.context.topology.visited.update(
|
|
source.location
|
|
for source in source_frontier
|
|
)
|
|
loop.context.active_topology = None
|
|
loop.context.persist()
|
|
|
|
assert not backlog.remediation
|
|
assert [_focused_metadata(item).location for item in backlog.testing] == ["tests/test_foo.py"]
|
|
|
|
def test_stabilization_keeps_nonforward_remediation_outside_current_frontier_runnable(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify an outside-trajectory remediation no longer needs local-frontier membership."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "zbar.py", "def zbar():\n return 1\n")
|
|
file_manager.resync()
|
|
blame = file_manager.require_src("src/zbar.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
backlog = BackLog()
|
|
remediation = _failed_item(blame, test, summary="repair zbar")
|
|
backlog.queue_remediation(remediation)
|
|
loop = _stabilization_loop(file_manager, backlog, FakeCoverageStrategy(1.0))
|
|
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert loop.context.deferred_remediation_backlog == {}
|
|
|
|
def test_stabilization_promotes_deferred_remediation_in_current_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify deferred remediation is queued once its blame source is visible."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
blame = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(blame, test, summary="repair foo")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=blame,
|
|
deferred_remediation_backlog={blame.location: [remediation]},
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
_scenario_file_manager(loop).set_diff(
|
|
"src/helper.py",
|
|
"@@ -1 +1 @@\n-def helper():\n+def helper():\n+ return 2\n",
|
|
)
|
|
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert _loop_backlog(loop).remediation
|
|
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert not loop.context.deferred_remediation_backlog
|
|
|
|
|
|
def test_stabilization_promotion_waits_for_every_blamed_source(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify promotion does not use only the deferred focused-location key."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "later.py", "def later():\n return 1\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
later = file_manager.require_src("src/later.py")
|
|
_freeze_dependency_edges(file_manager, [(root.location, later.location)])
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = FailedQAItem(
|
|
focused_spotlight=root,
|
|
summary="repair two sources",
|
|
downstream_classes=[root, later],
|
|
failed_test=FailingTest(
|
|
test=test,
|
|
error_message="failed",
|
|
stack_trace=["trace"],
|
|
raw_output="raw",
|
|
),
|
|
)
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
deferred_remediation_backlog={root.location: [remediation]},
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
|
|
loop._promote_deferred_remediation(loop._current_cot_forward_remediation_locations())
|
|
assert not _loop_backlog(loop).remediation
|
|
assert loop.context.deferred_remediation_backlog[root.location] == [remediation]
|
|
|
|
loop.context.append_frontier([later])
|
|
loop._promote_deferred_remediation(loop._current_cot_forward_remediation_locations())
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert not loop.context.deferred_remediation_backlog
|
|
|
|
|
|
def test_stabilization_promotion_extends_remediation_with_previous_frontier_sources(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify deferred remediation promotion appends previous-frontier readonly sources."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "helper.py", "def helper():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "bar.py", "def bar():\n return helper()\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
previous = file_manager.require_src("src/helper.py")
|
|
blame = file_manager.require_src("src/bar.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(blame, test, summary="repair bar", upstream_sources=[root])
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
topology=TopologicalLiveliness(
|
|
frontiers=[[previous], [blame]],
|
|
bfs_depth=2,
|
|
),
|
|
deferred_remediation_backlog={blame.location: [remediation]},
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
_scenario_file_manager(loop).set_diff(
|
|
"src/helper.py",
|
|
"@@ -1 +1 @@\n-def helper():\n+def helper():\n+ return 2\n",
|
|
)
|
|
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert _loop_backlog(loop).remediation
|
|
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert [
|
|
upstream.location
|
|
for upstream in remediation.upstream_classes
|
|
] == ["src/foo.py"]
|
|
assert not loop.context.deferred_remediation_backlog
|
|
|
|
|
|
def test_stabilization_shaping_extends_active_remediation_with_previous_frontier_sources(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify active remediation is enriched from the previous frontier during shaping."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "helper.py", "def helper():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "bar.py", "def bar():\n return helper()\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
previous = file_manager.require_src("src/helper.py")
|
|
blame = file_manager.require_src("src/bar.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
backlog = BackLog()
|
|
remediation = _failed_item(blame, test, summary="repair bar", upstream_sources=[root])
|
|
backlog.queue_remediation(remediation)
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
topology=TopologicalLiveliness(
|
|
frontiers=[[previous], [blame]],
|
|
bfs_depth=2,
|
|
),
|
|
backlog=backlog,
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
_scenario_file_manager(loop).set_diff(
|
|
"src/helper.py",
|
|
"@@ -1 +1 @@\n-def helper():\n+def helper():\n+ return 2\n",
|
|
)
|
|
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert _loop_backlog(loop).remediation
|
|
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert [
|
|
upstream.location
|
|
for upstream in remediation.upstream_classes
|
|
] == ["src/foo.py"]
|
|
|
|
def test_deferred_expansion_schedules_alpha_before_another_expansion(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify each deferred expansion is followed by one runnable Alpha pass."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "mid.py", "def mid():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "ztarget.py", "def ztarget():\n return 1\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
mid = file_manager.require_src("src/mid.py")
|
|
target = file_manager.require_src("src/ztarget.py")
|
|
_freeze_dependency_edges(
|
|
file_manager,
|
|
[(root.location, mid.location), (mid.location, target.location)],
|
|
)
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
current_remediation = _failed_item(
|
|
root,
|
|
test,
|
|
summary="redundant current-frontier repair",
|
|
)
|
|
remediation = _failed_item(target, test, summary="repair target")
|
|
inspection = InspectionBacklog(
|
|
root,
|
|
"inspect root",
|
|
downstream_classes=[root],
|
|
)
|
|
testing = _testing_item(root, test)
|
|
backlog = BackLog(
|
|
remediation=[current_remediation],
|
|
inspection=[inspection],
|
|
testing=[testing],
|
|
)
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
total_cot=2,
|
|
backlog=backlog,
|
|
deferred_remediation_backlog={target.location: [remediation]},
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
loop = ScenarioStabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
phase_events: list[str] = []
|
|
|
|
events: list[tuple[str, int, list[str]]] = []
|
|
handled_remediation: list[FailedQAItem] = []
|
|
|
|
def run_source(index: int, source_backlog: ContributorBacklog) -> str:
|
|
"""Commit one declared source item and retain its scheduler-visible result."""
|
|
|
|
lane = "remediation" if _loop_backlog(loop).remediation else "inspection"
|
|
popped = source_backlog.pop_next_source_work()
|
|
assert popped is not None
|
|
with popped as scope:
|
|
item = scope.value
|
|
source_locations = {
|
|
item.downstream_classes[0].location
|
|
if isinstance(item, FailedQAItem)
|
|
else _focused_metadata(item).location
|
|
}
|
|
events.append((lane, index, sorted(source_locations)))
|
|
phase_events.append(f"{lane}:{next(iter(source_locations))}")
|
|
if isinstance(item, FailedQAItem):
|
|
handled_remediation.append(item)
|
|
scope.commit()
|
|
return "recorded"
|
|
|
|
def run_testing_pass() -> None:
|
|
phase_events.append("testing")
|
|
popped = _loop_backlog(loop).pop_testing()
|
|
assert popped is not None
|
|
assert popped.value is testing
|
|
with popped as scope:
|
|
scope.commit()
|
|
|
|
loop.set_source_runner(run_source)
|
|
loop.set_testing_runner(run_testing_pass)
|
|
|
|
loop.run()
|
|
|
|
assert [
|
|
[source.location for source in frontier]
|
|
for frontier in loop.context.full_topologies
|
|
] == [[root.location], [mid.location], [target.location]]
|
|
assert events == [
|
|
("remediation", 0, ["src/foo.py"]),
|
|
("inspection", 0, ["src/foo.py"]),
|
|
("inspection", 1, ["src/mid.py"]),
|
|
("remediation", 2, ["src/ztarget.py"]),
|
|
("inspection", 2, ["src/ztarget.py"]),
|
|
]
|
|
assert handled_remediation == [current_remediation, remediation]
|
|
assert phase_events == [
|
|
"remediation:src/foo.py",
|
|
"inspection:src/foo.py",
|
|
"testing",
|
|
"inspection:src/mid.py",
|
|
"remediation:src/ztarget.py",
|
|
"inspection:src/ztarget.py",
|
|
]
|
|
assert not loop.context.deferred_remediation_backlog
|
|
|
|
|
|
def test_deferred_expansion_drains_inspection_before_second_expansion(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify a reached frontier gets an Alpha turn before deferred expansion repeats."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "mid.py", "def mid():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "relay.py", "def relay():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "ztarget.py", "def ztarget():\n return 1\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
mid = file_manager.require_src("src/mid.py")
|
|
relay = file_manager.require_src("src/relay.py")
|
|
target = file_manager.require_src("src/ztarget.py")
|
|
_freeze_dependency_edges(
|
|
file_manager,
|
|
[
|
|
(root.location, mid.location),
|
|
(mid.location, relay.location),
|
|
(relay.location, target.location),
|
|
],
|
|
)
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(target, test, summary="repair target")
|
|
loop = _scenario_stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
)
|
|
context = loop.context
|
|
context.append_frontier([mid])
|
|
context.topology.visited.update({root.location, mid.location})
|
|
context.frontier_open_hashes[1] = {mid.location: "before-alpha"}
|
|
context.deferred_remediation_backlog = {target.location: [remediation]}
|
|
context.persist()
|
|
events: list[str] = []
|
|
inspection_depths_after_alpha: list[int] = []
|
|
|
|
def run_source(_: int, source_backlog: ContributorBacklog) -> str:
|
|
"""Drain one source item and expose the resulting inspection depth."""
|
|
|
|
lane = "remediation" if _context_backlog(context).remediation else "inspection"
|
|
popped = source_backlog.pop_next_source_work()
|
|
assert popped is not None
|
|
with popped as scope:
|
|
item = scope.value
|
|
events.append(f"alpha:{lane}:{_focused_metadata(item).location}")
|
|
if _focused_metadata(item).location == relay.location:
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.source_path(relay.location),
|
|
"def relay():\n return 2\n",
|
|
)
|
|
scope.commit()
|
|
inspection_depths_after_alpha.append(len(_context_backlog(context).inspection))
|
|
return "recorded"
|
|
|
|
loop.set_source_runner(run_source)
|
|
|
|
loop.run()
|
|
|
|
assert events[:2] == [
|
|
"alpha:inspection:src/relay.py",
|
|
"alpha:remediation:src/ztarget.py",
|
|
]
|
|
assert inspection_depths_after_alpha == [0, 1, 0]
|
|
assert not _context_backlog(context).inspection
|
|
assert context.topology.visited == {
|
|
root.location,
|
|
mid.location,
|
|
relay.location,
|
|
target.location,
|
|
}
|
|
|
|
|
|
def test_deferred_remediation_promotes_only_after_blame_enters_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify deferred remediation waits until its blamed source is visible."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.client = FakeClient(acceptance_reasons=("return 1",))
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "mid.py", "def mid():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "ztarget.py", "def ztarget():\n return 1\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
mid = file_manager.require_src("src/mid.py")
|
|
target = file_manager.require_src("src/ztarget.py")
|
|
_freeze_dependency_edges(
|
|
file_manager,
|
|
[(root.location, mid.location), (mid.location, target.location)],
|
|
)
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(target, test, summary="repair target")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
deferred_remediation_backlog={target.location: [remediation]},
|
|
)
|
|
_context_backlog(context).replace_inspection([])
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert not _loop_backlog(loop).remediation
|
|
assert not _loop_backlog(loop).remediation
|
|
|
|
loop.context.topology.visited.add(root.location)
|
|
frontier = service_topology(
|
|
loop,
|
|
[root],
|
|
{root.location: "before-alpha"},
|
|
)
|
|
assert frontier == [mid]
|
|
loop.context.append_frontier(frontier)
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert not _loop_backlog(loop).remediation
|
|
assert not _loop_backlog(loop).remediation
|
|
|
|
_loop_backlog(loop).replace_inspection([])
|
|
|
|
frontier = service_topology(
|
|
loop,
|
|
[mid],
|
|
{mid.location: "before-alpha"},
|
|
)
|
|
assert frontier == [target]
|
|
loop.context.append_frontier(frontier)
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert _loop_backlog(loop).remediation
|
|
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert not loop.context.deferred_remediation_backlog
|
|
loop.aider_worker = NoChangeWorker()
|
|
|
|
_, source_index = loop.context.frontier
|
|
loop.context.active_topology = source_index
|
|
loop.context.persist()
|
|
alpha = loop._create_alpha_loop(source_index)
|
|
loop._run(alpha)
|
|
source_frontier, _ = loop.context.frontier
|
|
loop.context.topology.visited.update(
|
|
source.location
|
|
for source in source_frontier
|
|
)
|
|
loop.context.active_topology = None
|
|
loop.context.persist()
|
|
|
|
assert not _loop_backlog(loop).remediation
|
|
|
|
def test_frontier_expansion_for_deferred_remediation_does_not_mutate_project(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify frontier-only remediation shaping leaves project contents unchanged."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "zbar.py", "def zbar():\n return 1\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
blame = file_manager.require_src("src/zbar.py")
|
|
_freeze_dependency_edges(file_manager, [(root.location, blame.location)])
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(blame, test, summary="repair zbar")
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
deferred_remediation_backlog={blame.location: [remediation]},
|
|
)
|
|
context.file_manager = file_manager
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
)
|
|
loop = StabilizationLoop(
|
|
gate=gate,
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
original_source = file_manager.repo_exec.read_text(
|
|
file_manager.source_path(root.location)
|
|
)
|
|
original_test = file_manager.repo_exec.read_text(
|
|
file_manager.test_path(test.location)
|
|
)
|
|
_loop_backlog(loop).replace_inspection([])
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
frontier = service_topology(
|
|
loop,
|
|
[root],
|
|
{root.location: "before-alpha"},
|
|
)
|
|
assert frontier == [blame]
|
|
loop.context.append_frontier(frontier)
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
assert _loop_backlog(loop).remediation
|
|
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert file_manager.repo_exec.read_text(file_manager.source_path(root.location)) == original_source
|
|
assert file_manager.repo_exec.read_text(file_manager.test_path(test.location)) == original_test
|
|
|
|
|
|
def test_unavailable_forward_expansion_falls_through_to_runnable_alpha_and_beta_work(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify an unavailable forward frontier still falls through to runnable phase work."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "outside.py", "def outside():\n return 1\n")
|
|
file_manager.resync()
|
|
root = file_manager.require_src("src/foo.py")
|
|
outside = file_manager.require_src("src/outside.py")
|
|
_freeze_dependency_edges(file_manager, [(root.location, outside.location)])
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
current_remediation = _failed_item(root, test, summary="repair current source")
|
|
deferred_remediation = _failed_item(outside, test, summary="repair outside source")
|
|
testing = _testing_item(root, test)
|
|
backlog = BackLog(
|
|
remediation=[current_remediation],
|
|
testing=[testing],
|
|
)
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
total_cot=2,
|
|
backlog=backlog,
|
|
deferred_remediation_backlog={outside.location: [deferred_remediation]},
|
|
file_manager=_scoped_file_manager(file_manager),
|
|
)
|
|
loop = ScenarioStabilizationLoop(
|
|
gate=MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: True,
|
|
),
|
|
context=context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
dependency_strategy=PythonAstDependencyStrategy(),
|
|
),
|
|
)
|
|
events: list[str] = []
|
|
|
|
def run_source(_: int, source_backlog: ContributorBacklog) -> str:
|
|
"""Commit one runnable Alpha item before the declared Beta pass."""
|
|
|
|
assert not loop.context.blocked_summary
|
|
popped = source_backlog.pop_next_source_work()
|
|
assert popped is not None
|
|
events.append(f"alpha:{_focused_metadata(popped.value).location}")
|
|
with popped as scope:
|
|
scope.commit()
|
|
return "recorded"
|
|
|
|
def run_testing_pass() -> None:
|
|
assert not loop.context.blocked_summary
|
|
popped = _loop_backlog(loop).pop_testing()
|
|
assert popped is not None
|
|
assert popped.value is testing
|
|
events.append("beta")
|
|
with popped as scope:
|
|
scope.commit()
|
|
|
|
loop.set_source_runner(run_source)
|
|
loop.set_testing_runner(run_testing_pass)
|
|
|
|
loop.run()
|
|
|
|
assert events == [
|
|
"alpha:src/foo.py",
|
|
"beta",
|
|
"alpha:src/outside.py",
|
|
"alpha:src/outside.py",
|
|
]
|
|
assert current_remediation not in _loop_backlog(loop).remediation
|
|
assert testing not in _loop_backlog(loop).testing
|
|
assert loop.context.deferred_remediation_backlog == {}
|
|
assert not loop.context.blocked_summary
|
|
assert loop.context.accepted
|
|
|
|
|
|
def test_final_cot_runs_direct_remediation_outside_forward_topology(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A direct remediation outside forward topology remains runnable by Alpha."""
|
|
|
|
class RemediationClient(FakeClient):
|
|
"""Cite outside.py before the stationary editor replaces it with foo."""
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Ground initial remediation in the explicit pre-edit source."""
|
|
|
|
# This decision precedes the editor's replacement of the selected source.
|
|
if "SOURCE REMEDIATION GUIDANCE" in prompt:
|
|
return json.dumps({"work_needed": "Refresh metadata.", "reason": ["def outside():", "return 1"]})
|
|
return super().complete(prompt, logger)
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.client = RemediationClient()
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "outside.py", "def outside():\n return 1\n")
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/outside.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(source, test, summary="deferred repair")
|
|
backlog = BackLog()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
backlog,
|
|
FakeCoverageStrategy(1.0),
|
|
total_cot=1,
|
|
)
|
|
loop.context.deferred_remediation_backlog[source.location] = [remediation]
|
|
loop.run()
|
|
|
|
assert loop.context.blocked_summary == ""
|
|
assert loop.context.accepted
|
|
assert not loop.context.remaining_deferred_remediation.deferred_items
|
|
assert len(_scenario_file_manager(loop).squash_summaries) == 1
|
|
|
|
def test_stabilization_squashes_once_after_clear_backlogs_and_no_contextual_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify Stabilization squashes after backlog clears and no frontier remains."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(0.0),
|
|
)
|
|
loop.run()
|
|
|
|
assert loop.context.accepted
|
|
assert not loop.context.topology.deferred_topology_locations
|
|
assert not loop.context.remaining_deferred_remediation.deferred_items
|
|
assert len(_scenario_file_manager(loop).squash_summaries) == 1
|
|
|
|
|
|
def test_last_cot_exhaustion_schedules_final_exam(tmp_path: Path) -> None:
|
|
"""Verify final COT frontier exhaustion schedules the final exam before acceptance."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
cot_index=0,
|
|
total_cot=1,
|
|
)
|
|
loop._schedule_final_exam()
|
|
|
|
assert loop.context.final_exam_started is True
|
|
assert loop.context.final_exam_finished is False
|
|
assert loop.context.fully_explored is False
|
|
assert loop.context.accepted is False
|
|
assert len(_loop_backlog(loop).testing) == 1
|
|
assert _loop_backlog(loop).testing[0].last_exam is True
|
|
assert _loop_backlog(loop).testing[0].downstream_classes == []
|
|
|
|
|
|
def test_non_last_cot_exhaustion_accepts_without_final_exam(tmp_path: Path) -> None:
|
|
"""Verify non-final COT frontier exhaustion keeps the prior completion behavior."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
cot_index=0,
|
|
total_cot=2,
|
|
)
|
|
loop.run()
|
|
|
|
assert loop.context.final_exam_started is False
|
|
assert loop.context.final_exam_finished is False
|
|
assert loop.context.fully_explored is True
|
|
assert loop.context.accepted is True
|
|
|
|
|
|
def test_final_exam_finishes_when_frontier_exhausted_again_with_empty_backlog(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify final exam completion is marked after the last exam leaves no work."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
loop = _stabilization_loop(
|
|
file_manager,
|
|
BackLog(),
|
|
FakeCoverageStrategy(1.0),
|
|
cot_index=0,
|
|
total_cot=1,
|
|
)
|
|
loop.context.final_exam_started = True
|
|
loop.run()
|
|
|
|
assert loop.context.final_exam_finished is True
|
|
assert loop.context.fully_explored is True
|
|
assert loop.context.accepted is True
|
|
|
|
|
|
def test_final_exam_remediation_does_not_defer_outside_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify final exam blocks the deferred-remediation lane."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "outside.py", "def outside():\n return 1\n")
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/outside.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(source, test, summary="repair outside")
|
|
backlog = BackLog()
|
|
backlog.queue_remediation(remediation)
|
|
loop = _stabilization_loop(file_manager, backlog, FakeCoverageStrategy(1.0))
|
|
loop.context.final_exam_started = True
|
|
|
|
loop._defer_forward_remediation(loop._current_cot_forward_remediation_locations())
|
|
|
|
assert _loop_backlog(loop).remediation == (remediation,)
|
|
assert loop.context.deferred_remediation_backlog == {}
|
|
|
|
|
|
def test_final_exam_remediation_does_not_synthesize_upstream_context(tmp_path: Path) -> None:
|
|
"""Verify final exam remediation keeps only preserved topology context."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "bar.py", "def bar():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "baz.py", "def baz():\n return 1\n")
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
remediation = _failed_item(source, test, summary="repair foo")
|
|
backlog = BackLog()
|
|
backlog.queue_remediation(remediation)
|
|
loop = _stabilization_loop(file_manager, backlog, FakeCoverageStrategy(1.0))
|
|
loop.context.final_exam_started = True
|
|
_scenario_file_manager(loop).set_diff(
|
|
"src/bar.py",
|
|
"diff --git a/src/bar.py b/src/bar.py",
|
|
)
|
|
|
|
forward_locations = loop._current_cot_forward_remediation_locations()
|
|
loop._promote_deferred_remediation(forward_locations)
|
|
loop._defer_forward_remediation(forward_locations)
|
|
loop.context.persist()
|
|
|
|
assert remediation.upstream_classes == []
|
|
|
|
|
|
def test_root_surrogate_testing_item_uses_current_graph_upstream(tmp_path: Path) -> None:
|
|
"""Verify surrogate-root QA backlog keeps current-graph upstream source context."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "base.py", "def base():\n return 1\n")
|
|
file_manager.repo_exec.write_text(file_manager.test_root / "test_new_file.py",
|
|
"from new_file import new_value\n\n"
|
|
"def test_new_value():\n"
|
|
" assert new_value() == 1\n"
|
|
)
|
|
file_manager.resync()
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "new_file.py",
|
|
"from base import base\n\n"
|
|
"def new_value():\n"
|
|
" return base()\n"
|
|
)
|
|
backlog = BackLog()
|
|
loop = _stabilization_loop(file_manager, backlog, FakeCoverageStrategy(1.0))
|
|
|
|
assert loop._root_surrogates_needed_callback({"src/new_file.py": "create root surrogate"})
|
|
|
|
assert len(backlog.testing) == 1
|
|
testing_item = backlog.testing[0]
|
|
assert [
|
|
upstream.location
|
|
for upstream in testing_item.upstream_classes
|
|
] == ["src/base.py"]
|
|
assert [
|
|
downstream.location
|
|
for downstream in testing_item.downstream_classes
|
|
] == ["src/new_file.py"]
|
|
|
|
|
|
def test_hydration_resets_unfinished_final_exam(tmp_path: Path) -> None:
|
|
"""Verify resume repeats an interrupted final exam from a clean queue state."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.resync()
|
|
source = file_manager.require_src("src/foo.py")
|
|
test = file_manager.require_test("tests/test_foo.py")
|
|
backlog = BackLog()
|
|
backlog.queue_testing(
|
|
_final_exam_item(test)
|
|
)
|
|
context = StabilizationContext(
|
|
goal="goal",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=source,
|
|
completion=backlog.completion,
|
|
backlog=backlog,
|
|
final_exam_started=True,
|
|
final_exam_finished=False,
|
|
fully_explored=False,
|
|
deferred_remediation_backlog={source.location: []},
|
|
)
|
|
|
|
StabilizationContextReassembly().post_thaw(context)
|
|
|
|
assert context.final_exam_started is False
|
|
assert context.final_exam_finished is False
|
|
assert context.fully_explored is False
|
|
assert _context_backlog(context).testing == ()
|
|
assert _context_backlog(context).remediation == ()
|
|
assert _context_backlog(context).inspection == ()
|
|
assert context.deferred_remediation_backlog == {source.location: []}
|