async def scoped_neighborhood(
backend: Any, scope: TenantScope, center_id: str, *, depth: int,
min_weight: float,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Return reachable nodes and induced edges visible to this scope.
Backends must explicitly declare the CORE-74 edge-owner query contract.
An inherited marker does not certify a custom query implementation.
"""
if type(backend).__dict__.get("supports_visible_edge_owners") is not True:
raise RuntimeError("graph backend lacks visible edge owner reads")
visited = {center_id}
neighbors: list[dict[str, Any]] = []
queue = deque([(center_id, 0)])
while queue:
current, distance = queue.popleft()
if distance >= depth:
continue
incident = await backend.query_edges(
scope, {"incident_node_id": current, "visible_edge_owners": True},
limit=None,
)
for edge in incident:
source, target = str(edge.source), str(edge.target)
if current not in (source, target):
continue
neighbor_id = target if source == current else source
if neighbor_id in visited:
continue
node = await backend.get_node(scope, NodeId(neighbor_id))
if node is None or is_retracted(node.properties):
continue
visited.add(neighbor_id)
neighbors.append({
"id": str(node.id),
"label": str(node.label),
"layer": node.layer.value if hasattr(node.layer, "value") else str(node.layer),
"importance": node.importance,
"depth": distance + 1,
})
queue.append((neighbor_id, distance + 1))
induced = await backend.query_edges(
scope, {"induced_node_ids": tuple(sorted(visited)),
"visible_edge_owners": True},
limit=None,
)
edges: list[dict[str, Any]] = []
seen_edges: set[str] = set()
for edge in induced:
edge_id = str(edge.id)
source, target = str(edge.source), str(edge.target)
if (edge_id in seen_edges or source not in visited or target not in visited
or edge.weight < min_weight):
continue
seen_edges.add(edge_id)
edges.append({
"id": edge_id, "source": source, "target": target,
"relationship": edge.relationship, "weight": edge.weight,
})
return neighbors, edges