282 lines
12 KiB
Python
282 lines
12 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.judgement.code_dependency_graph import (
|
|
BindingSite,
|
|
CodeDependencyGraph,
|
|
DeclarationSite,
|
|
DependencySelector,
|
|
DependencyType,
|
|
IndirectReference,
|
|
ReturnSite,
|
|
)
|
|
from band_aider.judgement.code_dependency_graph_maintainer import (
|
|
CodeDependencyGraphMaintainer,
|
|
GraphSignature,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class FrozenCodeDependencyGraphMaintainer(CodeDependencyGraphMaintainer):
|
|
"""FrozenCodeDependencyGraphMaintainer exposes an existing graph without mutating it."""
|
|
|
|
_graph: CodeDependencyGraph = field(default_factory=CodeDependencyGraph)
|
|
signature: GraphSignature | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Freeze the stored graph immediately."""
|
|
|
|
self._graph.freeze()
|
|
|
|
@property
|
|
def graph(self) -> CodeDependencyGraph:
|
|
"""Return the stored graph snapshot."""
|
|
|
|
return self._graph
|
|
|
|
@property
|
|
def is_frozen(self) -> bool:
|
|
"""Return whether mutation is disabled."""
|
|
|
|
return True
|
|
|
|
def freeze(self) -> None:
|
|
"""Keep the stored graph frozen."""
|
|
|
|
self._graph.freeze()
|
|
|
|
def build(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
"""Return the stored graph snapshot without rebuilding it."""
|
|
|
|
raise RuntimeError(
|
|
"FrozenCodeDependencyGraphMaintainer is frozen and cannot be mutated."
|
|
)
|
|
|
|
def upsert(
|
|
self,
|
|
class_meta: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
"""Reject ClassMeta upserts for the frozen graph."""
|
|
|
|
raise RuntimeError(
|
|
"FrozenCodeDependencyGraphMaintainer is frozen and cannot be mutated."
|
|
)
|
|
|
|
def delete(
|
|
self,
|
|
class_meta: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
"""Reject ClassMeta deletion for the frozen graph."""
|
|
|
|
raise RuntimeError(
|
|
"FrozenCodeDependencyGraphMaintainer is frozen and cannot be mutated."
|
|
)
|
|
|
|
def direct_tests_for(self, source_files: list[ClassMeta]) -> list[ClassMeta]:
|
|
locations = {
|
|
test_location
|
|
for source in source_files
|
|
for test_location in self._graph.tests_for_source.get(source.location, set())
|
|
}
|
|
return [self._graph.test_by_location[location] for location in sorted(locations) if location in self._graph.test_by_location]
|
|
|
|
def trace_types_defined_in(self, source_location: str) -> list[DependencyType]:
|
|
return [
|
|
self._graph.type_definitions[type_id]
|
|
for type_id in sorted(self._graph.type_ids_by_source.get(source_location, set()))
|
|
]
|
|
|
|
def direct_parent_type_ids_of(self, child_type_id: str) -> set[str]:
|
|
return {
|
|
self._graph.parent_relationships_by_id[relationship_id].parent_type_id
|
|
for relationship_id in self._graph.parent_relationship_ids_by_child.get(child_type_id, set())
|
|
}
|
|
|
|
def direct_child_type_ids_of(self, parent_type_id: str) -> set[str]:
|
|
return {
|
|
self._graph.parent_relationships_by_id[relationship_id].child_type_id
|
|
for relationship_id in self._graph.parent_relationship_ids_by_parent.get(parent_type_id, set())
|
|
}
|
|
|
|
def trace_declaration_sites_in(self, source_location: str) -> list[DeclarationSite]:
|
|
return [
|
|
self._graph.declaration_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.declaration_site_ids_by_source.get(source_location, set()))
|
|
]
|
|
|
|
def trace_declaration_sites_declaring(self, type_id: str) -> list[DeclarationSite]:
|
|
return [
|
|
self._graph.declaration_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.declaration_site_ids_by_declared_type.get(type_id, set()))
|
|
]
|
|
|
|
def trace_declaration_sites_for_callable(self, callable_id: str) -> list[DeclarationSite]:
|
|
return [
|
|
self._graph.declaration_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.declaration_site_ids_by_callable.get(callable_id, set()))
|
|
]
|
|
|
|
def trace_binding_sites_in(self, source_location: str) -> list[BindingSite]:
|
|
return [
|
|
self._graph.binding_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.binding_site_ids_by_source.get(source_location, set()))
|
|
]
|
|
|
|
def trace_binding_sites_for_declaration(self, declaration_site_id: str) -> list[BindingSite]:
|
|
return [
|
|
self._graph.binding_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.binding_site_ids_by_declaration_site.get(declaration_site_id, set()))
|
|
]
|
|
|
|
def trace_binding_sites_supplying(self, type_id: str) -> list[BindingSite]:
|
|
return [
|
|
self._graph.binding_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.binding_site_ids_by_actual_type.get(type_id, set()))
|
|
]
|
|
|
|
def binding_sites_returned_by(self, callable_id: str) -> list[BindingSite]:
|
|
return [
|
|
self._graph.binding_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.binding_site_ids_by_returned_callable.get(callable_id, set()))
|
|
]
|
|
|
|
def trace_return_sites_in(self, source_location: str) -> list[ReturnSite]:
|
|
return [
|
|
self._graph.return_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.return_site_ids_by_source.get(source_location, set()))
|
|
]
|
|
|
|
def trace_return_sites_for_callable(self, callable_id: str) -> list[ReturnSite]:
|
|
return [
|
|
self._graph.return_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.return_site_ids_by_callable.get(callable_id, set()))
|
|
]
|
|
|
|
def trace_return_sites_declaring(self, type_id: str) -> list[ReturnSite]:
|
|
return [
|
|
self._graph.return_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.return_site_ids_by_declared_type.get(type_id, set()))
|
|
]
|
|
|
|
def trace_return_sites_returning(self, type_id: str) -> list[ReturnSite]:
|
|
return [
|
|
self._graph.return_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.return_site_ids_by_actual_type.get(type_id, set()))
|
|
]
|
|
|
|
def trace_return_sites_calling(self, callable_id: str) -> list[ReturnSite]:
|
|
return [
|
|
self._graph.return_sites_by_id[site_id]
|
|
for site_id in sorted(self._graph.return_site_ids_by_returned_callable.get(callable_id, set()))
|
|
]
|
|
|
|
def trace_indirect_references_for_declaration(self, declaration_site_id: str) -> list[IndirectReference]:
|
|
return [
|
|
self._graph.indirect_references_by_id[reference_id]
|
|
for reference_id in sorted(self._graph.indirect_reference_ids_by_declaration_site.get(declaration_site_id, set()))
|
|
]
|
|
|
|
def trace_indirect_references_through_binding(self, binding_site_id: str) -> list[IndirectReference]:
|
|
return [
|
|
self._graph.indirect_references_by_id[reference_id]
|
|
for reference_id in sorted(self._graph.indirect_reference_ids_by_binding_site.get(binding_site_id, set()))
|
|
]
|
|
|
|
def trace_indirect_references_through_return(self, return_site_id: str) -> list[IndirectReference]:
|
|
return [
|
|
self._graph.indirect_references_by_id[reference_id]
|
|
for reference_id in sorted(self._graph.indirect_reference_ids_by_return_site.get(return_site_id, set()))
|
|
]
|
|
|
|
def trace_indirect_references_using_actual_type(self, type_id: str) -> list[IndirectReference]:
|
|
return [
|
|
self._graph.indirect_references_by_id[reference_id]
|
|
for reference_id in sorted(self._graph.indirect_reference_ids_by_actual_type.get(type_id, set()))
|
|
]
|
|
|
|
def trace_indirect_references_using_parent_relationship(self, relationship_id: str) -> list[IndirectReference]:
|
|
return [
|
|
self._graph.indirect_references_by_id[reference_id]
|
|
for reference_id in sorted(self._graph.indirect_reference_ids_by_parent_relationship.get(relationship_id, set()))
|
|
]
|
|
|
|
def trace_dynamic_dependencies_of(self, source_location: str) -> set[str]:
|
|
return {
|
|
self._graph.type_definitions[reference.actual_type_id].source.location
|
|
for reference_id in self._graph.indirect_reference_ids_by_dependent_source.get(source_location, set())
|
|
if (reference := self._graph.indirect_references_by_id.get(reference_id)) is not None
|
|
and reference.actual_type_id in self._graph.type_definitions
|
|
and self._graph.type_definitions[reference.actual_type_id].source.location != source_location
|
|
}
|
|
|
|
def trace_dynamic_dependents_of(self, source_location: str) -> set[str]:
|
|
return {
|
|
self._graph.declaration_sites_by_id[reference.declaration_site_id].source.location
|
|
for reference_id in self._graph.indirect_reference_ids_by_dependency_source.get(source_location, set())
|
|
if (reference := self._graph.indirect_references_by_id.get(reference_id)) is not None
|
|
and reference.declaration_site_id in self._graph.declaration_sites_by_id
|
|
and self._graph.declaration_sites_by_id[reference.declaration_site_id].source.location != source_location
|
|
}
|
|
|
|
def source_dependencies_of(self, source_location: str, selector: DependencySelector = DependencySelector.BOTH) -> set[str]:
|
|
static = set(self._graph.source_dependencies.get(source_location, set()))
|
|
dynamic = self.trace_dynamic_dependencies_of(source_location)
|
|
if selector is DependencySelector.STATIC:
|
|
return static
|
|
if selector is DependencySelector.DYNAMIC:
|
|
return dynamic - static
|
|
return static | dynamic
|
|
|
|
def source_dependents_of(self, source_location: str, selector: DependencySelector = DependencySelector.BOTH) -> set[str]:
|
|
static = set(self._graph.source_dependents.get(source_location, set()))
|
|
dynamic = self.trace_dynamic_dependents_of(source_location)
|
|
if selector is DependencySelector.STATIC:
|
|
return static
|
|
if selector is DependencySelector.DYNAMIC:
|
|
return dynamic - static
|
|
return static | dynamic
|
|
|
|
def dependency_kind_between(self, dependent: str, dependency: str) -> DependencySelector | None:
|
|
static = dependency in self._graph.source_dependencies.get(dependent, set())
|
|
if static:
|
|
return DependencySelector.STATIC
|
|
if dependency in self.trace_dynamic_dependencies_of(dependent):
|
|
return DependencySelector.DYNAMIC
|
|
return None
|
|
|
|
def upstream_sources_for(self, source_files: list[ClassMeta], selector: DependencySelector = DependencySelector.BOTH) -> list[ClassMeta]:
|
|
locations: set[str] = set()
|
|
for source in source_files:
|
|
if source.location in self._graph.source_by_location:
|
|
locations.update(self.source_dependencies_of(source.location, selector))
|
|
elif source.location in self._graph.test_by_location:
|
|
locations.update(self._graph.sources_for_test.get(source.location, set()))
|
|
return [self._graph.source_by_location[location] for location in sorted(locations) if location in self._graph.source_by_location]
|
|
|
|
def downstream_sources_for(self, source_files: list[ClassMeta], selector: DependencySelector = DependencySelector.BOTH) -> list[ClassMeta]:
|
|
locations = {dependent for source in source_files for dependent in self.source_dependents_of(source.location, selector)}
|
|
return [self._graph.source_by_location[location] for location in sorted(locations) if location in self._graph.source_by_location]
|