async def shortest_path(
graph: Any, scope: MemoryScope, start: str, end: str, *, admit: Any = None,
) -> dict[str, Any]:
"""Read an isolated published path, or raise if the answer is unknown.
Missing and hidden endpoints raise LookupError. Exhausted edge/node budgets
and backend failures raise GraphPathUnavailable, never an empty-path answer.
Like the backend read contract, nodes must lie on the caller's ancestor
chain; an edge naming a sibling or descendant cannot widen that visibility.
Pagination uses the public offset port, so concurrent writes are not a
transactional snapshot; callers needing snapshots must provide one below.
"""
if not start.strip() or not end.strip():
raise ValueError("start and end must name nodes")
trusted = tenant_scope(scope)
nodes: dict[str, Any] = {}
async def read(node_id: str) -> Any:
if node_id not in nodes:
if len(nodes) >= MAX_PATH_NODES:
raise GraphPathUnavailable("graph path node budget exhausted")
node = await graph.get_node(trusted, NodeId(node_id))
if not _visible(node, node_id, scope) or (
admit is not None and not await admit(node)
):
node = None
nodes[node_id] = node
return nodes[node_id]
async def finish(path: list[str], edges: list[Any]) -> dict[str, Any]:
# Do not serialize cached labels from a row whose visibility changed
# while the edge scan or BFS was in flight.
verified = []
for node_id in path:
node = await graph.get_node(trusted, NodeId(node_id))
if not _visible(node, node_id, scope) or (
admit is not None and not await admit(node)
):
raise GraphPathUnavailable("graph path changed during traversal")
verified.append(node)
for index, edge in enumerate(edges):
if edge.tenant_id != scope.tenant or {
str(edge.source), str(edge.target)
} != {path[index], path[index + 1]}:
raise GraphPathUnavailable("graph path edge changed during traversal")
return _body(verified, edges)
try:
for endpoint in (start, end):
if await read(endpoint) is None:
raise GraphPathEndpointMissing("graph path endpoint not found")
if start == end:
return await finish([start], [])
edge_reader = next(
(base for base in type(graph).__mro__ if "query_edges" in base.__dict__),
None,
)
if edge_reader is None or edge_reader.__dict__.get(
"supports_visible_edge_owners"
) is not True:
raise GraphPathUnavailable("graph path requires scoped edge reads")
adjacent: dict[str, list[tuple[str, Any]]] = {}
offset = 0
seen_edges: set[str] = set()
while True:
# One extra row establishes exhaustion at the exact budget.
limit = min(EDGE_PAGE_SIZE, MAX_PATH_EDGES - offset + 1)
page = await graph.query_edges(
trusted, {"visible_edge_owners": True}, limit=limit, offset=offset,
)
if len(page) > limit or offset + len(page) > MAX_PATH_EDGES:
raise GraphPathUnavailable("graph path edge budget exhausted")
for edge in page:
edge_id = str(edge.id)
if edge_id in seen_edges:
raise GraphPathUnavailable("graph path pagination repeated an edge")
seen_edges.add(edge_id)
if edge.tenant_id != scope.tenant:
continue
source, target = str(edge.source), str(edge.target)
adjacent.setdefault(source, []).append((target, edge))
adjacent.setdefault(target, []).append((source, edge))
offset += len(page)
if len(page) < limit:
break
parents: dict[str, tuple[str, Any] | None] = {start: None}
queue = deque([start])
while queue:
current = queue.popleft()
for neighbor, edge in adjacent.get(current, ()):
if neighbor in parents or await read(neighbor) is None:
continue
parents[neighbor] = (current, edge)
if neighbor == end:
path, edges = [end], []
while parents[path[-1]] is not None:
previous, connecting = parents[path[-1]]
path.append(previous)
edges.append(connecting)
return await finish(list(reversed(path)), list(reversed(edges)))
queue.append(neighbor)
return _body([], [])
except (GraphPathEndpointMissing, GraphPathUnavailable):
raise
except Exception as failed:
raise GraphPathUnavailable("graph path backend read failed") from failed