use std::collections::{BTreeMap, BTreeSet};
use fnx_classes::Graph;
use fnx_runtime::CompatibilityMode;
use serde::Serialize;
pub const GRAPH_DIFF_SCHEMA_V1: &str = "ee.graph.diff.v1";
pub const GRAPH_DIFF_DETAIL_CAP: usize = 64;
pub const GRAPH_DIFF_MOVERS_CAP: usize = 10;
pub const COMMUNITY_MATCH_JACCARD_THRESHOLD: f64 = 0.5;
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DiffEdge {
pub key: String,
pub source: String,
pub target: String,
pub relation: String,
pub directed: bool,
}
#[derive(Clone, Debug, PartialEq)]
pub struct SnapshotGraphView {
pub snapshot_id: String,
pub snapshot_version: u32,
pub created_at: String,
pub nodes: BTreeMap<String, Option<f64>>,
pub edges: BTreeMap<String, DiffEdge>,
}
#[must_use]
pub fn edge_key(source: &str, target: &str, relation: &str, directed: bool) -> String {
let (a, b) = if directed || source <= target {
(source, target)
} else {
(target, source)
};
let canonical = format!("{a}|{relation}|{b}|{directed}");
let hash = blake3::hash(canonical.as_bytes()).to_hex();
format!("edge_{}", &hash.as_str()[..26])
}
pub fn parse_snapshot_graph(
snapshot_id: &str,
snapshot_version: u32,
created_at: &str,
metrics_json: &str,
) -> Result<SnapshotGraphView, String> {
let metrics: serde_json::Value = serde_json::from_str(metrics_json)
.map_err(|error| format!("snapshot {snapshot_id} metrics_json did not parse: {error}"))?;
let node_values = metrics
.get("nodes")
.or_else(|| metrics.pointer("/graph/nodes"))
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default();
let mut nodes = BTreeMap::new();
for node in &node_values {
let Some(id) = node
.get("id")
.or_else(|| node.get("memoryId"))
.and_then(serde_json::Value::as_str)
else {
continue;
};
let pagerank = node.get("pagerank").and_then(serde_json::Value::as_f64);
nodes.insert(id.to_owned(), pagerank);
}
let edge_values = metrics
.get("edges")
.or_else(|| metrics.pointer("/graph/edges"))
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default();
let mut edges = BTreeMap::new();
for edge in &edge_values {
let source = edge
.get("source")
.or_else(|| edge.get("sourceMemoryId"))
.and_then(serde_json::Value::as_str);
let target = edge
.get("target")
.or_else(|| edge.get("targetMemoryId"))
.and_then(serde_json::Value::as_str);
let (Some(source), Some(target)) = (source, target) else {
continue;
};
let relation = edge
.get("relation")
.and_then(serde_json::Value::as_str)
.unwrap_or("related");
let directed = edge
.get("directed")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let key = edge_key(source, target, relation, directed);
edges.insert(
key.clone(),
DiffEdge {
key,
source: source.to_owned(),
target: target.to_owned(),
relation: relation.to_owned(),
directed,
},
);
}
Ok(SnapshotGraphView {
snapshot_id: snapshot_id.to_owned(),
snapshot_version,
created_at: created_at.to_owned(),
nodes,
edges,
})
}
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CommunityMatch {
pub from_fingerprint: String,
pub to_fingerprint: String,
pub jaccard: f64,
pub joined: Vec<String>,
pub left: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CommunityEndpoint {
pub fingerprint: String,
pub size: usize,
pub members: Vec<String>,
}
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CommunityDelta {
pub matched: Vec<CommunityMatch>,
pub births: Vec<CommunityEndpoint>,
pub deaths: Vec<CommunityEndpoint>,
pub unchanged: usize,
}
fn louvain_from_edges(edges: &BTreeMap<String, DiffEdge>) -> Vec<BTreeSet<String>> {
let mut graph = Graph::new(CompatibilityMode::Strict);
for edge in edges.values() {
let _ = graph.extend_edges_unrecorded([(edge.source.as_str(), edge.target.as_str())]);
}
crate::graph::health::detect_louvain_communities(&graph)
.into_iter()
.map(|members| members.into_iter().collect::<BTreeSet<String>>())
.filter(|members| !members.is_empty())
.collect()
}
fn jaccard(a: &BTreeSet<String>, b: &BTreeSet<String>) -> f64 {
let intersection = a.intersection(b).count();
let union = a.len() + b.len() - intersection;
if union == 0 {
0.0
} else {
intersection as f64 / union as f64
}
}
fn fingerprint(members: &BTreeSet<String>) -> String {
members
.iter()
.next()
.cloned()
.unwrap_or_else(|| "<empty>".to_owned())
}
fn endpoint(members: &BTreeSet<String>) -> CommunityEndpoint {
let mut listed: Vec<String> = members.iter().cloned().collect();
listed.truncate(GRAPH_DIFF_DETAIL_CAP);
CommunityEndpoint {
fingerprint: fingerprint(members),
size: members.len(),
members: listed,
}
}
#[must_use]
pub fn diff_communities(
from_edges: &BTreeMap<String, DiffEdge>,
to_edges: &BTreeMap<String, DiffEdge>,
) -> CommunityDelta {
let from = louvain_from_edges(from_edges);
let to = louvain_from_edges(to_edges);
let mut candidates: Vec<(usize, usize, f64)> = Vec::new();
for (i, a) in from.iter().enumerate() {
for (j, b) in to.iter().enumerate() {
let overlap = jaccard(a, b);
if overlap >= COMMUNITY_MATCH_JACCARD_THRESHOLD {
candidates.push((i, j, overlap));
}
}
}
candidates.sort_by(|x, y| {
y.2.partial_cmp(&x.2)
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| fingerprint(&from[x.0]).cmp(&fingerprint(&from[y.0])))
.then_with(|| fingerprint(&to[x.1]).cmp(&fingerprint(&to[y.1])))
});
let mut from_used = vec![false; from.len()];
let mut to_used = vec![false; to.len()];
let mut matched = Vec::new();
let mut unchanged = 0usize;
for (i, j, overlap) in candidates {
if from_used[i] || to_used[j] {
continue;
}
from_used[i] = true;
to_used[j] = true;
let joined: Vec<String> = to[j].difference(&from[i]).cloned().collect();
let left: Vec<String> = from[i].difference(&to[j]).cloned().collect();
if joined.is_empty() && left.is_empty() {
unchanged += 1;
continue;
}
let mut joined = joined;
let mut left = left;
joined.truncate(GRAPH_DIFF_DETAIL_CAP);
left.truncate(GRAPH_DIFF_DETAIL_CAP);
matched.push(CommunityMatch {
from_fingerprint: fingerprint(&from[i]),
to_fingerprint: fingerprint(&to[j]),
jaccard: overlap,
joined,
left,
});
}
let mut deaths: Vec<CommunityEndpoint> = from
.iter()
.enumerate()
.filter(|(i, _)| !from_used[*i])
.map(|(_, members)| endpoint(members))
.collect();
let mut births: Vec<CommunityEndpoint> = to
.iter()
.enumerate()
.filter(|(j, _)| !to_used[*j])
.map(|(_, members)| endpoint(members))
.collect();
deaths.sort_by(|a, b| a.fingerprint.cmp(&b.fingerprint));
births.sort_by(|a, b| a.fingerprint.cmp(&b.fingerprint));
deaths.truncate(GRAPH_DIFF_DETAIL_CAP);
births.truncate(GRAPH_DIFF_DETAIL_CAP);
matched.sort_by(|a, b| a.from_fingerprint.cmp(&b.from_fingerprint));
CommunityDelta {
matched,
births,
deaths,
unchanged,
}
}
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CentralityMover {
pub memory_id: String,
pub from_pagerank: f64,
pub to_pagerank: f64,
pub delta: f64,
}
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DiffTruncation {
pub nodes_added: bool,
pub nodes_removed: bool,
pub edges_added: bool,
pub edges_removed: bool,
}
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SnapshotRef {
pub snapshot_id: String,
pub snapshot_version: u32,
pub created_at: String,
pub node_count: usize,
pub edge_count: usize,
}
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DiffSummary {
pub nodes_added: usize,
pub nodes_removed: usize,
pub edges_added: usize,
pub edges_removed: usize,
pub communities_matched: usize,
pub community_births: usize,
pub community_deaths: usize,
pub centrality_movers: usize,
pub centrality_omitted: usize,
}
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GraphDiffReport {
pub schema: &'static str,
pub graph_type: String,
pub from: SnapshotRef,
pub to: SnapshotRef,
pub summary: DiffSummary,
pub nodes_added: Vec<String>,
pub nodes_removed: Vec<String>,
pub edges_added: Vec<DiffEdge>,
pub edges_removed: Vec<DiffEdge>,
pub communities: CommunityDelta,
pub movers: Vec<CentralityMover>,
pub detail_cap: usize,
pub truncated: DiffTruncation,
}
fn snapshot_ref(view: &SnapshotGraphView) -> SnapshotRef {
SnapshotRef {
snapshot_id: view.snapshot_id.clone(),
snapshot_version: view.snapshot_version,
created_at: view.created_at.clone(),
node_count: view.nodes.len(),
edge_count: view.edges.len(),
}
}
#[must_use]
pub fn diff_snapshots(
graph_type: &str,
from: &SnapshotGraphView,
to: &SnapshotGraphView,
) -> GraphDiffReport {
let nodes_added_full: Vec<String> = to
.nodes
.keys()
.filter(|id| !from.nodes.contains_key(*id))
.cloned()
.collect();
let nodes_removed_full: Vec<String> = from
.nodes
.keys()
.filter(|id| !to.nodes.contains_key(*id))
.cloned()
.collect();
let edges_added_full: Vec<DiffEdge> = to
.edges
.iter()
.filter(|(key, _)| !from.edges.contains_key(*key))
.map(|(_, edge)| edge.clone())
.collect();
let edges_removed_full: Vec<DiffEdge> = from
.edges
.iter()
.filter(|(key, _)| !to.edges.contains_key(*key))
.map(|(_, edge)| edge.clone())
.collect();
let mut movers: Vec<CentralityMover> = Vec::new();
let mut centrality_omitted = 0usize;
for (id, from_rank) in &from.nodes {
let Some(to_rank) = to.nodes.get(id) else {
continue;
};
match (from_rank, to_rank) {
(Some(from_rank), Some(to_rank)) => {
let delta = to_rank - from_rank;
if delta != 0.0 {
movers.push(CentralityMover {
memory_id: id.clone(),
from_pagerank: *from_rank,
to_pagerank: *to_rank,
delta,
});
}
}
_ => centrality_omitted += 1,
}
}
movers.sort_by(|a, b| {
b.delta
.abs()
.partial_cmp(&a.delta.abs())
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| a.memory_id.cmp(&b.memory_id))
});
movers.truncate(GRAPH_DIFF_MOVERS_CAP);
let communities = diff_communities(&from.edges, &to.edges);
let summary = DiffSummary {
nodes_added: nodes_added_full.len(),
nodes_removed: nodes_removed_full.len(),
edges_added: edges_added_full.len(),
edges_removed: edges_removed_full.len(),
communities_matched: communities.matched.len() + communities.unchanged,
community_births: communities.births.len(),
community_deaths: communities.deaths.len(),
centrality_movers: movers.len(),
centrality_omitted,
};
let truncated = DiffTruncation {
nodes_added: nodes_added_full.len() > GRAPH_DIFF_DETAIL_CAP,
nodes_removed: nodes_removed_full.len() > GRAPH_DIFF_DETAIL_CAP,
edges_added: edges_added_full.len() > GRAPH_DIFF_DETAIL_CAP,
edges_removed: edges_removed_full.len() > GRAPH_DIFF_DETAIL_CAP,
};
let mut nodes_added = nodes_added_full;
let mut nodes_removed = nodes_removed_full;
let mut edges_added = edges_added_full;
let mut edges_removed = edges_removed_full;
nodes_added.truncate(GRAPH_DIFF_DETAIL_CAP);
nodes_removed.truncate(GRAPH_DIFF_DETAIL_CAP);
edges_added.truncate(GRAPH_DIFF_DETAIL_CAP);
edges_removed.truncate(GRAPH_DIFF_DETAIL_CAP);
GraphDiffReport {
schema: GRAPH_DIFF_SCHEMA_V1,
graph_type: graph_type.to_owned(),
from: snapshot_ref(from),
to: snapshot_ref(to),
summary,
nodes_added,
nodes_removed,
edges_added,
edges_removed,
communities,
movers,
detail_cap: GRAPH_DIFF_DETAIL_CAP,
truncated,
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used)]
use super::{
COMMUNITY_MATCH_JACCARD_THRESHOLD, DiffEdge, GRAPH_DIFF_DETAIL_CAP, SnapshotGraphView,
diff_communities, diff_snapshots, edge_key, parse_snapshot_graph,
};
use std::collections::BTreeMap;
fn view(
id: &str,
version: u32,
nodes: &[(&str, Option<f64>)],
edges: &[(&str, &str)],
) -> SnapshotGraphView {
let mut node_map = BTreeMap::new();
for (node, rank) in nodes {
node_map.insert((*node).to_owned(), *rank);
}
let mut edge_map = BTreeMap::new();
for (source, target) in edges {
let key = edge_key(source, target, "related", false);
edge_map.insert(
key.clone(),
DiffEdge {
key,
source: (*source).to_owned(),
target: (*target).to_owned(),
relation: "related".to_owned(),
directed: false,
},
);
}
SnapshotGraphView {
snapshot_id: id.to_owned(),
snapshot_version: version,
created_at: format!("2026-08-0{version}T00:00:00Z"),
nodes: node_map,
edges: edge_map,
}
}
#[test]
fn edge_key_is_orientation_canonical_for_undirected_edges() {
assert_eq!(
edge_key("mem_b", "mem_a", "related", false),
edge_key("mem_a", "mem_b", "related", false)
);
assert_ne!(
edge_key("mem_b", "mem_a", "related", true),
edge_key("mem_a", "mem_b", "related", true)
);
assert_ne!(
edge_key("mem_a", "mem_b", "related", false),
edge_key("mem_a", "mem_b", "contradicts", false)
);
}
#[test]
fn add_remove_sets_are_exact_complements() {
let a = view(
"gsnap_a",
1,
&[("m1", Some(0.4)), ("m2", Some(0.3)), ("m3", Some(0.3))],
&[("m1", "m2"), ("m2", "m3")],
);
let b = view(
"gsnap_b",
2,
&[("m1", Some(0.5)), ("m2", Some(0.2)), ("m4", Some(0.3))],
&[("m1", "m2"), ("m1", "m4")],
);
let ab = diff_snapshots("memory_links", &a, &b);
let ba = diff_snapshots("memory_links", &b, &a);
assert_eq!(ab.nodes_added, vec!["m4".to_owned()]);
assert_eq!(ab.nodes_removed, vec!["m3".to_owned()]);
assert_eq!(ab.nodes_added, ba.nodes_removed);
assert_eq!(ab.nodes_removed, ba.nodes_added);
assert_eq!(ab.edges_added, ba.edges_removed);
assert_eq!(ab.edges_removed, ba.edges_added);
assert_eq!(ab.summary.nodes_added, 1);
assert_eq!(ab.summary.edges_added, 1);
assert_eq!(ab.summary.edges_removed, 1);
}
#[test]
fn identical_structure_yields_empty_community_delta() {
let edges_one = view(
"gsnap_a",
1,
&[],
&[
("m1", "m2"),
("m2", "m3"),
("m1", "m3"),
("m7", "m8"),
("m8", "m9"),
("m7", "m9"),
],
)
.edges;
let edges_two = view(
"gsnap_b",
2,
&[],
&[
("m8", "m9"),
("m7", "m9"),
("m7", "m8"),
("m1", "m3"),
("m2", "m3"),
("m1", "m2"),
],
)
.edges;
let delta = diff_communities(&edges_one, &edges_two);
assert!(delta.matched.is_empty(), "no churn to report: {delta:?}");
assert!(delta.births.is_empty(), "no births: {delta:?}");
assert!(delta.deaths.is_empty(), "no deaths: {delta:?}");
assert!(delta.unchanged >= 1, "communities matched silently");
}
#[test]
fn community_birth_death_and_membership_churn_are_reported() {
let from = view(
"gsnap_a",
1,
&[],
&[("m1", "m2"), ("m2", "m3"), ("m1", "m3")],
);
let to = view(
"gsnap_b",
2,
&[],
&[
("m1", "m2"),
("m2", "m3"),
("m1", "m3"),
("m3", "m4"),
("m8", "m9"),
],
);
let delta = diff_communities(&from.edges, &to.edges);
assert!(
delta
.matched
.iter()
.any(|matched| matched.joined.contains(&"m4".to_owned())),
"m4 joins the surviving community: {delta:?}"
);
assert!(
delta
.births
.iter()
.any(|birth| birth.members.contains(&"m8".to_owned())),
"the m8-m9 community is a birth: {delta:?}"
);
assert!(delta.deaths.is_empty(), "nothing died: {delta:?}");
}
#[test]
fn movers_rank_by_abs_delta_with_id_tie_break_and_omit_missing_sides() {
let from = view(
"gsnap_a",
1,
&[
("m1", Some(0.10)),
("m2", Some(0.75)),
("m3", Some(0.25)),
("m4", None),
("m5", Some(0.20)),
],
&[],
);
let to = view(
"gsnap_b",
2,
&[
("m1", Some(0.40)), ("m2", Some(0.50)), ("m3", Some(0.50)), ("m4", Some(0.50)), ("m5", Some(0.20)), ],
&[],
);
let report = diff_snapshots("memory_links", &from, &to);
let ids: Vec<&str> = report
.movers
.iter()
.map(|mover| mover.memory_id.as_str())
.collect();
assert_eq!(ids, vec!["m1", "m2", "m3"], "rank + tie-break drifted");
assert_eq!(report.summary.centrality_omitted, 1);
assert!((report.movers[0].delta - 0.30).abs() < 1e-9);
}
#[test]
fn detail_arrays_truncate_at_the_declared_governor_cap() {
let mut nodes: Vec<(String, Option<f64>)> = Vec::new();
for index in 0..(GRAPH_DIFF_DETAIL_CAP + 8) {
nodes.push((format!("mem_{index:04}"), None));
}
let node_refs: Vec<(&str, Option<f64>)> = nodes
.iter()
.map(|(id, rank)| (id.as_str(), *rank))
.collect();
let from = view("gsnap_a", 1, &[], &[]);
let to = view("gsnap_b", 2, &node_refs, &[]);
let report = diff_snapshots("memory_links", &from, &to);
assert_eq!(report.summary.nodes_added, GRAPH_DIFF_DETAIL_CAP + 8);
assert_eq!(report.nodes_added.len(), GRAPH_DIFF_DETAIL_CAP);
assert!(report.truncated.nodes_added);
assert_eq!(report.detail_cap, GRAPH_DIFF_DETAIL_CAP);
}
#[test]
fn parse_snapshot_graph_reads_persisted_metrics_shape() {
let metrics = serde_json::json!({
"schema": "ee.graph.snapshot.metrics.v1",
"nodes": [
{"id": "m1", "pagerank": 0.5},
{"id": "m2", "pagerank": null},
],
"edges": [
{"source": "m1", "target": "m2", "relation": "supports", "directed": true},
],
})
.to_string();
let parsed = parse_snapshot_graph("gsnap_x", 3, "2026-08-09T00:00:00Z", &metrics).unwrap();
assert_eq!(parsed.nodes.len(), 2);
assert_eq!(parsed.nodes["m1"], Some(0.5));
assert_eq!(parsed.nodes["m2"], None);
assert_eq!(parsed.edges.len(), 1);
let edge = parsed.edges.values().next().unwrap();
assert_eq!(edge.relation, "supports");
assert!(edge.directed);
assert!(parse_snapshot_graph("gsnap_y", 4, "t", "not json").is_err());
let _ = COMMUNITY_MATCH_JACCARD_THRESHOLD;
}
}