1355 lines
48 KiB
Python
1355 lines
48 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Behavioral coverage for COT ownership, remediation handoff, and predecessor evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
from band_aider.context.alpha_context import AlphaContext
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.global_context import GlobalContext, StoryCursor
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.context.working_backlog import (
|
|
BackLog,
|
|
DeferredRemediationHandoff,
|
|
FailedQAItem,
|
|
RemediationDeduplicationKey,
|
|
TestingBacklog as _TestingBacklog,
|
|
)
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import (
|
|
BiDependencyDiscoveryStrategy,
|
|
)
|
|
from band_aider.judgement.code_dependency_graph import (
|
|
CodeDependencyGraph,
|
|
DependencySelector,
|
|
)
|
|
from band_aider.judgement.dependency_discovery_strategy import (
|
|
DependencyDirection,
|
|
DependencyTraversalCache,
|
|
)
|
|
from band_aider.judgement.frozen_code_dependency_graph_maintainer import (
|
|
FrozenCodeDependencyGraphMaintainer,
|
|
)
|
|
from band_aider.judgement.python_ast_dependency_strategy import (
|
|
PythonAstDependencyStrategy,
|
|
)
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
from band_aider.loop.mechanical_loop_gate import MechanicalLoopGate
|
|
from tests.band_aider_loop_impl import (
|
|
BackLog,
|
|
FakeCoverageStrategy,
|
|
FailingTest,
|
|
PassingRunner,
|
|
_file_manager,
|
|
_language_strategy_set,
|
|
)
|
|
from tests.topology_test_support import service_topology
|
|
|
|
|
|
def _add_sources(file_manager, names: list[str]) -> dict[str, ClassMeta]:
|
|
"""Add minimal source files and return their current metadata by location."""
|
|
|
|
# Materialize only the graph locations required by this ownership example.
|
|
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: dict[str, ClassMeta] = {}
|
|
# Resolve every requested location through FileManager's authoritative index.
|
|
for name in names:
|
|
location = f"src/{name}"
|
|
source = file_manager.current_src(location)
|
|
assert source is not None
|
|
sources[location] = source
|
|
return sources
|
|
|
|
|
|
def _freeze_edges(
|
|
file_manager,
|
|
sources: dict[str, ClassMeta],
|
|
edges: list[tuple[str, str]],
|
|
) -> None:
|
|
"""Install an explicit immutable downstream graph for one COT ownership example."""
|
|
|
|
dependencies: dict[str, set[str]] = {}
|
|
dependents: dict[str, set[str]] = {}
|
|
# Encode each upstream-to-downstream relationship in the graph's native directions.
|
|
for upstream, downstream in edges:
|
|
dependents.setdefault(upstream, set()).add(downstream)
|
|
dependencies.setdefault(downstream, set()).add(upstream)
|
|
graph = CodeDependencyGraph(
|
|
source_by_location=dict(sources),
|
|
source_dependencies=dependencies,
|
|
source_dependents=dependents,
|
|
)
|
|
graph.freeze()
|
|
file_manager._frozen = graph
|
|
|
|
|
|
def _failed_remediation(source: ClassMeta) -> FailedQAItem:
|
|
"""Return one source-owned remediation with complete failure evidence."""
|
|
|
|
test = ClassMeta(
|
|
location="tests/test_example.py",
|
|
summary="example test",
|
|
relevance=[],
|
|
relevant_symbols={},
|
|
)
|
|
failure = FailingTest(
|
|
test=test,
|
|
error_message="removed lifecycle method",
|
|
stack_trace=["trace line one", "trace line two"],
|
|
raw_output="raw pytest output",
|
|
)
|
|
return FailedQAItem(
|
|
focused_spotlight=source,
|
|
summary="Repair the removed lifecycle call.",
|
|
upstream_classes=[],
|
|
downstream_classes=[source],
|
|
original_cot="Implement the predecessor lifecycle.",
|
|
failure_error_message=failure.error_message,
|
|
failure_stack_trace=list(failure.stack_trace),
|
|
failure_raw_output=failure.raw_output,
|
|
failure_context_sources=[source],
|
|
failed_test=failure,
|
|
)
|
|
|
|
|
|
class _HandoffContext(GlobalContext):
|
|
"""Provide the parent-owned durable cursor without creating host-backed state."""
|
|
|
|
def __init__(self, records: list[CotTaskRecord]) -> None:
|
|
"""Initialize only the parent state exercised by remediation handoff tests."""
|
|
|
|
self.goal = "handoff"
|
|
self.original_prompt = "handoff"
|
|
self.story_cursor = StoryCursor(cot_task_records=records)
|
|
self.relevant_files = []
|
|
self._file_manager = None
|
|
|
|
def persist(self, keep: int | None = None) -> Path:
|
|
"""Record no checkpoint because each handoff fixture is already in memory."""
|
|
|
|
del keep
|
|
return Path("handoff-checkpoint")
|
|
|
|
def record_stabilization_result(
|
|
self,
|
|
files: list[ClassMeta],
|
|
summary: str,
|
|
) -> None:
|
|
"""Keep this cursor test focused on parent handoff state transitions."""
|
|
|
|
del files, summary
|
|
|
|
|
|
def _handoff_context(records: list[CotTaskRecord]) -> _HandoffContext:
|
|
"""Return the smallest parent-owned cursor fixture for live COT handoff tests."""
|
|
|
|
return _HandoffContext(records)
|
|
|
|
|
|
def test_frozen_future_closure_reserves_the_root_and_all_downstream_sources(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A future root reserves its frozen downstream closure but never the current root."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py"])
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[
|
|
("src/a.py", "src/b.py"),
|
|
("src/b.py", "src/c.py"),
|
|
],
|
|
)
|
|
|
|
closure = file_manager.frozen_downstream_closure_locations([sources["src/b.py"]])
|
|
reservation = closure - {sources["src/a.py"].location}
|
|
|
|
assert closure == {"src/b.py", "src/c.py"}
|
|
assert reservation == {"src/b.py", "src/c.py"}
|
|
|
|
|
|
def test_stabilization_uses_a_snapshot_of_later_roots_for_frozen_reservation(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The parent supplies later roots once while Stabilization computes frozen ownership."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py"])
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[
|
|
("src/a.py", "src/b.py"),
|
|
("src/b.py", "src/c.py"),
|
|
],
|
|
)
|
|
parent = _handoff_context(
|
|
[
|
|
CotTaskRecord("A", "src/a.py", sources["src/a.py"]),
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
]
|
|
)
|
|
parent._file_manager = file_manager
|
|
|
|
loop = _ownership_loop(
|
|
file_manager,
|
|
sources["src/a.py"],
|
|
_TopologyStrategy({}),
|
|
future_cot_records=parent.future_cot_records_for_current_cot(),
|
|
)
|
|
|
|
assert parent.future_cot_records_for_current_cot() == [
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
]
|
|
assert loop.future_cot_records == parent.future_cot_records_for_current_cot()
|
|
assert loop.context.topology.future_cot_owned_locations == {
|
|
"src/b.py",
|
|
"src/c.py",
|
|
}
|
|
|
|
|
|
def test_future_closure_exempts_the_current_root_when_a_cycle_reaches_it(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A future cyclic closure cannot blacklist the current explicit root."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[
|
|
("src/a.py", "src/b.py"),
|
|
("src/b.py", "src/a.py"),
|
|
],
|
|
)
|
|
|
|
reservation = (
|
|
file_manager.frozen_downstream_closure_locations([sources["src/b.py"]])
|
|
- {sources["src/a.py"].location}
|
|
)
|
|
|
|
assert reservation == {"src/b.py"}
|
|
|
|
|
|
def test_multiple_future_roots_union_their_frozen_downstream_closures(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Every later COT root contributes its own frozen source territory."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(
|
|
file_manager,
|
|
["a.py", "x.py", "y.py", "b.py", "c.py", "z.py"],
|
|
)
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[
|
|
("src/a.py", "src/x.py"),
|
|
("src/x.py", "src/b.py"),
|
|
("src/b.py", "src/z.py"),
|
|
("src/a.py", "src/y.py"),
|
|
("src/y.py", "src/c.py"),
|
|
],
|
|
)
|
|
|
|
reservation = file_manager.frozen_downstream_closure_locations(
|
|
[sources["src/b.py"], sources["src/c.py"]]
|
|
) - {sources["src/a.py"].location}
|
|
|
|
assert reservation == {"src/b.py", "src/c.py", "src/z.py"}
|
|
|
|
|
|
def test_future_closure_reserves_a_shared_downstream_on_an_alternate_path(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A shared target remains reserved even when the current root has another path to it."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "x.py", "b.py", "z.py"])
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[
|
|
("src/a.py", "src/x.py"),
|
|
("src/x.py", "src/z.py"),
|
|
("src/a.py", "src/b.py"),
|
|
("src/b.py", "src/z.py"),
|
|
],
|
|
)
|
|
|
|
reservation = file_manager.frozen_downstream_closure_locations(
|
|
[sources["src/b.py"]]
|
|
) - {sources["src/a.py"].location}
|
|
|
|
assert reservation == {"src/b.py", "src/z.py"}
|
|
|
|
|
|
def test_frozen_future_closure_is_independent_of_graph_edge_insertion_order(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Frozen traversal returns the same ownership set for equivalent graph construction orders."""
|
|
|
|
first_root = tmp_path / "first"
|
|
second_root = tmp_path / "second"
|
|
first = _file_manager(first_root)
|
|
second = _file_manager(second_root)
|
|
first_sources = _add_sources(first, ["a.py", "b.py", "c.py", "d.py"])
|
|
second_sources = _add_sources(second, ["a.py", "b.py", "c.py", "d.py"])
|
|
edges = [
|
|
("src/a.py", "src/b.py"),
|
|
("src/b.py", "src/c.py"),
|
|
("src/b.py", "src/d.py"),
|
|
]
|
|
_freeze_edges(first, first_sources, edges)
|
|
_freeze_edges(second, second_sources, list(reversed(edges)))
|
|
|
|
first_result = first.frozen_downstream_closure_locations([first_sources["src/b.py"]])
|
|
second_result = second.frozen_downstream_closure_locations([second_sources["src/b.py"]])
|
|
|
|
assert first_result == second_result == {"src/b.py", "src/c.py", "src/d.py"}
|
|
|
|
|
|
class _TopologyStrategy(PythonAstDependencyStrategy, BiDependencyDiscoveryStrategy):
|
|
"""Provide a deterministic graph that isolates Stabilization admission.
|
|
|
|
Example: ``_TopologyStrategy({root.location: [child]})`` makes ``child``
|
|
the sole raw downstream discovery from ``root``.
|
|
"""
|
|
|
|
def __init__(self, downstream_by_location: dict[str, list[ClassMeta]]) -> None:
|
|
"""Store one direct downstream answer per source location."""
|
|
|
|
self.downstream_by_location = downstream_by_location
|
|
source_by_location = {
|
|
source.location: source
|
|
for sources in downstream_by_location.values()
|
|
for source in sources
|
|
}
|
|
source_by_location.update(
|
|
{
|
|
location: source
|
|
for location, sources in downstream_by_location.items()
|
|
for source in sources
|
|
if source.location == location
|
|
}
|
|
)
|
|
dependencies = {
|
|
source.location: {upstream}
|
|
for upstream, sources in downstream_by_location.items()
|
|
for source in sources
|
|
}
|
|
dependents = {
|
|
upstream: {source.location for source in sources}
|
|
for upstream, sources in downstream_by_location.items()
|
|
}
|
|
graph = CodeDependencyGraph(
|
|
source_by_location=source_by_location,
|
|
source_dependencies=dependencies,
|
|
source_dependents=dependents,
|
|
)
|
|
super().__init__(FrozenCodeDependencyGraphMaintainer(graph))
|
|
|
|
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 the direct downstream candidates configured for the supplied source wave."""
|
|
|
|
del all_source_files, src_root, recursive, dependency_cache, selector
|
|
result: dict[str, ClassMeta] = {}
|
|
# Combine each source's known direct downstream candidates without duplicates.
|
|
for source in source_files:
|
|
for downstream in self.downstream_by_location.get(source.location, []):
|
|
result.setdefault(downstream.location, downstream)
|
|
return [result[location] for location in sorted(result)]
|
|
|
|
def source_dependencies_of(
|
|
self,
|
|
source_location: str,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> set[str]:
|
|
"""Return direct prerequisites derived from the configured downstream index."""
|
|
|
|
del selector
|
|
return {
|
|
upstream
|
|
for upstream, downstream in self.downstream_by_location.items()
|
|
if source_location in {source.location for source in downstream}
|
|
}
|
|
|
|
def source_dependents_of(
|
|
self,
|
|
source_location: str,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> set[str]:
|
|
"""Return direct dependent locations from the configured downstream index."""
|
|
|
|
del selector
|
|
return {
|
|
source.location
|
|
for source in self.downstream_by_location.get(source_location, [])
|
|
}
|
|
|
|
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 no prerequisites so this test isolates the ownership fence."""
|
|
|
|
del files, all_source_files, src_root, test_root, recursive, dependency_cache, selector
|
|
return []
|
|
|
|
def has_path(
|
|
self,
|
|
start: str,
|
|
target: str,
|
|
direction: DependencyDirection = DependencyDirection.UPSTREAM,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
excluded_locations: frozenset[str] = frozenset(),
|
|
) -> bool:
|
|
"""Report no extra path constraints for this admission-only test graph."""
|
|
|
|
del start, target, direction, selector, excluded_locations
|
|
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 deferred beam candidates for this ordinary-frontier example."""
|
|
|
|
del current_frontier, deferred_blames, all_source_files, src_root
|
|
return []
|
|
|
|
|
|
class _DeferredBeamStrategy(_TopologyStrategy):
|
|
"""Record deferred discovery inputs while returning one explicit whitelist relay."""
|
|
|
|
def __init__(
|
|
self,
|
|
downstream_by_location: dict[str, list[ClassMeta]],
|
|
whitelist: list[ClassMeta],
|
|
) -> None:
|
|
"""Store deterministic discovery answers and their observed inputs."""
|
|
|
|
super().__init__(downstream_by_location)
|
|
self.whitelist = whitelist
|
|
self.joint_beam_target_locations: list[str] = []
|
|
self.downstream_seed_locations: list[str] = []
|
|
self.downstream_calls: list[tuple[list[str], bool]] = []
|
|
|
|
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]:
|
|
"""Record the unfiltered seed wave before returning its configured candidates."""
|
|
|
|
self.downstream_seed_locations = sorted(
|
|
source.location
|
|
for source in source_files
|
|
)
|
|
self.downstream_calls.append((self.downstream_seed_locations, recursive))
|
|
return super().downstream_sources_for(
|
|
source_files,
|
|
all_source_files,
|
|
src_root,
|
|
recursive,
|
|
dependency_cache,
|
|
selector,
|
|
)
|
|
|
|
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]:
|
|
"""Record every deferred target considered by normal beam discovery."""
|
|
|
|
del current_frontier, all_source_files, src_root
|
|
self.joint_beam_target_locations = sorted(
|
|
source.location
|
|
for source in deferred_blames
|
|
)
|
|
return list(self.whitelist)
|
|
|
|
|
|
class _DrainSourceWork(AlphaLoop):
|
|
"""Drain one public source-work backlog scope and optionally record its item.
|
|
|
|
Example: ``_DrainSourceWork(backlog).run()`` commits the next Alpha-owned
|
|
remediation or inspection through the same public claim scope as production.
|
|
"""
|
|
|
|
def __init__(self, backlog: BackLog, handled: list[object] | None = None) -> None:
|
|
"""Retain the owned backlog and optional record of committed work."""
|
|
|
|
self.backlog = backlog
|
|
self.handled = handled
|
|
alpha_context = object.__new__(AlphaContext)
|
|
alpha_context.goal = "drain source work"
|
|
self.context = alpha_context
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Commit the next remediation-or-inspection item selected by Alpha policy."""
|
|
|
|
del logger_name
|
|
popped = self.backlog.pop_remediation() or self.backlog.pop_inspection()
|
|
assert popped is not None
|
|
with popped as scope:
|
|
# Preserve optional evidence without changing the public claim lifecycle.
|
|
if self.handled is not None:
|
|
self.handled.append(scope.value)
|
|
scope.commit()
|
|
return "drained"
|
|
|
|
|
|
def _ownership_loop(
|
|
file_manager,
|
|
root: ClassMeta,
|
|
strategy: _TopologyStrategy,
|
|
future_cot_records: list[CotTaskRecord] | None = None,
|
|
) -> StabilizationLoop:
|
|
"""Create one stabilization loop with its live topology strategy isolated for testing."""
|
|
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
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),
|
|
),
|
|
future_cot_records=future_cot_records,
|
|
)
|
|
loop.bi_dependency_strategy = strategy
|
|
loop._ensure_strategies = lambda: None
|
|
loop.context.topology.bind_runtime(strategy, loop.logger)
|
|
return loop
|
|
|
|
|
|
def test_future_ownership_filters_only_the_computed_frontier(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Future sources remain visible to deferred discovery but cannot enter the current frontier."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "x.py", "z.py"])
|
|
root = sources["src/a.py"]
|
|
future_root = sources["src/b.py"]
|
|
ordinary = sources["src/x.py"]
|
|
future_descendant = sources["src/z.py"]
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[(future_root.location, future_descendant.location)],
|
|
)
|
|
strategy = _DeferredBeamStrategy(
|
|
{
|
|
root.location: [ordinary],
|
|
future_root.location: [future_descendant],
|
|
},
|
|
[future_root],
|
|
)
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
deferred_remediation_backlog={
|
|
future_root.location: [_failed_remediation(future_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),
|
|
),
|
|
future_cot_records=[
|
|
CotTaskRecord("future", future_root.location, future_root),
|
|
],
|
|
)
|
|
loop.bi_dependency_strategy = strategy
|
|
loop._ensure_strategies = lambda: None
|
|
loop.context.topology.bind_runtime(strategy, loop.logger)
|
|
|
|
loop._grow_deferred_frontier_whitelist_locations([root])
|
|
context.topology.visited.update({root.location, future_root.location})
|
|
next_frontier = service_topology(loop, [root], {})
|
|
|
|
assert loop.context.topology.future_cot_owned_locations == {
|
|
future_root.location,
|
|
future_descendant.location,
|
|
}
|
|
assert strategy.joint_beam_target_locations == [future_root.location]
|
|
assert strategy.downstream_calls == [
|
|
([root.location, future_root.location], False),
|
|
]
|
|
assert [source.location for source in next_frontier] == [ordinary.location]
|
|
assert future_descendant.location not in {
|
|
source.location
|
|
for source in next_frontier
|
|
}
|
|
|
|
|
|
def test_reserved_deferred_topology_is_filtered_at_candidate_admission(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Queue cleanup preserves deferred topology until the normal candidate set applies ownership."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
root = sources["src/a.py"]
|
|
future_root = sources["src/b.py"]
|
|
_freeze_edges(file_manager, sources, [])
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
topology=TopologicalLiveliness(
|
|
deferred_topology_locations={future_root.location},
|
|
),
|
|
)
|
|
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),
|
|
),
|
|
future_cot_records=[
|
|
CotTaskRecord("future", future_root.location, future_root),
|
|
],
|
|
)
|
|
loop.bi_dependency_strategy = _TopologyStrategy({})
|
|
loop._ensure_strategies = lambda: None
|
|
loop.context.topology.bind_runtime(loop.bi_dependency_strategy, loop.logger)
|
|
|
|
assert context.topology.deferred_topology_locations == {future_root.location}
|
|
assert service_topology(loop, [root], {}) == []
|
|
assert context.topology.deferred_topology_locations == set()
|
|
|
|
|
|
def test_remaining_deferred_remediation_preserves_separate_staging_for_parent_handoff(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The parent can read both deferred lanes without taking ownership of either lane."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["b.py", "c.py", "local.py"])
|
|
root = sources["src/b.py"]
|
|
future_root = sources["src/c.py"]
|
|
local_source = sources["src/local.py"]
|
|
_freeze_edges(file_manager, sources, [])
|
|
inherited = _failed_remediation(future_root)
|
|
local = _failed_remediation(local_source)
|
|
context = StabilizationContext(
|
|
goal="relay COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
inherited_remediation_backlog={future_root.location: [inherited]},
|
|
deferred_remediation_backlog={local_source.location: [local]},
|
|
)
|
|
|
|
payload = context.__getstate__()
|
|
assert payload["inherited_remediation_backlog"] == {
|
|
future_root.location: [inherited],
|
|
}
|
|
assert payload["deferred_remediation_backlog"] == {
|
|
local_source.location: [local],
|
|
}
|
|
handoff = context.remaining_deferred_remediation
|
|
assert isinstance(handoff, DeferredRemediationHandoff)
|
|
assert handoff.inherited_items == [inherited]
|
|
assert handoff.deferred_items == [local]
|
|
assert context.inherited_remediation_backlog == {
|
|
future_root.location: [inherited],
|
|
}
|
|
assert context.deferred_remediation_backlog == {
|
|
local_source.location: [local],
|
|
}
|
|
|
|
|
|
def test_frontier_fence_skips_reserved_sources_but_keeps_ordinary_downstream_work(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Future-owned candidates never join current topology while ordinary candidates still do."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "ordinary.py"])
|
|
root = sources["src/a.py"]
|
|
reserved = sources["src/b.py"]
|
|
ordinary = sources["src/ordinary.py"]
|
|
_freeze_edges(file_manager, sources, [(root.location, reserved.location)])
|
|
loop = _ownership_loop(
|
|
file_manager,
|
|
root,
|
|
_TopologyStrategy({root.location: [reserved, ordinary]}),
|
|
future_cot_records=[
|
|
CotTaskRecord("future", reserved.location, reserved),
|
|
],
|
|
)
|
|
loop.context.topology.visited.add(root.location)
|
|
|
|
next_frontier = service_topology(loop, [root], {root.location: "before"})
|
|
|
|
assert [source.location for source in next_frontier] == [ordinary.location]
|
|
loop.context.append_frontier(next_frontier)
|
|
assert [source.location for source in loop.context.full_topologies[-1]] == [ordinary.location]
|
|
assert reserved.location not in {
|
|
source.location
|
|
for frontier in loop.context.full_topologies
|
|
for source in frontier
|
|
}
|
|
|
|
|
|
def test_current_explicit_root_remains_eligible_when_a_future_cycle_includes_it(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The current root can remain frontier-zero even when a future closure reaches it."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
root = sources["src/a.py"]
|
|
future = sources["src/b.py"]
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[(root.location, future.location), (future.location, root.location)],
|
|
)
|
|
loop = _ownership_loop(
|
|
file_manager,
|
|
root,
|
|
_TopologyStrategy({root.location: [future]}),
|
|
future_cot_records=[
|
|
CotTaskRecord("future", future.location, future),
|
|
],
|
|
)
|
|
|
|
assert [source.location for source in loop.context.t0] == [root.location]
|
|
assert root.location not in loop.context.topology.future_cot_owned_locations
|
|
assert not service_topology(loop, [root], {root.location: "before"})
|
|
|
|
|
|
def test_remediation_deduplication_key_names_each_failure_evidence_dimension(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Remediation deduplication keeps its evidence dimensions explicit rather than positional."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["source.py", "upstream.py", "downstream.py"])
|
|
remediation = _failed_remediation(sources["src/source.py"])
|
|
remediation.upstream_classes = [sources["src/upstream.py"]]
|
|
remediation.downstream_classes = [sources["src/downstream.py"]]
|
|
|
|
identity = RemediationDeduplicationKey.from_item(remediation)
|
|
|
|
assert identity.focused_location == "src/source.py"
|
|
assert identity.upstream_locations == ["src/upstream.py"]
|
|
assert identity.downstream_locations == ["src/downstream.py"]
|
|
assert identity.failure_context_locations == ["src/source.py"]
|
|
assert identity.failed_test_stack_trace == ["trace line one", "trace line two"]
|
|
assert identity.failure_stack_trace == ["trace line one", "trace line two"]
|
|
|
|
|
|
def test_future_owned_remediation_is_durably_relayed_without_losing_evidence(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A future source failure moves through each next-COT inbox unchanged and without duplicates."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py"])
|
|
records = [
|
|
CotTaskRecord("A", "src/a.py", sources["src/a.py"]),
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
CotTaskRecord("C", "src/c.py", sources["src/c.py"]),
|
|
]
|
|
parent = _handoff_context(records)
|
|
remediation = _failed_remediation(sources["src/c.py"])
|
|
|
|
handoff = DeferredRemediationHandoff(deferred_items=[remediation])
|
|
assert parent.handoff_deferred_remediation_to_next_cot(handoff)
|
|
assert parent.handoff_deferred_remediation_to_next_cot(handoff)
|
|
parent.story_cursor.task_cursor.current_cot_index = 1
|
|
inherited_by_b = parent.inherited_deferred_remediation_for_current_cot()
|
|
assert inherited_by_b == {"src/c.py": [remediation]}
|
|
|
|
assert parent.handoff_deferred_remediation_to_next_cot(
|
|
DeferredRemediationHandoff(
|
|
inherited_items=inherited_by_b["src/c.py"],
|
|
)
|
|
)
|
|
parent.story_cursor.task_cursor.current_cot_index = 2
|
|
inherited_by_c = parent.inherited_deferred_remediation_for_current_cot()
|
|
|
|
assert inherited_by_c == {"src/c.py": [remediation]}
|
|
preserved = inherited_by_c["src/c.py"][0]
|
|
assert preserved.failed_test == remediation.failed_test
|
|
assert preserved.failure_error_message == "removed lifecycle method"
|
|
assert preserved.failure_stack_trace == ["trace line one", "trace line two"]
|
|
assert preserved.failure_raw_output == "raw pytest output"
|
|
assert preserved.original_cot == "Implement the predecessor lifecycle."
|
|
assert preserved.failure_context_sources == [sources["src/c.py"]]
|
|
|
|
|
|
def test_stabilized_cot_acknowledges_its_parent_remediation_inbox(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Completing inherited work consumes only the current COT's durable inbox."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py"])
|
|
records = [
|
|
CotTaskRecord("A", "src/a.py", sources["src/a.py"]),
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
CotTaskRecord("C", "src/c.py", sources["src/c.py"]),
|
|
]
|
|
parent = _handoff_context(records)
|
|
remediation = _failed_remediation(sources["src/b.py"])
|
|
later_remediation = _failed_remediation(sources["src/c.py"])
|
|
parent.story_cursor.task_cursor.current_cot_index = 1
|
|
parent.story_cursor.task_cursor.deferred_remediation_by_cot = {
|
|
"1": [remediation],
|
|
"2": [later_remediation],
|
|
}
|
|
parent.mark_root_stabilized(records[1].root, "resolved")
|
|
|
|
assert "1" not in parent.story_cursor.task_cursor.deferred_remediation_by_cot
|
|
assert parent.story_cursor.task_cursor.deferred_remediation_by_cot["2"] == [later_remediation]
|
|
|
|
|
|
def test_handoff_rejects_testing_backlog_items_without_creating_next_cot_work(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Only FailedQAItem remediation may cross the parent-owned COT handoff seam."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
records = [
|
|
CotTaskRecord("A", "src/a.py", sources["src/a.py"]),
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
]
|
|
parent = _handoff_context(records)
|
|
focused_test = file_manager.current_test("tests/test_foo.py")
|
|
assert focused_test is not None
|
|
testing = _TestingBacklog(
|
|
focused_spotlight=focused_test,
|
|
summary="Add current-COT regression coverage.",
|
|
downstream_classes=[sources["src/a.py"]],
|
|
)
|
|
|
|
assert not parent.handoff_deferred_remediation_to_next_cot(
|
|
DeferredRemediationHandoff(
|
|
deferred_items=cast(list[FailedQAItem], [testing]),
|
|
)
|
|
)
|
|
assert parent.story_cursor.task_cursor.deferred_remediation_by_cot == {}
|
|
|
|
|
|
def test_next_cot_uses_only_the_immediately_previous_root(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Sequential COT flow makes the immediate preceding root the only readonly predecessor."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py", "incidental.py"])
|
|
records = [
|
|
CotTaskRecord("A", "src/a.py", sources["src/a.py"]),
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
CotTaskRecord("C", "src/c.py", sources["src/c.py"]),
|
|
]
|
|
parent = _handoff_context(records)
|
|
|
|
parent.story_cursor.task_cursor.current_cot_index = 1
|
|
assert parent.previous_cot_root_location_for_current_cot() == "src/a.py"
|
|
parent.story_cursor.task_cursor.current_cot_index = 2
|
|
assert parent.previous_cot_root_location_for_current_cot() == "src/b.py"
|
|
|
|
|
|
def test_later_cot_keeps_every_earlier_cot_as_ordered_context_candidate(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Deferral across two COTs retains both earlier contracts for turn selection."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["memento.py", "alpha.py", "patcher.py"])
|
|
records = [
|
|
CotTaskRecord(
|
|
"Remove Memento.clear; stale callers must stop using it.",
|
|
"src/memento.py",
|
|
sources["src/memento.py"],
|
|
),
|
|
CotTaskRecord(
|
|
"Update Alpha without restoring removed lifecycle APIs.",
|
|
"src/alpha.py",
|
|
sources["src/alpha.py"],
|
|
),
|
|
CotTaskRecord(
|
|
"Repair the cached file patcher.",
|
|
"src/patcher.py",
|
|
sources["src/patcher.py"],
|
|
),
|
|
]
|
|
parent = _handoff_context(records)
|
|
parent.story_cursor.task_cursor.current_cot_index = 2
|
|
|
|
candidates = parent.previous_cot_records_for_current_cot()
|
|
|
|
assert candidates == tuple(records[:2])
|
|
assert candidates[0].task == "Remove Memento.clear; stale callers must stop using it."
|
|
|
|
|
|
def test_parent_predecessor_root_is_not_stabilization_context_state(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The child persists live remediation, not a predecessor root owned by the parent."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
remediation = _failed_remediation(sources["src/b.py"])
|
|
original = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=sources["src/b.py"],
|
|
deferred_remediation_backlog={"src/b.py": [remediation]},
|
|
)
|
|
payload = original.__getstate__()
|
|
|
|
assert not hasattr(original, "previous_cot_root_location")
|
|
assert payload["deferred_remediation_backlog"] == {"src/b.py": [remediation]}
|
|
|
|
|
|
def test_unattributed_inherited_remediation_remains_durable_and_blocking(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""An ambiguous handoff item cannot disappear merely because no source owner is known."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
records = [
|
|
CotTaskRecord("A", "src/a.py", sources["src/a.py"]),
|
|
CotTaskRecord("B", "src/b.py", sources["src/b.py"]),
|
|
]
|
|
parent = _handoff_context(records)
|
|
parent.story_cursor.task_cursor.current_cot_index = 1
|
|
remediation = _failed_remediation(sources["src/b.py"])
|
|
remediation.focused_spotlight = None
|
|
parent.story_cursor.task_cursor.deferred_remediation_by_cot = {"1": [remediation]}
|
|
|
|
inherited = parent.inherited_deferred_remediation_for_current_cot()
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=sources["src/b.py"],
|
|
inherited_remediation_backlog=inherited,
|
|
)
|
|
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),
|
|
),
|
|
)
|
|
|
|
assert list(inherited.values()) == [[remediation]]
|
|
assert context.remaining_deferred_remediation.inherited_items == [remediation]
|
|
assert not context.accepted
|
|
assert context.backlog is not None
|
|
assert not context.backlog.has_ordinary_work()
|
|
|
|
|
|
def test_inherited_later_owner_remediation_stays_deferred_until_its_own_cot(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A relay COT cannot activate work that the next future COT still owns."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py"])
|
|
remediation = _failed_remediation(sources["src/c.py"])
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[(sources["src/b.py"].location, sources["src/c.py"].location)],
|
|
)
|
|
relay_context = StabilizationContext(
|
|
goal="B COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=sources["src/b.py"],
|
|
inherited_remediation_backlog={"src/c.py": [remediation]},
|
|
)
|
|
relay = StabilizationLoop(
|
|
gate=MechanicalLoopGate(relay_context, lambda: True, lambda: True),
|
|
context=relay_context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
),
|
|
future_cot_records=[
|
|
CotTaskRecord("C", "src/c.py", sources["src/c.py"]),
|
|
],
|
|
)
|
|
|
|
relay._promote_deferred_remediation(relay._current_cot_forward_remediation_locations())
|
|
|
|
assert relay.context.backlog is not None
|
|
assert relay.context.backlog.remediation == ()
|
|
assert relay.context.inherited_remediation_backlog == {"src/c.py": [remediation]}
|
|
|
|
owner_context = StabilizationContext(
|
|
goal="C COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=sources["src/c.py"],
|
|
inherited_remediation_backlog={"src/c.py": [remediation]},
|
|
)
|
|
owner = StabilizationLoop(
|
|
gate=MechanicalLoopGate(owner_context, lambda: True, lambda: True),
|
|
context=owner_context,
|
|
language_strategy_set=_language_strategy_set(
|
|
file_manager,
|
|
runner=PassingRunner(),
|
|
coverage=FakeCoverageStrategy(1.0),
|
|
),
|
|
)
|
|
|
|
owner._promote_deferred_remediation(owner._current_cot_forward_remediation_locations())
|
|
|
|
assert owner.context.backlog is not None
|
|
assert owner.context.backlog.remediation == (remediation,)
|
|
assert owner.context.inherited_remediation_backlog == {}
|
|
|
|
|
|
def test_non_final_completion_leaves_deferred_remediation_for_supreme_handoff(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A non-final COT accepts locally but retains remediation for its parent transition."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
root = sources["src/a.py"]
|
|
future = sources["src/b.py"]
|
|
remediation = _failed_remediation(future)
|
|
_freeze_edges(file_manager, sources, [(root.location, future.location)])
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
cot_index=0,
|
|
total_cot=2,
|
|
deferred_remediation_backlog={future.location: [remediation]},
|
|
)
|
|
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),
|
|
),
|
|
future_cot_records=[
|
|
CotTaskRecord("future", future.location, future),
|
|
],
|
|
)
|
|
assert context.backlog is not None
|
|
backlog = context.backlog
|
|
loop._create_alpha_loop = lambda bfs_depth: _DrainSourceWork(backlog)
|
|
|
|
loop.run()
|
|
assert loop.context.accepted is True
|
|
handoff = loop.context.remaining_deferred_remediation
|
|
assert handoff.inherited_items == []
|
|
assert handoff.deferred_items == [remediation]
|
|
assert loop.context.deferred_remediation_backlog == {future.location: [remediation]}
|
|
|
|
|
|
def test_non_final_exhaustion_runs_remediation_outside_the_forward_trajectory(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A direct source repair outside the forward path remains runnable by Alpha."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "outside.py"])
|
|
root = sources["src/a.py"]
|
|
remediation = _failed_remediation(sources["src/outside.py"])
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
cot_index=0,
|
|
total_cot=2,
|
|
deferred_remediation_backlog={"src/outside.py": [remediation]},
|
|
)
|
|
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),
|
|
),
|
|
)
|
|
handled: list[object] = []
|
|
assert context.backlog is not None
|
|
backlog = context.backlog
|
|
loop._create_alpha_loop = lambda bfs_depth: _DrainSourceWork(
|
|
backlog,
|
|
handled,
|
|
)
|
|
|
|
loop.run()
|
|
|
|
assert loop.context.accepted is True
|
|
handoff = loop.context.remaining_deferred_remediation
|
|
assert handoff.inherited_items == []
|
|
assert handoff.deferred_items == []
|
|
assert remediation in handled
|
|
assert loop.context.deferred_remediation_backlog == {}
|
|
|
|
|
|
def test_deferral_keeps_independently_attributed_source_remediation_items(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Deferral cannot merge two blamed sources merely because their test evidence matches."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py", "c.py"])
|
|
first = _failed_remediation(sources["src/b.py"])
|
|
second = _failed_remediation(sources["src/c.py"])
|
|
# FileBlame can give each owner the same downstream failure region.
|
|
first.downstream_classes = [sources["src/b.py"], sources["src/c.py"]]
|
|
second.downstream_classes = [sources["src/b.py"], sources["src/c.py"]]
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=sources["src/a.py"],
|
|
)
|
|
_freeze_edges(
|
|
file_manager,
|
|
sources,
|
|
[
|
|
(sources["src/a.py"].location, sources["src/b.py"].location),
|
|
(sources["src/a.py"].location, sources["src/c.py"].location),
|
|
],
|
|
)
|
|
assert context.backlog is not None
|
|
context.backlog.queue_remediation(first)
|
|
context.backlog.queue_remediation(second)
|
|
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),
|
|
),
|
|
)
|
|
|
|
loop._defer_forward_remediation(loop._current_cot_forward_remediation_locations())
|
|
|
|
assert context.deferred_remediation_backlog == {
|
|
"src/b.py": [first],
|
|
"src/c.py": [second],
|
|
}
|
|
|
|
|
|
def test_non_final_completion_keeps_deferred_remediation_durable_for_parent_retry(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A locally accepted COT leaves its live remediation in the durable child context."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "b.py"])
|
|
root = sources["src/a.py"]
|
|
future = sources["src/b.py"]
|
|
remediation = _failed_remediation(future)
|
|
_freeze_edges(file_manager, sources, [(root.location, future.location)])
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=root,
|
|
cot_index=0,
|
|
total_cot=2,
|
|
deferred_remediation_backlog={future.location: [remediation]},
|
|
)
|
|
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),
|
|
),
|
|
future_cot_records=[
|
|
CotTaskRecord("future", future.location, future),
|
|
],
|
|
)
|
|
assert context.backlog is not None
|
|
backlog = context.backlog
|
|
loop._create_alpha_loop = lambda bfs_depth: _DrainSourceWork(backlog)
|
|
|
|
loop.run()
|
|
assert loop.context.accepted is True
|
|
assert loop.context.deferred_remediation_backlog == {
|
|
future.location: [remediation]
|
|
}
|
|
handoff = loop.context.remaining_deferred_remediation
|
|
assert handoff.inherited_items == []
|
|
assert handoff.deferred_items == [remediation]
|
|
|
|
|
|
def test_late_current_surrogate_is_exempt_from_the_future_closure(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A legitimate current surrogate root cannot remain blacklisted by a cyclic closure."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["a.py", "surrogate.py"])
|
|
_freeze_edges(file_manager, sources, [])
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
classes=[*file_manager.src_classes, *file_manager.test_classes],
|
|
upstream_file_manager=file_manager,
|
|
root_change=sources["src/a.py"],
|
|
)
|
|
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),
|
|
),
|
|
future_cot_records=[
|
|
CotTaskRecord(
|
|
"future",
|
|
"src/surrogate.py",
|
|
sources["src/surrogate.py"],
|
|
),
|
|
],
|
|
)
|
|
|
|
context.extend_roots(sources["src/surrogate.py"])
|
|
|
|
assert [source.location for source in context.t0] == [
|
|
"src/a.py",
|
|
"src/surrogate.py",
|
|
]
|
|
assert loop.context.topology.future_cot_owned_locations == set()
|
|
|
|
|
|
def test_next_cot_exposes_its_predecessor_root_only_to_alpha_source_work(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The immediate predecessor is Alpha evidence, not forced Beta request context."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
sources = _add_sources(file_manager, ["prior_one.py", "prior_two.py", "unrelated.py"])
|
|
root = file_manager.current_src("src/foo.py")
|
|
assert root is not None
|
|
previous_root = file_manager.source_path(sources["src/prior_two.py"].location)
|
|
context = StabilizationContext(
|
|
goal="current COT",
|
|
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),
|
|
),
|
|
previous_cot_root_readonly_file=previous_root,
|
|
)
|
|
|
|
alpha = loop._create_alpha_loop(0)
|
|
beta = loop._create_beta_loop()
|
|
request = alpha._source_edit_request(
|
|
editable_sources=[root],
|
|
readonly_sources=[],
|
|
readonly_tests=[],
|
|
work_needed_by_location={root.location: "current work"},
|
|
failing_test=None,
|
|
is_remediation=False,
|
|
extra_guidance=None,
|
|
)
|
|
|
|
assert alpha.context.pinned_context == []
|
|
assert alpha.previous_cot_root_readonly_file == previous_root
|
|
assert not hasattr(alpha.context, "previous_cot_root_readonly_file")
|
|
assert request.readonly_files == [previous_root]
|
|
assert request.editable_files == [file_manager.source_path(root.location)]
|
|
assert file_manager.source_path(sources["src/prior_one.py"].location) not in request.readonly_files
|
|
assert file_manager.source_path(sources["src/unrelated.py"].location) not in request.readonly_files
|
|
beta_item = _TestingBacklog(
|
|
focused_spotlight=file_manager.current_test("tests/test_foo.py"),
|
|
summary="Current Beta work.",
|
|
downstream_classes=[root],
|
|
)
|
|
assert beta._readonly_paths(beta_item) == [
|
|
file_manager.source_path(root.location),
|
|
]
|
|
assert not hasattr(beta.context, "previous_cot_root_readonly_file")
|