def create_graph_router(agent: SymfonicAgent) -> APIRouter:
"""Create a sub-router for graph query endpoints.
All endpoints are prefixed under the parent router's prefix and
follow the same ``get_tenant_scope`` dependency pattern.
"""
router = APIRouter(tags=["graph"])
def _get_graph_store(scope: FrameworkTenantScope):
"""Resolve the graph store from the semantic layer."""
semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
if semantic is None or not hasattr(semantic, "_graph"):
raise HTTPException(404, "Semantic layer not available")
return semantic._graph # type: ignore[union-attr]
# ------------------------------------------------------------------
# GET /graph/nodes/{node_id}/neighborhood
# ------------------------------------------------------------------
@router.get("/graph/nodes/{node_id}/neighborhood")
async def neighborhood(
node_id: str,
depth: int = 2,
min_weight: float = 0,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> NeighborhoodResponse:
"""Return a node's neighborhood up to *depth* hops."""
try:
graph_store = _get_graph_store(scope)
memory_scope = scope.to_memory_scope()
center_node = await graph_store._backend.get_node(
memory_scope, NodeId(node_id),
)
# A soft-retracted node is treated as absent by the graph API
# (same as the store's filtered reads) -- ``_backend`` calls here
# bypass GraphMemoryStore's retraction filter, so guard inline.
if center_node is None or _is_retracted(center_node):
raise HTTPException(404, f"Node {node_id} not found")
# Filtered BFS is the single source of truth for BOTH which
# nodes are reachable AND their depth. It expands only through
# ``graph_store.get_neighbors`` (which excludes soft-retracted
# nodes), so a node reachable ONLY through a retracted
# intermediate is never enqueued. The previous
# ``_backend.traverse`` + post-filter approach leaked such
# nodes with a wrong depth of 0 (3rd review, P1): for
# A -> retracted B -> C the unfiltered traversal reached C
# through B, and dropping B from the result still exposed C.
from collections import deque
depth_map: dict[str, int] = {node_id: 0}
reachable: dict[str, Any] = {}
bfs_queue: deque[tuple[str, int]] = deque([(node_id, 0)])
visited: set[str] = {node_id}
while bfs_queue:
curr_id, curr_depth = bfs_queue.popleft()
if curr_depth >= depth:
continue
nbrs = await graph_store.get_neighbors(
memory_scope, NodeId(curr_id),
)
for nbr in nbrs:
nid = str(nbr.id)
if nid not in visited:
visited.add(nid)
depth_map[nid] = curr_depth + 1
reachable[nid] = nbr
bfs_queue.append((nid, curr_depth + 1))
# Collect neighbors (center is not in ``reachable`` -- it seeds
# ``visited`` before the loop, so it can never be re-added).
neighbors: list[NeighborNode] = [
NeighborNode(
id=nid,
label=str(node.label),
layer=node.layer.value if hasattr(node.layer, "value") else str(node.layer),
importance=node.importance,
depth=depth_map[nid],
)
for nid, node in reachable.items()
]
# Collect edges between reachable nodes (center + neighbors)
edges: list[EdgeData] = []
seen_edges: set[str] = set()
all_node_ids = visited
edge_list = await graph_store.list_edges(memory_scope)
for edge in edge_list:
eid = str(edge.id)
src, tgt = str(edge.source), str(edge.target)
if eid in seen_edges:
continue
if src in all_node_ids and tgt in all_node_ids and edge.weight >= min_weight:
seen_edges.add(eid)
edges.append(EdgeData(
id=eid,
source=src,
target=tgt,
relationship=edge.relationship,
weight=edge.weight,
))
center_data = {
"id": str(center_node.id),
"label": str(center_node.label),
"layer": (
center_node.layer.value
if hasattr(center_node.layer, "value")
else str(center_node.layer)
),
"importance": center_node.importance,
}
return NeighborhoodResponse(
center=center_data,
neighbors=neighbors,
edges=edges,
)
except HTTPException:
raise
except SecurityScopeError as exc:
raise HTTPException(403, "Access denied") from exc
except ScopeValidationError as exc:
raise HTTPException(400, str(exc)) from exc
except Exception as exc:
logger.exception("Neighborhood query failed")
raise HTTPException(500, "Internal server error") from exc
# ------------------------------------------------------------------
# GET /graph/paths
# ------------------------------------------------------------------
@router.get("/graph/paths")
async def find_path(
start: str,
end: str,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> PathResponse:
"""Find the shortest path between two nodes."""
try:
graph_store = _get_graph_store(scope)
memory_scope = scope.to_memory_scope()
# Verify both nodes exist
start_node = await graph_store._backend.get_node(
memory_scope, NodeId(start),
)
if start_node is None or _is_retracted(start_node):
raise HTTPException(404, f"Node {start} not found")
end_node = await graph_store._backend.get_node(
memory_scope, NodeId(end),
)
if end_node is None or _is_retracted(end_node):
raise HTTPException(404, f"Node {end} not found")
traversal = GraphTraversal(graph_store)
path_ids = await traversal.shortest_path(
memory_scope, NodeId(start), NodeId(end),
)
if path_ids is None:
return PathResponse(path=[], edges=[], hops=0)
# Resolve node data for the path
path_nodes: list[PathNode] = []
for nid in path_ids:
node = await graph_store._backend.get_node(
memory_scope, NodeId(str(nid)),
)
if node is not None and not _is_retracted(node):
path_nodes.append(PathNode(
id=str(node.id),
label=str(node.label),
layer=node.layer.value if hasattr(node.layer, "value") else str(node.layer),
))
# Collect connecting edges for consecutive pairs
edges: list[EdgeData] = []
all_edges = await graph_store.list_edges(memory_scope)
for i in range(len(path_ids) - 1):
src, tgt = str(path_ids[i]), str(path_ids[i + 1])
for edge in all_edges:
es, et = str(edge.source), str(edge.target)
if (es == src and et == tgt) or (es == tgt and et == src):
edges.append(EdgeData(
id=str(edge.id),
source=es,
target=et,
relationship=edge.relationship,
weight=edge.weight,
))
break
return PathResponse(
path=path_nodes,
edges=edges,
hops=len(path_ids) - 1,
)
except HTTPException:
raise
except SecurityScopeError as exc:
raise HTTPException(403, "Access denied") from exc
except ScopeValidationError as exc:
raise HTTPException(400, str(exc)) from exc
except Exception as exc:
logger.exception("Pathfinding failed")
raise HTTPException(500, "Internal server error") from exc
# ------------------------------------------------------------------
# Edge CRUD
# ------------------------------------------------------------------
@router.get("/edges")
async def list_edges(
limit: int = 50,
offset: int = 0,
relationship: str | None = None,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[EdgeData]:
"""List edges for the tenant with optional filtering and pagination.
Query parameters:
- ``limit``: max edges to return (default 50)
- ``offset``: number of edges to skip (default 0)
- ``relationship``: optional exact-match filter on relationship type
"""
try:
if limit <= 0:
return []
graph_store = _get_graph_store(scope)
memory_scope = scope.to_memory_scope()
# Retraction hygiene (3rd review, P2): an edge whose source or
# target is a soft-retracted node is a false/dangling
# relationship during the grace window before consolidation
# prunes it. Drop only edges touching a RETRACTED node -- NOT
# edges to ids that merely lack a materialised node row (those
# are a pre-existing, separate concern and are left untouched).
all_nodes = await graph_store.query_nodes(
memory_scope, include_retracted=True,
)
retracted_ids = {
str(n.id) for n in all_nodes
if is_retracted(n.properties)
}
# ``offset``/``limit`` count LIVE edges (neither endpoint
# retracted), not raw backend rows. Filtering strictly after a
# backend-paginated fetch would let a retracted-endpoint edge
# inside the requested page shrink or empty it even though live
# edges exist further down. Over-fetch from offset=0 with a
# doubling window and refill until ``offset + limit`` live edges
# are collected or the backend is exhausted -- mirroring
# ``GraphMemoryStore.query_nodes``'s limit-refill fix -- then
# slice the live list to the requested page.
want = offset + limit
fetch = want
while True:
batch = await graph_store.list_edges(
memory_scope,
limit=fetch,
offset=0,
relationship=relationship,
)
live = [
e for e in batch
if str(e.source) not in retracted_ids
and str(e.target) not in retracted_ids
]
if len(live) >= want or len(batch) < fetch:
break
fetch *= 2
page = live[offset:offset + limit]
return [
EdgeData(
id=str(e.id),
source=str(e.source),
target=str(e.target),
relationship=e.relationship,
weight=e.weight,
)
for e in page
]
except HTTPException:
raise
except Exception as exc:
logger.exception("Edge listing failed")
raise HTTPException(500, "Internal server error") from exc
@router.post("/edges")
async def create_edge(
request: CreateEdgeRequest,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> EdgeData:
"""Create a new edge between two nodes."""
from symfonic.memory.models.edge import MemoryEdge
try:
graph_store = _get_graph_store(scope)
memory_scope = scope.to_memory_scope()
edge = MemoryEdge(
source=NodeId(request.source),
target=NodeId(request.target),
relationship=request.relationship,
tenant_id=scope.tenant_id,
weight=request.weight,
)
persisted = await graph_store.add_edge(memory_scope, edge)
return EdgeData(
id=str(persisted.id),
source=str(persisted.source),
target=str(persisted.target),
relationship=persisted.relationship,
weight=persisted.weight,
)
except HTTPException:
raise
except Exception as exc:
logger.exception("Edge creation failed")
raise HTTPException(500, "Internal server error") from exc
@router.delete("/edges/{edge_id}")
async def delete_edge(
edge_id: str,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, str]:
"""Delete an edge by ID."""
from symfonic.memory.types import EdgeId
try:
graph_store = _get_graph_store(scope)
memory_scope = scope.to_memory_scope()
await graph_store.delete_edge(memory_scope, EdgeId(edge_id))
return {"deleted": edge_id}
except HTTPException:
raise
except Exception as exc:
logger.exception("Edge deletion failed")
raise HTTPException(500, "Internal server error") from exc
# ------------------------------------------------------------------
# GET /graph/clusters
# ------------------------------------------------------------------
@router.get("/graph/clusters")
async def list_clusters(
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> list[ClusterProjection]:
"""Project semantic nodes into clusters by label prefix."""
try:
graph_store = _get_graph_store(scope)
memory_scope = scope.to_memory_scope()
nodes = await graph_store.query_nodes(
memory_scope, layer=MemoryLayer.SEMANTIC,
)
groups: dict[str, list[tuple[str, str, float]]] = {}
for node in nodes:
label = str(node.label)
if ":" in label:
prefix = label.split(":")[0].strip()
elif label:
prefix = label.split()[0]
else:
prefix = "unknown"
groups.setdefault(prefix, []).append(
(str(node.id), label, node.importance)
)
clusters: list[ClusterProjection] = []
for prefix, members in sorted(
groups.items(), key=lambda kv: len(kv[1]), reverse=True,
):
member_list = [
ClusterMember(id=m[0], label=m[1]) for m in members
]
# Representative is the highest-importance member
best = max(members, key=lambda m: m[2])
clusters.append(ClusterProjection(
prefix=prefix,
count=len(members),
members=member_list,
representative=ClusterMember(id=best[0], label=best[1]),
))
return clusters
except HTTPException:
raise
except Exception as exc:
logger.exception("Cluster projection failed")
raise HTTPException(500, "Internal server error") from exc
return router