751 lines
42 KiB
Python
751 lines
42 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 enum import Enum
|
|
from typing import Mapping
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DependencyType:
|
|
"""One normalized project type."""
|
|
|
|
type_id: str
|
|
qualified_name: str
|
|
local_name: str
|
|
source: ClassMeta
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ParentRelationship:
|
|
"""One direct child-to-parent relationship."""
|
|
|
|
parent_type_id: str
|
|
child_type_id: str
|
|
source_location: str
|
|
line: int
|
|
column: int
|
|
relationship_id: str = field(init=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Derive the stable relationship identity from both endpoint types."""
|
|
object.__setattr__(self, "relationship_id", f"{self.child_type_id}->{self.parent_type_id}")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeclarationSite:
|
|
"""One substitution-eligible receiving slot."""
|
|
|
|
declaration_site_id: str
|
|
declared_type_id: str
|
|
source: ClassMeta
|
|
callable_id: str | None
|
|
field_owner_id: str | None
|
|
receiving_name: str
|
|
declaration_kind: str
|
|
line: int
|
|
column: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BindingSite:
|
|
"""Syntactic evidence supplying a value to one declaration site."""
|
|
|
|
binding_site_id: str
|
|
declaration_site_id: str
|
|
source: ClassMeta
|
|
binding_kind: str
|
|
direct_actual_type_id: str | None
|
|
returned_callable_id: str | None
|
|
line: int
|
|
column: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReturnSite:
|
|
"""One return-value provenance site."""
|
|
|
|
return_site_id: str
|
|
callable_id: str
|
|
declared_return_type_id: str
|
|
source: ClassMeta
|
|
direct_actual_type_id: str | None
|
|
returned_callable_id: str | None
|
|
line: int
|
|
column: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IndirectReference:
|
|
"""One proven declaration/binding/actual relationship."""
|
|
|
|
indirect_reference_id: str
|
|
declaration_site_id: str
|
|
binding_site_id: str
|
|
return_site_ids: tuple[str, ...]
|
|
declared_type_id: str
|
|
actual_type_id: str
|
|
compatibility_kind: str
|
|
parent_relationship_ids: tuple[str, ...]
|
|
structural_member_names: tuple[str, ...]
|
|
|
|
|
|
class DependencySelector(str, Enum):
|
|
STATIC = "static"
|
|
DYNAMIC = "dynamic"
|
|
BOTH = "both"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DependencyDelta:
|
|
"""Changed-only records for one atomic graph mutation."""
|
|
|
|
source_added: Mapping[str, ClassMeta] = field(default_factory=dict)
|
|
source_removed: frozenset[str] = frozenset()
|
|
source_changed: Mapping[str, ClassMeta] = field(default_factory=dict)
|
|
test_added: Mapping[str, ClassMeta] = field(default_factory=dict)
|
|
test_removed: frozenset[str] = frozenset()
|
|
test_changed: Mapping[str, ClassMeta] = field(default_factory=dict)
|
|
static_pairs_added: frozenset[tuple[str, str]] = frozenset()
|
|
static_pairs_removed: frozenset[tuple[str, str]] = frozenset()
|
|
types_added: Mapping[str, DependencyType] = field(default_factory=dict)
|
|
types_removed: frozenset[str] = frozenset()
|
|
types_changed: Mapping[str, DependencyType] = field(default_factory=dict)
|
|
relationships_added: Mapping[str, ParentRelationship] = field(default_factory=dict)
|
|
relationships_removed: frozenset[str] = frozenset()
|
|
relationships_changed: Mapping[str, ParentRelationship] = field(default_factory=dict)
|
|
declarations_added: Mapping[str, DeclarationSite] = field(default_factory=dict)
|
|
declarations_removed: frozenset[str] = frozenset()
|
|
declarations_changed: Mapping[str, DeclarationSite] = field(default_factory=dict)
|
|
bindings_added: Mapping[str, BindingSite] = field(default_factory=dict)
|
|
bindings_removed: frozenset[str] = frozenset()
|
|
bindings_changed: Mapping[str, BindingSite] = field(default_factory=dict)
|
|
returns_added: Mapping[str, ReturnSite] = field(default_factory=dict)
|
|
returns_removed: frozenset[str] = frozenset()
|
|
returns_changed: Mapping[str, ReturnSite] = field(default_factory=dict)
|
|
references_added: Mapping[str, IndirectReference] = field(default_factory=dict)
|
|
references_removed: frozenset[str] = frozenset()
|
|
references_changed: Mapping[str, IndirectReference] = field(default_factory=dict)
|
|
tests_for_source_added: frozenset[tuple[str, str]] = frozenset()
|
|
tests_for_source_removed: frozenset[tuple[str, str]] = frozenset()
|
|
|
|
|
|
@dataclass
|
|
class CodeDependencyGraph:
|
|
"""Own canonical dependency evidence and its query indexes.
|
|
|
|
The graph encapsulates source/test records, dependency relationships, and
|
|
reverse indexes used by discovery and context selection. Callers provide
|
|
normalized records or changed-only deltas, query the graph, and freeze it
|
|
before handing it to consumers that must observe a stable snapshot.
|
|
|
|
Example:
|
|
graph = CodeDependencyGraph()
|
|
graph.has_source("src/example.py")
|
|
graph.freeze()
|
|
"""
|
|
|
|
# Canonical source records keyed by their stable source location.
|
|
source_by_location: dict[str, ClassMeta] = field(default_factory=dict)
|
|
# Canonical test records keyed by their stable test location.
|
|
test_by_location: dict[str, ClassMeta] = field(default_factory=dict)
|
|
# Directed source-to-dependency edges: dependent location -> dependency location.
|
|
source_dependencies: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reverse dependency edges used to find affected sources: dependency -> dependents.
|
|
source_dependents: dict[str, set[str]] = field(default_factory=dict)
|
|
|
|
# Type records keyed by stable type identity.
|
|
type_definitions: dict[str, DependencyType] = field(default_factory=dict)
|
|
# Type identities grouped by the source location that defines them.
|
|
type_ids_by_source: dict[str, set[str]] = field(default_factory=dict)
|
|
# Parent relationships keyed by their stable relationship identity.
|
|
parent_relationships_by_id: dict[str, ParentRelationship] = field(default_factory=dict)
|
|
# Relationship identities grouped by child type.
|
|
parent_relationship_ids_by_child: dict[str, set[str]] = field(default_factory=dict)
|
|
# Relationship identities grouped by parent type.
|
|
parent_relationship_ids_by_parent: dict[str, set[str]] = field(default_factory=dict)
|
|
|
|
# Declaration records keyed by stable declaration-site identity.
|
|
declaration_sites_by_id: dict[str, DeclarationSite] = field(default_factory=dict)
|
|
# Declaration identities grouped by source location.
|
|
declaration_site_ids_by_source: dict[str, set[str]] = field(default_factory=dict)
|
|
# Declaration identities grouped by declared type.
|
|
declaration_site_ids_by_declared_type: dict[str, set[str]] = field(default_factory=dict)
|
|
# Declaration identities grouped by containing callable when present.
|
|
declaration_site_ids_by_callable: dict[str, set[str]] = field(default_factory=dict)
|
|
|
|
# Binding records keyed by stable binding-site identity.
|
|
binding_sites_by_id: dict[str, BindingSite] = field(default_factory=dict)
|
|
# Binding identities grouped by source location.
|
|
binding_site_ids_by_source: dict[str, set[str]] = field(default_factory=dict)
|
|
# Binding identities grouped by the declaration they supply.
|
|
binding_site_ids_by_declaration_site: dict[str, set[str]] = field(default_factory=dict)
|
|
# Binding identities grouped by directly observed actual type.
|
|
binding_site_ids_by_actual_type: dict[str, set[str]] = field(default_factory=dict)
|
|
# Binding identities grouped by returned callable provenance.
|
|
binding_site_ids_by_returned_callable: dict[str, set[str]] = field(default_factory=dict)
|
|
|
|
# Return-value provenance records keyed by stable return-site identity.
|
|
return_sites_by_id: dict[str, ReturnSite] = field(default_factory=dict)
|
|
# Return identities grouped by source location.
|
|
return_site_ids_by_source: dict[str, set[str]] = field(default_factory=dict)
|
|
# Return identities grouped by containing callable.
|
|
return_site_ids_by_callable: dict[str, set[str]] = field(default_factory=dict)
|
|
# Return identities grouped by declared return type.
|
|
return_site_ids_by_declared_type: dict[str, set[str]] = field(default_factory=dict)
|
|
# Return identities grouped by directly observed actual type.
|
|
return_site_ids_by_actual_type: dict[str, set[str]] = field(default_factory=dict)
|
|
# Return identities grouped by returned callable provenance.
|
|
return_site_ids_by_returned_callable: dict[str, set[str]] = field(default_factory=dict)
|
|
|
|
# Indirect-reference records keyed by stable reference identity.
|
|
indirect_references_by_id: dict[str, IndirectReference] = field(default_factory=dict)
|
|
# Reference identities grouped by declaration site.
|
|
indirect_reference_ids_by_declaration_site: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reference identities grouped by binding site.
|
|
indirect_reference_ids_by_binding_site: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reference identities grouped by return site.
|
|
indirect_reference_ids_by_return_site: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reference identities grouped by actual type.
|
|
indirect_reference_ids_by_actual_type: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reference identities grouped by parent relationship.
|
|
indirect_reference_ids_by_parent_relationship: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reference identities grouped by dependent source location.
|
|
indirect_reference_ids_by_dependent_source: dict[str, set[str]] = field(default_factory=dict)
|
|
# Reference identities grouped by dependency source location.
|
|
indirect_reference_ids_by_dependency_source: dict[str, set[str]] = field(default_factory=dict)
|
|
|
|
# Test locations grouped by the source locations they cover.
|
|
tests_for_source: dict[str, set[str]] = field(default_factory=dict)
|
|
# Source locations grouped by the tests that cover them.
|
|
sources_for_test: dict[str, set[str]] = field(default_factory=dict)
|
|
# Once frozen, all graph attributes reject mutation except the freeze marker itself.
|
|
_frozen: bool = field(default=False, init=False, repr=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Copy caller-owned indexes and rebuild canonical evidence indexes."""
|
|
self.source_by_location = dict(self.source_by_location)
|
|
self.test_by_location = dict(self.test_by_location)
|
|
self.source_dependencies = self._coerce_adjacency(self.source_dependencies)
|
|
self.source_dependents = self._coerce_adjacency(self.source_dependents)
|
|
# Reconstruct reverse edges when callers provide only forward dependencies.
|
|
if not self.source_dependents and self.source_dependencies:
|
|
self.source_dependents = self._reverse_adjacency(self.source_dependencies)
|
|
self.tests_for_source = self._coerce_adjacency(self.tests_for_source)
|
|
self.sources_for_test = self._coerce_adjacency(self.sources_for_test)
|
|
self._replace_all_dependency_evidence(
|
|
type_definitions=self.type_definitions,
|
|
parent_relationships_by_id=self.parent_relationships_by_id,
|
|
declaration_sites_by_id=self.declaration_sites_by_id,
|
|
binding_sites_by_id=self.binding_sites_by_id,
|
|
return_sites_by_id=self.return_sites_by_id,
|
|
indirect_references_by_id=self.indirect_references_by_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def _coerce_adjacency(adjacency: Mapping[str, set[str]]) -> dict[str, set[str]]:
|
|
"""Copy non-empty adjacency sets into the graph's canonical shape."""
|
|
return {key: set(values) for key, values in adjacency.items() if values}
|
|
|
|
@staticmethod
|
|
def _reverse_adjacency(adjacency: Mapping[str, set[str]]) -> dict[str, set[str]]:
|
|
"""Build reverse edges while excluding self-dependencies."""
|
|
reverse: dict[str, set[str]] = {}
|
|
# Visit each forward edge so the reverse index answers dependent lookups.
|
|
for source, dependencies in adjacency.items():
|
|
# Visit each dependency in the source's adjacency set.
|
|
for dependency in dependencies:
|
|
# Exclude self-edges that do not represent a dependency relationship.
|
|
if source != dependency:
|
|
reverse.setdefault(dependency, set()).add(source)
|
|
return reverse
|
|
|
|
def __setattr__(self, name: str, value) -> None:
|
|
"""Reject attribute mutation after the graph has been frozen."""
|
|
# Protect the frozen snapshot while allowing the lifecycle marker to change.
|
|
if getattr(self, "_frozen", False) and name != "_frozen":
|
|
raise RuntimeError("CodeDependencyGraph is frozen and cannot be mutated.")
|
|
object.__setattr__(self, name, value)
|
|
|
|
@property
|
|
def is_frozen(self) -> bool:
|
|
"""Report whether this graph has entered its immutable snapshot state."""
|
|
return self._frozen
|
|
|
|
def freeze(self) -> None:
|
|
"""Seal the graph so subsequent attribute mutation fails."""
|
|
self._frozen = True
|
|
|
|
def _assert_mutable(self) -> None:
|
|
"""Enforce the graph's mutable-building lifecycle."""
|
|
# Reject updates after consumers may rely on a stable graph snapshot.
|
|
if self._frozen:
|
|
raise RuntimeError("CodeDependencyGraph is frozen and cannot be mutated.")
|
|
|
|
@staticmethod
|
|
def _discard(index: dict[str, set[str]], key: str, value: str) -> None:
|
|
"""Remove one indexed value and prune an empty reverse-index bucket."""
|
|
values = index.get(key)
|
|
# Ignore absent buckets because the requested relationship is already gone.
|
|
if values is None:
|
|
return
|
|
values.discard(value)
|
|
# Remove empty buckets so indexes retain only meaningful relationships.
|
|
if not values:
|
|
index.pop(key, None)
|
|
|
|
def has_source(self, source_location: str) -> bool:
|
|
"""Report whether a source record exists at the requested location."""
|
|
return source_location in self.source_by_location
|
|
|
|
def _add_source_dependency(self, dependent: str, dependency: str) -> None:
|
|
"""Add one directed source dependency and its reverse edge."""
|
|
self._assert_mutable()
|
|
# Ignore self-dependencies because a source cannot depend on itself.
|
|
if dependent == dependency:
|
|
return
|
|
self.source_dependencies.setdefault(dependent, set()).add(dependency)
|
|
self.source_dependents.setdefault(dependency, set()).add(dependent)
|
|
|
|
def _remove_source_dependency(self, dependent: str, dependency: str) -> None:
|
|
"""Remove one directed source dependency and its reverse edge."""
|
|
self._assert_mutable()
|
|
self._discard(self.source_dependencies, dependent, dependency)
|
|
self._discard(self.source_dependents, dependency, dependent)
|
|
|
|
def _add_source_test_mapping(self, source_location: str, test_location: str) -> None:
|
|
"""Associate one source with one test in both directions."""
|
|
self._assert_mutable()
|
|
self.tests_for_source.setdefault(source_location, set()).add(test_location)
|
|
self.sources_for_test.setdefault(test_location, set()).add(source_location)
|
|
|
|
def _remove_source_test_mapping(self, source_location: str, test_location: str) -> None:
|
|
"""Remove one source/test association in both directions."""
|
|
self._assert_mutable()
|
|
self._discard(self.tests_for_source, source_location, test_location)
|
|
self._discard(self.sources_for_test, test_location, source_location)
|
|
|
|
def _remove_test_mappings_for_source(self, source_location: str) -> None:
|
|
"""Remove every test association owned by one source location."""
|
|
# Remove each reverse association before the source record leaves the graph.
|
|
for test_location in tuple(self.tests_for_source.get(source_location, ())):
|
|
self._remove_source_test_mapping(source_location, test_location)
|
|
|
|
def _remove_test_mappings_for_test(self, test_location: str) -> None:
|
|
"""Remove every source association owned by one test location."""
|
|
# Remove each reverse association before the test record leaves the graph.
|
|
for source_location in tuple(self.sources_for_test.get(test_location, ())):
|
|
self._remove_source_test_mapping(source_location, test_location)
|
|
|
|
def _replace_all_dependency_evidence(
|
|
self,
|
|
*,
|
|
type_definitions: Mapping[str, DependencyType],
|
|
parent_relationships_by_id: Mapping[str, ParentRelationship],
|
|
declaration_sites_by_id: Mapping[str, DeclarationSite],
|
|
binding_sites_by_id: Mapping[str, BindingSite],
|
|
return_sites_by_id: Mapping[str, ReturnSite],
|
|
indirect_references_by_id: Mapping[str, IndirectReference],
|
|
) -> None:
|
|
"""Reconstruct canonical evidence and paired indexes for full build, hydration, or union."""
|
|
self.type_definitions = {}
|
|
self.type_ids_by_source = {}
|
|
# Admit only type records whose map key agrees with their stable identity.
|
|
for key, record in type_definitions.items():
|
|
# Reject malformed identity pairs before indexing their provenance.
|
|
if key != record.type_id:
|
|
raise ValueError("dependency type key does not match its identity")
|
|
self.type_definitions[key] = record
|
|
self.type_ids_by_source.setdefault(record.source.location, set()).add(key)
|
|
|
|
self.parent_relationships_by_id = {}
|
|
self.parent_relationship_ids_by_child = {}
|
|
self.parent_relationship_ids_by_parent = {}
|
|
# Retain parent edges only when both endpoint types are known.
|
|
for key, record in parent_relationships_by_id.items():
|
|
# Reject malformed relationship identities before indexing the edge.
|
|
if key != record.relationship_id:
|
|
raise ValueError("parent relationship key does not match its identity")
|
|
# Skip relationships whose endpoints cannot be resolved in this graph.
|
|
if record.child_type_id not in self.type_definitions or record.parent_type_id not in self.type_definitions:
|
|
continue
|
|
self.parent_relationships_by_id[key] = record
|
|
self.parent_relationship_ids_by_child.setdefault(record.child_type_id, set()).add(key)
|
|
self.parent_relationship_ids_by_parent.setdefault(record.parent_type_id, set()).add(key)
|
|
|
|
self.declaration_sites_by_id = {}
|
|
self.declaration_site_ids_by_source = {}
|
|
self.declaration_site_ids_by_declared_type = {}
|
|
self.declaration_site_ids_by_callable = {}
|
|
# Retain declarations whose identity and declared type are both canonical.
|
|
for key, record in declaration_sites_by_id.items():
|
|
# Skip malformed or orphaned declaration records.
|
|
if key != record.declaration_site_id or record.declared_type_id not in self.type_definitions:
|
|
continue
|
|
self.declaration_sites_by_id[key] = record
|
|
self.declaration_site_ids_by_source.setdefault(record.source.location, set()).add(key)
|
|
self.declaration_site_ids_by_declared_type.setdefault(record.declared_type_id, set()).add(key)
|
|
# Index callable ownership only when the declaration is nested in one.
|
|
if record.callable_id is not None:
|
|
self.declaration_site_ids_by_callable.setdefault(record.callable_id, set()).add(key)
|
|
|
|
self.binding_sites_by_id = {}
|
|
self.binding_site_ids_by_source = {}
|
|
self.binding_site_ids_by_declaration_site = {}
|
|
self.binding_site_ids_by_actual_type = {}
|
|
self.binding_site_ids_by_returned_callable = {}
|
|
# Retain bindings whose identity and declaration owner are canonical.
|
|
for key, record in binding_sites_by_id.items():
|
|
# Skip malformed or orphaned binding records.
|
|
if key != record.binding_site_id or record.declaration_site_id not in self.declaration_sites_by_id:
|
|
continue
|
|
self.binding_sites_by_id[key] = record
|
|
self.binding_site_ids_by_source.setdefault(record.source.location, set()).add(key)
|
|
self.binding_site_ids_by_declaration_site.setdefault(record.declaration_site_id, set()).add(key)
|
|
# Index direct actual types only when syntax proves one.
|
|
if record.direct_actual_type_id is not None:
|
|
self.binding_site_ids_by_actual_type.setdefault(record.direct_actual_type_id, set()).add(key)
|
|
# Index returned-callable provenance only when syntax proves one.
|
|
if record.returned_callable_id is not None:
|
|
self.binding_site_ids_by_returned_callable.setdefault(record.returned_callable_id, set()).add(key)
|
|
|
|
self.return_sites_by_id = {}
|
|
self.return_site_ids_by_source = {}
|
|
self.return_site_ids_by_callable = {}
|
|
self.return_site_ids_by_declared_type = {}
|
|
self.return_site_ids_by_actual_type = {}
|
|
self.return_site_ids_by_returned_callable = {}
|
|
# Retain returns whose identity, callable, and declared type are canonical.
|
|
for key, record in return_sites_by_id.items():
|
|
# Skip malformed or orphaned return records.
|
|
if key != record.return_site_id or record.callable_id is None or record.declared_return_type_id not in self.type_definitions:
|
|
continue
|
|
self.return_sites_by_id[key] = record
|
|
self.return_site_ids_by_source.setdefault(record.source.location, set()).add(key)
|
|
self.return_site_ids_by_callable.setdefault(record.callable_id, set()).add(key)
|
|
self.return_site_ids_by_declared_type.setdefault(record.declared_return_type_id, set()).add(key)
|
|
# Index direct actual types only when syntax proves one.
|
|
if record.direct_actual_type_id is not None:
|
|
self.return_site_ids_by_actual_type.setdefault(record.direct_actual_type_id, set()).add(key)
|
|
# Index returned-callable provenance only when syntax proves one.
|
|
if record.returned_callable_id is not None:
|
|
self.return_site_ids_by_returned_callable.setdefault(record.returned_callable_id, set()).add(key)
|
|
|
|
self.indirect_references_by_id = {}
|
|
self.indirect_reference_ids_by_declaration_site = {}
|
|
self.indirect_reference_ids_by_binding_site = {}
|
|
self.indirect_reference_ids_by_return_site = {}
|
|
self.indirect_reference_ids_by_actual_type = {}
|
|
self.indirect_reference_ids_by_parent_relationship = {}
|
|
self.indirect_reference_ids_by_dependent_source = {}
|
|
self.indirect_reference_ids_by_dependency_source = {}
|
|
# Retain indirect references only when every supporting record is canonical.
|
|
for key, record in indirect_references_by_id.items():
|
|
# Skip references whose identity does not match their map key.
|
|
if key != record.indirect_reference_id:
|
|
continue
|
|
# Skip references without a canonical declaration and binding pair.
|
|
if record.declaration_site_id not in self.declaration_sites_by_id or record.binding_site_id not in self.binding_sites_by_id:
|
|
continue
|
|
# Skip references whose declared and actual types are unknown.
|
|
if record.declared_type_id not in self.type_definitions or record.actual_type_id not in self.type_definitions:
|
|
continue
|
|
# Skip references that point to unknown return provenance.
|
|
if any(item not in self.return_sites_by_id for item in record.return_site_ids):
|
|
continue
|
|
# Skip references that point to unknown parent relationships.
|
|
if any(item not in self.parent_relationships_by_id for item in record.parent_relationship_ids):
|
|
continue
|
|
binding = self.binding_sites_by_id[record.binding_site_id]
|
|
# Retain references only when the binding points to the same declaration.
|
|
if binding.declaration_site_id != record.declaration_site_id:
|
|
continue
|
|
declaration = self.declaration_sites_by_id[record.declaration_site_id]
|
|
self.indirect_references_by_id[key] = record
|
|
self.indirect_reference_ids_by_declaration_site.setdefault(record.declaration_site_id, set()).add(key)
|
|
self.indirect_reference_ids_by_binding_site.setdefault(record.binding_site_id, set()).add(key)
|
|
# Index every return site that contributes provenance to this reference.
|
|
for return_id in record.return_site_ids:
|
|
self.indirect_reference_ids_by_return_site.setdefault(return_id, set()).add(key)
|
|
self.indirect_reference_ids_by_actual_type.setdefault(record.actual_type_id, set()).add(key)
|
|
# Index every parent relationship that supports this reference.
|
|
for relationship_id in record.parent_relationship_ids:
|
|
self.indirect_reference_ids_by_parent_relationship.setdefault(relationship_id, set()).add(key)
|
|
self.indirect_reference_ids_by_dependent_source.setdefault(declaration.source.location, set()).add(key)
|
|
actual_source = self.type_definitions[record.actual_type_id].source.location
|
|
self.indirect_reference_ids_by_dependency_source.setdefault(actual_source, set()).add(key)
|
|
|
|
def _add_type(self, record: DependencyType) -> None:
|
|
"""Add one type record and index its defining source."""
|
|
self.type_definitions[record.type_id] = record
|
|
self.type_ids_by_source.setdefault(record.source.location, set()).add(record.type_id)
|
|
|
|
def _remove_type(self, type_id: str) -> None:
|
|
"""Remove one type record and its defining-source index entry."""
|
|
record = self.type_definitions.pop(type_id, None)
|
|
# Prune the source index only when a type record was actually removed.
|
|
if record is not None:
|
|
self._discard(self.type_ids_by_source, record.source.location, type_id)
|
|
|
|
def _add_parent(self, record: ParentRelationship) -> None:
|
|
"""Add one parent relationship and both direction indexes."""
|
|
self.parent_relationships_by_id[record.relationship_id] = record
|
|
self.parent_relationship_ids_by_child.setdefault(record.child_type_id, set()).add(record.relationship_id)
|
|
self.parent_relationship_ids_by_parent.setdefault(record.parent_type_id, set()).add(record.relationship_id)
|
|
|
|
def _remove_parent(self, relationship_id: str) -> None:
|
|
"""Remove one parent relationship and both direction indexes."""
|
|
record = self.parent_relationships_by_id.pop(relationship_id, None)
|
|
# Ignore an already-removed relationship so delta application stays idempotent.
|
|
if record is None:
|
|
return
|
|
self._discard(self.parent_relationship_ids_by_child, record.child_type_id, relationship_id)
|
|
self._discard(self.parent_relationship_ids_by_parent, record.parent_type_id, relationship_id)
|
|
|
|
def _add_declaration(self, record: DeclarationSite) -> None:
|
|
"""Add one declaration site and its lookup indexes."""
|
|
self.declaration_sites_by_id[record.declaration_site_id] = record
|
|
self.declaration_site_ids_by_source.setdefault(record.source.location, set()).add(record.declaration_site_id)
|
|
self.declaration_site_ids_by_declared_type.setdefault(record.declared_type_id, set()).add(record.declaration_site_id)
|
|
# Index callable ownership only when the declaration is nested in one.
|
|
if record.callable_id is not None:
|
|
self.declaration_site_ids_by_callable.setdefault(record.callable_id, set()).add(record.declaration_site_id)
|
|
|
|
def _remove_declaration(self, declaration_id: str) -> None:
|
|
"""Remove one declaration site and its lookup indexes."""
|
|
record = self.declaration_sites_by_id.pop(declaration_id, None)
|
|
# Ignore an already-removed declaration so delta application stays idempotent.
|
|
if record is None:
|
|
return
|
|
self._discard(self.declaration_site_ids_by_source, record.source.location, declaration_id)
|
|
self._discard(self.declaration_site_ids_by_declared_type, record.declared_type_id, declaration_id)
|
|
# Remove callable ownership only when the declaration had a callable owner.
|
|
if record.callable_id is not None:
|
|
self._discard(self.declaration_site_ids_by_callable, record.callable_id, declaration_id)
|
|
|
|
def _add_binding(self, record: BindingSite) -> None:
|
|
"""Add one binding site and its lookup indexes."""
|
|
self.binding_sites_by_id[record.binding_site_id] = record
|
|
self.binding_site_ids_by_source.setdefault(record.source.location, set()).add(record.binding_site_id)
|
|
self.binding_site_ids_by_declaration_site.setdefault(record.declaration_site_id, set()).add(record.binding_site_id)
|
|
# Index direct actual types only when syntax proves one.
|
|
if record.direct_actual_type_id is not None:
|
|
self.binding_site_ids_by_actual_type.setdefault(record.direct_actual_type_id, set()).add(record.binding_site_id)
|
|
# Index returned-callable provenance only when syntax proves one.
|
|
if record.returned_callable_id is not None:
|
|
self.binding_site_ids_by_returned_callable.setdefault(record.returned_callable_id, set()).add(record.binding_site_id)
|
|
|
|
def _remove_binding(self, binding_id: str) -> None:
|
|
"""Remove one binding site and its lookup indexes."""
|
|
record = self.binding_sites_by_id.pop(binding_id, None)
|
|
# Ignore an already-removed binding so delta application stays idempotent.
|
|
if record is None:
|
|
return
|
|
self._discard(self.binding_site_ids_by_source, record.source.location, binding_id)
|
|
self._discard(self.binding_site_ids_by_declaration_site, record.declaration_site_id, binding_id)
|
|
# Remove direct actual-type provenance only when syntax supplied one.
|
|
if record.direct_actual_type_id is not None:
|
|
self._discard(self.binding_site_ids_by_actual_type, record.direct_actual_type_id, binding_id)
|
|
# Remove returned-callable provenance only when syntax supplied one.
|
|
if record.returned_callable_id is not None:
|
|
self._discard(self.binding_site_ids_by_returned_callable, record.returned_callable_id, binding_id)
|
|
|
|
def _add_return(self, record: ReturnSite) -> None:
|
|
"""Add one return site and its provenance indexes."""
|
|
self.return_sites_by_id[record.return_site_id] = record
|
|
self.return_site_ids_by_source.setdefault(record.source.location, set()).add(record.return_site_id)
|
|
self.return_site_ids_by_callable.setdefault(record.callable_id, set()).add(record.return_site_id)
|
|
self.return_site_ids_by_declared_type.setdefault(record.declared_return_type_id, set()).add(record.return_site_id)
|
|
# Index direct actual types only when syntax proves one.
|
|
if record.direct_actual_type_id is not None:
|
|
self.return_site_ids_by_actual_type.setdefault(record.direct_actual_type_id, set()).add(record.return_site_id)
|
|
# Index returned-callable provenance only when syntax proves one.
|
|
if record.returned_callable_id is not None:
|
|
self.return_site_ids_by_returned_callable.setdefault(record.returned_callable_id, set()).add(record.return_site_id)
|
|
|
|
def _remove_return(self, return_id: str) -> None:
|
|
"""Remove one return site and its provenance indexes."""
|
|
record = self.return_sites_by_id.pop(return_id, None)
|
|
# Ignore an already-removed return site so delta application stays idempotent.
|
|
if record is None:
|
|
return
|
|
self._discard(self.return_site_ids_by_source, record.source.location, return_id)
|
|
self._discard(self.return_site_ids_by_callable, record.callable_id, return_id)
|
|
self._discard(self.return_site_ids_by_declared_type, record.declared_return_type_id, return_id)
|
|
# Remove direct actual-type provenance only when syntax supplied one.
|
|
if record.direct_actual_type_id is not None:
|
|
self._discard(self.return_site_ids_by_actual_type, record.direct_actual_type_id, return_id)
|
|
# Remove returned-callable provenance only when syntax supplied one.
|
|
if record.returned_callable_id is not None:
|
|
self._discard(self.return_site_ids_by_returned_callable, record.returned_callable_id, return_id)
|
|
|
|
def _add_reference(self, record: IndirectReference) -> None:
|
|
"""Add one indirect reference and every relationship index it supports."""
|
|
declaration = self.declaration_sites_by_id[record.declaration_site_id]
|
|
actual_source = self.type_definitions[record.actual_type_id].source.location
|
|
self.indirect_references_by_id[record.indirect_reference_id] = record
|
|
self.indirect_reference_ids_by_declaration_site.setdefault(record.declaration_site_id, set()).add(record.indirect_reference_id)
|
|
self.indirect_reference_ids_by_binding_site.setdefault(record.binding_site_id, set()).add(record.indirect_reference_id)
|
|
# Index every return site that contributes provenance to this reference.
|
|
for return_id in record.return_site_ids:
|
|
self.indirect_reference_ids_by_return_site.setdefault(return_id, set()).add(record.indirect_reference_id)
|
|
self.indirect_reference_ids_by_actual_type.setdefault(record.actual_type_id, set()).add(record.indirect_reference_id)
|
|
# Index every parent relationship that supports this reference.
|
|
for relationship_id in record.parent_relationship_ids:
|
|
self.indirect_reference_ids_by_parent_relationship.setdefault(relationship_id, set()).add(record.indirect_reference_id)
|
|
self.indirect_reference_ids_by_dependent_source.setdefault(declaration.source.location, set()).add(record.indirect_reference_id)
|
|
self.indirect_reference_ids_by_dependency_source.setdefault(actual_source, set()).add(record.indirect_reference_id)
|
|
|
|
def _remove_reference(self, reference_id: str) -> None:
|
|
"""Remove one indirect reference and every relationship index it supports."""
|
|
record = self.indirect_references_by_id.pop(reference_id, None)
|
|
# Ignore an already-removed reference so delta application stays idempotent.
|
|
if record is None:
|
|
return
|
|
declaration = self.declaration_sites_by_id.get(record.declaration_site_id)
|
|
self._discard(self.indirect_reference_ids_by_declaration_site, record.declaration_site_id, reference_id)
|
|
self._discard(self.indirect_reference_ids_by_binding_site, record.binding_site_id, reference_id)
|
|
# Remove each return-site provenance edge owned by this reference.
|
|
for return_id in record.return_site_ids:
|
|
self._discard(self.indirect_reference_ids_by_return_site, return_id, reference_id)
|
|
self._discard(self.indirect_reference_ids_by_actual_type, record.actual_type_id, reference_id)
|
|
# Remove each parent-relationship provenance edge owned by this reference.
|
|
for relationship_id in record.parent_relationship_ids:
|
|
self._discard(self.indirect_reference_ids_by_parent_relationship, relationship_id, reference_id)
|
|
# Remove the dependent-source index only when its declaration remains available.
|
|
if declaration is not None:
|
|
self._discard(self.indirect_reference_ids_by_dependent_source, declaration.source.location, reference_id)
|
|
actual = self.type_definitions.get(record.actual_type_id)
|
|
# Remove the dependency-source index only when its actual type remains available.
|
|
if actual is not None:
|
|
self._discard(self.indirect_reference_ids_by_dependency_source, actual.source.location, reference_id)
|
|
|
|
def _apply_dependency_delta(self, delta: DependencyDelta) -> None:
|
|
"""Apply one validated changed-only delta without rebuilding unrelated state."""
|
|
self._assert_mutable()
|
|
# Remove source records and their dependent test associations first.
|
|
for location in delta.source_removed:
|
|
self._remove_test_mappings_for_source(location)
|
|
self.source_by_location.pop(location, None)
|
|
self.source_by_location.update(delta.source_added)
|
|
self.source_by_location.update(delta.source_changed)
|
|
# Remove test records and their source associations before replacing test indexes.
|
|
for location in delta.test_removed:
|
|
self._remove_test_mappings_for_test(location)
|
|
self.test_by_location.pop(location, None)
|
|
self.test_by_location.update(delta.test_added)
|
|
self.test_by_location.update(delta.test_changed)
|
|
|
|
# Remove obsolete static edges before admitting newly observed edges.
|
|
for dependent, dependency in delta.static_pairs_removed:
|
|
self._remove_source_dependency(dependent, dependency)
|
|
# Add newly observed static edges after obsolete edges are gone.
|
|
for dependent, dependency in delta.static_pairs_added:
|
|
self._add_source_dependency(dependent, dependency)
|
|
|
|
# Remove deleted evidence in dependency order so indexes cannot retain orphans.
|
|
for reference_id in delta.references_removed:
|
|
self._remove_reference(reference_id)
|
|
# Remove changed references before their replacement evidence is indexed.
|
|
for reference in delta.references_changed.values():
|
|
self._remove_reference(reference.indirect_reference_id)
|
|
# Remove parent relationships before types that may own their endpoints.
|
|
for relationship_id in delta.relationships_removed:
|
|
self._remove_parent(relationship_id)
|
|
# Remove declarations before bindings and references that depend on them.
|
|
for declaration_id in delta.declarations_removed:
|
|
self._remove_declaration(declaration_id)
|
|
# Remove bindings before references that depend on them.
|
|
for binding_id in delta.bindings_removed:
|
|
self._remove_binding(binding_id)
|
|
# Remove return sites before references that depend on their provenance.
|
|
for return_id in delta.returns_removed:
|
|
self._remove_return(return_id)
|
|
# Remove types after all evidence that can reference them is detached.
|
|
for type_id in delta.types_removed:
|
|
self._remove_type(type_id)
|
|
|
|
# Remove changed evidence before adding its replacement records.
|
|
for type_id in delta.types_changed:
|
|
self._remove_type(type_id)
|
|
# Remove changed relationships before their replacement edges are indexed.
|
|
for relationship_id in delta.relationships_changed:
|
|
self._remove_parent(relationship_id)
|
|
# Remove changed declarations before their replacement records are indexed.
|
|
for declaration_id in delta.declarations_changed:
|
|
self._remove_declaration(declaration_id)
|
|
# Remove changed bindings before their replacement records are indexed.
|
|
for binding_id in delta.bindings_changed:
|
|
self._remove_binding(binding_id)
|
|
# Remove changed returns before their replacement records are indexed.
|
|
for return_id in delta.returns_changed:
|
|
self._remove_return(return_id)
|
|
|
|
# Add changed and new type records before relationships that reference them.
|
|
for record in delta.types_added.values():
|
|
self._add_type(record)
|
|
# Restore changed type records before dependent evidence is admitted.
|
|
for record in delta.types_changed.values():
|
|
self._add_type(record)
|
|
# Add relationships only when both endpoint types are now canonical.
|
|
for record in delta.relationships_added.values():
|
|
# Protect relationship indexes from unresolved endpoints.
|
|
if record.parent_type_id in self.type_definitions and record.child_type_id in self.type_definitions:
|
|
self._add_parent(record)
|
|
# Restore changed relationships after their endpoint types are available.
|
|
for record in delta.relationships_changed.values():
|
|
# Protect relationship indexes from unresolved endpoints.
|
|
if record.parent_type_id in self.type_definitions and record.child_type_id in self.type_definitions:
|
|
self._add_parent(record)
|
|
# Add declarations only when their declared types are now canonical.
|
|
for record in delta.declarations_added.values():
|
|
# Protect declaration indexes from unresolved declared types.
|
|
if record.declared_type_id in self.type_definitions:
|
|
self._add_declaration(record)
|
|
# Restore changed declarations after their declared types are available.
|
|
for record in delta.declarations_changed.values():
|
|
# Protect declaration indexes from unresolved declared types.
|
|
if record.declared_type_id in self.type_definitions:
|
|
self._add_declaration(record)
|
|
# Add bindings only when their declaration owners are now canonical.
|
|
for record in delta.bindings_added.values():
|
|
# Protect binding indexes from unresolved declarations.
|
|
if record.declaration_site_id in self.declaration_sites_by_id:
|
|
self._add_binding(record)
|
|
# Restore changed bindings after their declaration owners are available.
|
|
for record in delta.bindings_changed.values():
|
|
# Protect binding indexes from unresolved declarations.
|
|
if record.declaration_site_id in self.declaration_sites_by_id:
|
|
self._add_binding(record)
|
|
# Add returns only when their declared return types are now canonical.
|
|
for record in delta.returns_added.values():
|
|
# Protect return indexes from unresolved declared types.
|
|
if record.declared_return_type_id in self.type_definitions:
|
|
self._add_return(record)
|
|
# Restore changed returns after their declared types are available.
|
|
for record in delta.returns_changed.values():
|
|
# Protect return indexes from unresolved declared types.
|
|
if record.declared_return_type_id in self.type_definitions:
|
|
self._add_return(record)
|
|
# Add indirect references only when their declaration and binding owners exist.
|
|
for record in delta.references_added.values():
|
|
# Protect reference indexes from unresolved supporting records.
|
|
if record.declaration_site_id in self.declaration_sites_by_id and record.binding_site_id in self.binding_sites_by_id:
|
|
self._add_reference(record)
|
|
# Restore changed references after their supporting records are available.
|
|
for record in delta.references_changed.values():
|
|
# Protect reference indexes from unresolved supporting records.
|
|
if record.declaration_site_id in self.declaration_sites_by_id and record.binding_site_id in self.binding_sites_by_id:
|
|
self._add_reference(record)
|
|
|
|
# Remove stale source/test associations after evidence records are reconciled.
|
|
for source_location, test_location in delta.tests_for_source_removed:
|
|
self._remove_source_test_mapping(source_location, test_location)
|
|
# Add current source/test associations after stale links are removed.
|
|
for source_location, test_location in delta.tests_for_source_added:
|
|
self._add_source_test_mapping(source_location, test_location)
|