108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Architecture contracts for the stabilization scheduling surface."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import BiDependencyDiscoveryStrategy
|
|
from band_aider.judgement.composite_dependency_discovery_strategy import (
|
|
CompositeDependencyDiscoveryStrategy,
|
|
)
|
|
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
|
|
from band_aider.judgement.python_ast_dependency_strategy import PythonAstDependencyStrategy
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("owner", "keyframes"),
|
|
(
|
|
(
|
|
StabilizationLoop,
|
|
frozenset(
|
|
{
|
|
"_topology_cadence",
|
|
"_frontier_boundary",
|
|
"_log_deferred_topology",
|
|
"_enqueue_current_frontier_to_backlog",
|
|
"_schedule_final_exam",
|
|
}
|
|
),
|
|
),
|
|
(
|
|
TopologicalLiveliness,
|
|
frozenset(
|
|
{
|
|
"_track_live_beam",
|
|
"_index_source",
|
|
"_record_perimeter",
|
|
"_closed_cycle_tiebreak",
|
|
"_topological_readiness",
|
|
"_perimeter_proof",
|
|
"_log_reconciliation",
|
|
}
|
|
),
|
|
),
|
|
),
|
|
)
|
|
def test_stabilization_keyframes_are_owned_by_their_declared_workflow_type(
|
|
owner: type[object],
|
|
keyframes: frozenset[str],
|
|
) -> None:
|
|
"""Each workflow exposes its named scheduling keyframes at its owner boundary."""
|
|
|
|
declared = set(owner.__dict__)
|
|
|
|
assert keyframes <= declared
|
|
assert all(callable(getattr(owner, name)) for name in keyframes)
|
|
|
|
|
|
def test_stabilization_loop_does_not_expose_retired_topology_pass_throughs() -> None:
|
|
"""The loop owns cadence directly rather than publishing obsolete wrapper entry points."""
|
|
|
|
assert not {
|
|
"_next_frontier",
|
|
"_draw_bfs_slice",
|
|
"_dequeue_topologically",
|
|
"_enqueue_bfs_boundary",
|
|
} & set(StabilizationLoop.__dict__)
|
|
|
|
|
|
def test_topology_and_context_do_not_expose_bfs_scheduling_storage() -> None:
|
|
"""Breadth scheduling remains a loop concern, not topology or context state."""
|
|
|
|
assert not {
|
|
"enqueue_bfs_boundary",
|
|
"dequeue_bfs_boundary",
|
|
"has_bfs_slice",
|
|
"has_remaining_downstream",
|
|
} & set(TopologicalLiveliness.__dict__)
|
|
assert not {
|
|
"stage_bfs_boundary",
|
|
"draw_bfs_boundary",
|
|
"has_remaining_bfs",
|
|
} & set(StabilizationContext.__dict__)
|
|
|
|
|
|
def test_direct_adjacency_is_declared_by_each_graph_view_but_not_the_marker_protocol() -> None:
|
|
"""Concrete graph views and their composite own adjacency; the marker stays empty."""
|
|
|
|
adjacency = {"source_dependencies_of", "source_dependents_of"}
|
|
|
|
assert adjacency <= set(CompositeDependencyDiscoveryStrategy.__dict__)
|
|
assert adjacency <= set(PythonAstDependencyStrategy.__dict__)
|
|
assert adjacency <= set(DependencyDiscoveryStrategy.__dict__)
|
|
assert not adjacency & set(BiDependencyDiscoveryStrategy.__dict__)
|
|
assert "inners" in CompositeDependencyDiscoveryStrategy.__annotations__
|
|
|
|
|
|
def test_topology_cadence_remains_one_private_stabilization_keyframe() -> None:
|
|
"""The public loop owns one cadence keyframe without a persistence entry point."""
|
|
|
|
assert callable(StabilizationLoop.__dict__["_topology_cadence"])
|
|
assert "persist_topology_cadence" not in StabilizationLoop.__dict__
|