async def audit(
self, scope: MemoryScope, *, min_edge_weight: float = 1.5, admit: Any = None,
) -> dict[str, Any]:
"""Count all visible semantic nodes and at most 10,000 visible edges."""
scoped = tenant_scope(scope)
nodes = await self._graph.query_nodes(scoped, layer=MemoryLayer.SEMANTIC)
edges = await self._graph.list_edges(scoped, limit=10_000)
from symfonic.capabilities.memory.graph_owner_reads import admitted_edges
edges = await admitted_edges(self._graph.backend, scope, edges, admit)
total_nodes, total_edges = len(nodes), len(edges)
density = (
round(2 * total_edges / (total_nodes * (total_nodes - 1)), 4)
if total_nodes > 1 else 0.0
)
stale_before = datetime.now(UTC) - timedelta(days=7)
orphans: list[dict[str, Any]] = []
for node in nodes:
neighbors = await self._graph.get_neighbors(scoped, node.id)
edge_count = len(neighbors)
if edge_count <= 1 and node.importance < 4.0 and node.updated_at < stale_before:
reason = ", ".join((
"no edges" if edge_count == 0 else "only 1 edge",
f"importance {node.importance} < 4.0",
"stale (>7 days)",
))
orphans.append({
"id": str(node.id), "label": str(node.label),
"importance": node.importance, "edge_count": edge_count,
"reason": reason,
})
weak = [
{
"id": str(edge.id), "source": str(edge.source),
"target": str(edge.target), "weight": edge.weight,
"reason": f"weight {edge.weight} below {min_edge_weight} threshold",
}
for edge in edges if edge.weight < min_edge_weight
]
return {
"total_nodes": total_nodes, "total_edges": total_edges,
"density": density, "orphan_nodes": orphans, "weak_edges": weak,
"recommendations": {
"prune_orphans": len(orphans), "prune_weak_edges": len(weak),
},
}