91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Public-contract drivers for stabilization topology tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.loop.stabilization_dtos import FrontierBoundary
|
|
|
|
|
|
def boundary_for(
|
|
loop,
|
|
frontier: list[ClassMeta],
|
|
opening_hashes: dict[str, str],
|
|
) -> FrontierBoundary:
|
|
"""Build the ordinary unchanged/live seed boundary independently for a test."""
|
|
|
|
file_manager = loop.context.file_manager
|
|
current_hashes = dict(map(
|
|
lambda source: (
|
|
source.location,
|
|
(file_manager.current_src(source.location) or source).get_hash(
|
|
file_manager
|
|
),
|
|
),
|
|
frontier,
|
|
))
|
|
unchanged = set(map(
|
|
lambda item: item[0],
|
|
filter(
|
|
lambda item: current_hashes.get(item[0]) == item[1],
|
|
opening_hashes.items(),
|
|
),
|
|
))
|
|
current = set(map(lambda source: source.location, frontier))
|
|
topology = loop.context.topology
|
|
whitelist_seeds = set(filter(
|
|
lambda location: location in topology.visited or location in current,
|
|
topology.whitelist,
|
|
))
|
|
live_seeds = (current - unchanged) | whitelist_seeds
|
|
source_by_location = dict(map(
|
|
lambda source: (
|
|
source.location,
|
|
file_manager.current_src(source.location) or source,
|
|
),
|
|
frontier,
|
|
))
|
|
source_by_location.update(dict(filter(
|
|
lambda item: item[1] is not None,
|
|
map(
|
|
lambda location: (location, file_manager.current_src(location)),
|
|
live_seeds - source_by_location.keys(),
|
|
),
|
|
)))
|
|
return FrontierBoundary(
|
|
current_locations=frozenset(current),
|
|
unchanged_locations=frozenset(unchanged),
|
|
live_seed_locations=frozenset(live_seeds),
|
|
source_by_location=source_by_location,
|
|
)
|
|
|
|
|
|
def service_topology(
|
|
loop,
|
|
frontier: list[ClassMeta],
|
|
opening_hashes: dict[str, str] | None = None,
|
|
) -> list[ClassMeta]:
|
|
"""Run one public draw/push/dequeue cadence for an isolated topology assertion."""
|
|
|
|
opening_hashes = opening_hashes or {}
|
|
topology = loop.context.topology
|
|
boundary = boundary_for(loop, frontier, opening_hashes)
|
|
file_manager = loop.context.file_manager
|
|
strategy = loop.bi_dependency_strategy
|
|
seeds = list(map(
|
|
lambda location: boundary.source_by_location[location],
|
|
filter(
|
|
lambda location: location in boundary.source_by_location,
|
|
sorted(boundary.live_seed_locations),
|
|
),
|
|
))
|
|
bfs_slice = strategy.downstream_sources_for(
|
|
seeds,
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
)
|
|
topology.enqueue_bfs_slice(bfs_slice, boundary)
|
|
return topology.dequeue_topologically(input_complete=True)
|