feat: epidemic gossip implementation

Cargo.lock

@@ -534,6 +534,14 @@ dependencies = [
  "getrandom",
 ]
 
+[[package]]
+name = "swactor-gossip"
+version = "0.1.0"
+dependencies = [
+ "getrandom",
+ "swactor",
+]
+
 [[package]]
 name = "swactor-python"
 version = "0.1.0"

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [".", "crates/swactor-python", "crates/swactor-wasm"]
+members = [".", "crates/swactor-python", "crates/swactor-wasm", "crates/swactor-gossip"]
 exclude = ["tools/depgraph"]
 
 [package]

crates/swactor-gossip/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "swactor-gossip"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+swactor = { path = "../.." }
+getrandom = "0.2"

crates/swactor-gossip/examples/gossip_sim.rs

@@ -0,0 +1,73 @@
+use std::fs;
+
+use swactor_gossip::report::generate_html_report;
+use swactor_gossip::sim::{run_simulation, SimConfig, Topology};
+
+fn main() {
+    let scenarios = vec![
+        SimConfig {
+            name: "Ring (5 nodes)".into(),
+            topology: Topology::Ring,
+            num_nodes: 5,
+            initial_data: test_data(3),
+            num_rounds: 15,
+            ticks_per_round: 4,
+            heal_after_round: None,
+        },
+        SimConfig {
+            name: "Star (7 nodes)".into(),
+            topology: Topology::Star,
+            num_nodes: 7,
+            initial_data: test_data(3),
+            num_rounds: 10,
+            ticks_per_round: 4,
+            heal_after_round: None,
+        },
+        SimConfig {
+            name: "Full Mesh (5 nodes)".into(),
+            topology: Topology::FullMesh,
+            num_nodes: 5,
+            initial_data: test_data(3),
+            num_rounds: 8,
+            ticks_per_round: 4,
+            heal_after_round: None,
+        },
+        SimConfig {
+            name: "Chain (8 nodes)".into(),
+            topology: Topology::Chain,
+            num_nodes: 8,
+            initial_data: test_data(3),
+            num_rounds: 20,
+            ticks_per_round: 4,
+            heal_after_round: None,
+        },
+        SimConfig {
+            name: "Partition & Heal (6 nodes)".into(),
+            topology: Topology::Partitioned,
+            num_nodes: 6,
+            initial_data: test_data(3),
+            num_rounds: 20,
+            ticks_per_round: 4,
+            heal_after_round: Some(10),
+        },
+    ];
+
+    for config in scenarios {
+        let filename = format!(
+            "gossip_report_{}.html",
+            config.name.to_lowercase().replace(' ', "_").replace(['(', ')'], "")
+        );
+        println!("Running scenario: {} ...", config.name);
+        let trace = run_simulation(config);
+        let html = generate_html_report(&trace);
+        fs::write(&filename, &html).expect("failed to write report");
+        println!("  -> wrote {filename} ({} bytes)", html.len());
+    }
+    println!("Done.");
+}
+
+fn test_data(n: usize) -> Vec<(String, Vec<u8>)> {
+    (0..n)
+        .map(|i| (format!("key-{i}"), format!("value-{i}").into_bytes()))
+        .collect()
+}

crates/swactor-gossip/src/actor.rs

@@ -0,0 +1,179 @@
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+
+use crate::message::{GossipMessage, GossipQueryResponse};
+use crate::state::GossipState;
+use crate::trace::{
+    current_tick, record_event, resolve_name, EventLog, GossipEvent, GossipEventKind,
+    NameRegistry, NodeSnapshot, TickCounter,
+};
+
+pub struct GossipActor {
+    state: GossipState,
+    peers: Vec<ActorAddress>,
+    event_log: Option<EventLog>,
+    tick_counter: Option<TickCounter>,
+    name_registry: Option<NameRegistry>,
+}
+
+impl GossipActor {
+    pub fn new() -> Self {
+        Self {
+            state: GossipState::new(),
+            peers: Vec::new(),
+            event_log: None,
+            tick_counter: None,
+            name_registry: None,
+        }
+    }
+
+    /// Create a traced actor that records events into the shared log.
+    pub fn traced(log: EventLog, tick: TickCounter, names: NameRegistry) -> Self {
+        Self {
+            state: GossipState::new(),
+            peers: Vec::new(),
+            event_log: Some(log),
+            tick_counter: Some(tick),
+            name_registry: Some(names),
+        }
+    }
+
+    fn pick_random_peer(&self) -> Option<ActorAddress> {
+        if self.peers.is_empty() {
+            return None;
+        }
+        let mut buf = [0u8; 8];
+        getrandom::getrandom(&mut buf).unwrap();
+        let idx = usize::from_ne_bytes(buf) % self.peers.len();
+        Some(self.peers[idx])
+    }
+
+    fn record(&self, addr: ActorAddress, kind: GossipEventKind) {
+        if let (Some(log), Some(tick), Some(names)) =
+            (&self.event_log, &self.tick_counter, &self.name_registry)
+        {
+            let event = GossipEvent {
+                tick: current_tick(tick),
+                node_name: resolve_name(names, addr),
+                node_addr: addr,
+                kind,
+            };
+            record_event(log, event);
+        }
+    }
+}
+
+impl Default for GossipActor {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ActorInterface for GossipActor {
+    type Incoming = GossipMessage;
+    type Response = ();
+
+    fn handle(&mut self, ctx: &Ctx, msg: GossipMessage) {
+        let self_addr = ctx.self_addr();
+        match msg {
+            GossipMessage::AddPeer(addr) => {
+                if !self.peers.contains(&addr) {
+                    self.peers.push(addr);
+                    self.record(
+                        self_addr,
+                        GossipEventKind::PeerAdded {
+                            peer_name: self
+                                .name_registry
+                                .as_ref()
+                                .map(|r| resolve_name(r, addr))
+                                .unwrap_or_default(),
+                        },
+                    );
+                }
+            }
+            GossipMessage::RemovePeer(addr) => {
+                let before = self.peers.len();
+                self.peers.retain(|a| *a != addr);
+                if self.peers.len() < before {
+                    self.record(
+                        self_addr,
+                        GossipEventKind::PeerRemoved {
+                            peer_name: self
+                                .name_registry
+                                .as_ref()
+                                .map(|r| resolve_name(r, addr))
+                                .unwrap_or_default(),
+                        },
+                    );
+                }
+            }
+            GossipMessage::Set { key, value } => {
+                self.state.set(key.clone(), value);
+                self.record(self_addr, GossipEventKind::LocalSet { key });
+            }
+            GossipMessage::DoGossipRound => {
+                if let Some(peer) = self.pick_random_peer() {
+                    self.record(
+                        self_addr,
+                        GossipEventKind::GossipRoundStarted {
+                            target_name: self
+                                .name_registry
+                                .as_ref()
+                                .map(|r| resolve_name(r, peer))
+                                .unwrap_or_default(),
+                        },
+                    );
+                    let _ = ctx.send(
+                        peer,
+                        GossipMessage::Push {
+                            from: self_addr,
+                            state: self.state.clone(),
+                        },
+                    );
+                } else {
+                    self.record(self_addr, GossipEventKind::GossipRoundNoPeers);
+                }
+            }
+            GossipMessage::Push {
+                from,
+                state: remote,
+            } => {
+                let keys_updated = self.state.merge(&remote);
+                self.record(
+                    self_addr,
+                    GossipEventKind::PushReceived {
+                        from_name: self
+                            .name_registry
+                            .as_ref()
+                            .map(|r| resolve_name(r, from))
+                            .unwrap_or_default(),
+                        keys_updated,
+                    },
+                );
+            }
+            GossipMessage::TakeSnapshot => {
+                self.record(
+                    self_addr,
+                    GossipEventKind::StateSnapshot {
+                        snapshot: NodeSnapshot {
+                            entries: self.state.entries().clone(),
+                            peer_count: self.peers.len(),
+                        },
+                    },
+                );
+            }
+            GossipMessage::Query { key, reply_to } => {
+                self.record(
+                    self_addr,
+                    GossipEventKind::QueryReceived { key: key.clone() },
+                );
+                let entry = self.state.get(&key);
+                let resp = GossipQueryResponse {
+                    key,
+                    value: entry.map(|e| e.value.clone()),
+                    version: entry.map(|e| e.version),
+                };
+                let _ = ctx.send(reply_to, resp);
+            }
+        }
+    }
+}

crates/swactor-gossip/src/lib.rs

@@ -0,0 +1,11 @@
+pub mod actor;
+pub mod message;
+pub mod state;
+pub mod trace;
+
+pub mod report;
+pub mod sim;
+
+pub use actor::GossipActor;
+pub use message::{GossipMessage, GossipQueryResponse};
+pub use state::{GossipState, VersionedValue};

crates/swactor-gossip/src/message.rs

@@ -0,0 +1,34 @@
+use swactor::actor::ActorAddress;
+
+use crate::state::GossipState;
+
+#[derive(Debug, Clone)]
+pub enum GossipMessage {
+    /// Register a peer to gossip with.
+    AddPeer(ActorAddress),
+    /// Remove a peer from the gossip set.
+    RemovePeer(ActorAddress),
+    /// Set a key-value pair in this node's local state.
+    Set { key: String, value: Vec<u8> },
+    /// Trigger a gossip round: pick a random peer and push our full state.
+    DoGossipRound,
+    /// Incoming state push from a peer.
+    Push {
+        from: ActorAddress,
+        state: GossipState,
+    },
+    /// Query the current value for a key; response sent to `reply_to`.
+    Query {
+        key: String,
+        reply_to: ActorAddress,
+    },
+    /// Ask the actor to dump its current state into the event log (tracing only).
+    TakeSnapshot,
+}
+
+#[derive(Debug, Clone)]
+pub struct GossipQueryResponse {
+    pub key: String,
+    pub value: Option<Vec<u8>>,
+    pub version: Option<u64>,
+}

crates/swactor-gossip/src/report.rs

@@ -0,0 +1,550 @@
+use std::collections::HashMap;
+use std::f64::consts::PI;
+
+use crate::trace::{GossipEventKind, SimulationTrace};
+
+/// Generate a self-contained HTML report from a simulation trace.
+pub fn generate_html_report(trace: &SimulationTrace) -> String {
+    let mut html = String::with_capacity(32_000);
+
+    html.push_str("<!DOCTYPE html>\n<html lang=\"en\">\n<head>\n<meta charset=\"utf-8\">\n");
+    html.push_str(&format!(
+        "<title>Gossip Simulation: {}</title>\n",
+        escape_html(&trace.name)
+    ));
+    html.push_str("<style>\n");
+    html.push_str(CSS);
+    html.push_str("</style>\n</head>\n<body>\n");
+
+    html.push_str(&format!(
+        "<h1>Gossip Simulation: {}</h1>\n",
+        escape_html(&trace.name)
+    ));
+
+    // Summary metrics
+    render_summary(&mut html, trace);
+
+    // Network topology
+    render_topology_svg(&mut html, trace);
+
+    // Propagation heatmap
+    render_heatmap_svg(&mut html, trace);
+
+    // Convergence curve
+    render_convergence_svg(&mut html, trace);
+
+    // Message flow timeline
+    render_message_flow_svg(&mut html, trace);
+
+    // Event log table
+    render_event_table(&mut html, trace);
+
+    html.push_str("</body>\n</html>\n");
+    html
+}
+
+// ── CSS ──────────────────────────────────────────────────────────────────
+
+const CSS: &str = r#"
+body {
+    font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
+    max-width: 1200px; margin: 0 auto; padding: 20px;
+    background: #fafafa; color: #222;
+}
+h1 { border-bottom: 3px solid #333; padding-bottom: 8px; }
+h2 { margin-top: 32px; color: #444; }
+.metrics { display: flex; flex-wrap: wrap; gap: 16px; margin: 16px 0; }
+.metric {
+    background: #fff; border: 1px solid #ddd; border-radius: 8px;
+    padding: 12px 20px; min-width: 140px;
+}
+.metric .label { font-size: 0.85em; color: #666; }
+.metric .value { font-size: 1.5em; font-weight: bold; }
+svg { display: block; margin: 12px 0; }
+table { border-collapse: collapse; width: 100%; margin: 12px 0; }
+th, td { border: 1px solid #ddd; padding: 6px 10px; text-align: left; font-size: 0.85em; }
+th { background: #f0f0f0; }
+tr:nth-child(even) { background: #fafafa; }
+.capped { color: #999; font-style: italic; margin: 4px 0; }
+"#;
+
+// ── Summary metrics ──────────────────────────────────────────────────────
+
+fn render_summary(html: &mut String, trace: &SimulationTrace) {
+    let num_nodes = trace.node_names.len();
+    let total_pushes = trace
+        .events
+        .iter()
+        .filter(|e| matches!(e.kind, GossipEventKind::GossipRoundStarted { .. }))
+        .count();
+    let redundant_pushes = trace
+        .events
+        .iter()
+        .filter(|e| matches!(e.kind, GossipEventKind::PushReceived { keys_updated: 0, .. }))
+        .count();
+    let convergence_round = find_convergence_round(trace);
+
+    html.push_str("<h2>Summary</h2>\n<div class=\"metrics\">\n");
+    metric(html, "Nodes", &num_nodes.to_string());
+    metric(html, "Keys", &trace.total_keys.to_string());
+    metric(html, "Rounds", &trace.num_rounds.to_string());
+    metric(html, "Pushes", &total_pushes.to_string());
+    metric(html, "Redundant", &redundant_pushes.to_string());
+    metric(
+        html,
+        "Converged at",
+        &convergence_round
+            .map(|r| format!("round {r}"))
+            .unwrap_or_else(|| "never".into()),
+    );
+    if total_pushes > 0 {
+        let efficiency = 100.0 * (1.0 - redundant_pushes as f64 / total_pushes as f64);
+        metric(html, "Efficiency", &format!("{efficiency:.0}%"));
+    }
+    html.push_str("</div>\n");
+}
+
+fn metric(html: &mut String, label: &str, value: &str) {
+    html.push_str(&format!(
+        "<div class=\"metric\"><div class=\"label\">{label}</div><div class=\"value\">{value}</div></div>\n"
+    ));
+}
+
+fn find_convergence_round(trace: &SimulationTrace) -> Option<usize> {
+    if trace.total_keys == 0 {
+        return Some(0);
+    }
+    for (round_idx, snapshots) in trace.snapshots_per_round.iter().enumerate() {
+        let all_converged = snapshots
+            .iter()
+            .all(|(_, snap)| snap.entries.len() >= trace.total_keys);
+        if all_converged {
+            return Some(round_idx + 1);
+        }
+    }
+    None
+}
+
+// ── Topology SVG ─────────────────────────────────────────────────────────
+
+fn render_topology_svg(html: &mut String, trace: &SimulationTrace) {
+    html.push_str("<h2>Network Topology</h2>\n");
+
+    let n = trace.node_names.len();
+    let size = 400.0_f64;
+    let cx = size / 2.0;
+    let cy = size / 2.0;
+    let radius = size / 2.0 - 50.0;
+
+    // Compute node positions in circular layout.
+    let positions: Vec<(f64, f64)> = (0..n)
+        .map(|i| {
+            let angle = 2.0 * PI * (i as f64) / (n as f64) - PI / 2.0;
+            (cx + radius * angle.cos(), cy + radius * angle.sin())
+        })
+        .collect();
+
+    let name_to_idx: HashMap<&str, usize> = trace
+        .node_names
+        .iter()
+        .enumerate()
+        .map(|(i, n)| (n.as_str(), i))
+        .collect();
+
+    html.push_str(&format!(
+        "<svg width=\"{size}\" height=\"{size}\" viewBox=\"0 0 {size} {size}\">\n"
+    ));
+    html.push_str("<defs><marker id=\"arrow\" markerWidth=\"8\" markerHeight=\"6\" refX=\"8\" refY=\"3\" orient=\"auto\"><path d=\"M0,0 L8,3 L0,6\" fill=\"#888\"/></marker></defs>\n");
+
+    // Draw edges.
+    for (from_name, to_name) in &trace.topology_edges {
+        if let (Some(&fi), Some(&ti)) = (name_to_idx.get(from_name.as_str()), name_to_idx.get(to_name.as_str())) {
+            let (x1, y1) = positions[fi];
+            let (x2, y2) = positions[ti];
+            // Shorten line to not overlap circle.
+            let dx = x2 - x1;
+            let dy = y2 - y1;
+            let len = (dx * dx + dy * dy).sqrt();
+            if len > 0.0 {
+                let nx = dx / len;
+                let ny = dy / len;
+                let sx = x1 + nx * 18.0;
+                let sy = y1 + ny * 18.0;
+                let ex = x2 - nx * 18.0;
+                let ey = y2 - ny * 18.0;
+                html.push_str(&format!(
+                    "<line x1=\"{sx:.1}\" y1=\"{sy:.1}\" x2=\"{ex:.1}\" y2=\"{ey:.1}\" stroke=\"#aaa\" stroke-width=\"1\" marker-end=\"url(#arrow)\"/>\n"
+                ));
+            }
+        }
+    }
+
+    // Draw nodes.
+    for (i, name) in trace.node_names.iter().enumerate() {
+        let (x, y) = positions[i];
+        html.push_str(&format!(
+            "<circle cx=\"{x:.1}\" cy=\"{y:.1}\" r=\"16\" fill=\"#4a90d9\" stroke=\"#2a5a9d\" stroke-width=\"2\"/>\n"
+        ));
+        html.push_str(&format!(
+            "<text x=\"{x:.1}\" y=\"{ty:.1}\" text-anchor=\"middle\" fill=\"#fff\" font-size=\"10\" font-weight=\"bold\">{name}</text>\n",
+            ty = y + 4.0,
+        ));
+    }
+
+    html.push_str("</svg>\n");
+}
+
+// ── Propagation heatmap ──────────────────────────────────────────────────
+
+fn render_heatmap_svg(html: &mut String, trace: &SimulationTrace) {
+    html.push_str("<h2>Propagation Heatmap</h2>\n");
+    html.push_str("<p>Rows = nodes, columns = rounds. Color intensity = fraction of total keys held.</p>\n");
+
+    let n = trace.node_names.len();
+    let rounds = trace.snapshots_per_round.len();
+    if rounds == 0 || n == 0 {
+        html.push_str("<p>No data.</p>\n");
+        return;
+    }
+
+    let cell_w = 36.0_f64;
+    let cell_h = 28.0_f64;
+    let label_w = 80.0_f64;
+    let header_h = 28.0_f64;
+    let w = label_w + cell_w * rounds as f64 + 10.0;
+    let h = header_h + cell_h * n as f64 + 10.0;
+
+    html.push_str(&format!(
+        "<svg width=\"{w:.0}\" height=\"{h:.0}\" viewBox=\"0 0 {w:.0} {h:.0}\">\n"
+    ));
+
+    // Column headers.
+    for r in 0..rounds {
+        let x = label_w + r as f64 * cell_w + cell_w / 2.0;
+        let ty = header_h - 6.0;
+        let label = r + 1;
+        html.push_str(&format!(
+            "<text x=\"{x:.1}\" y=\"{ty}\" text-anchor=\"middle\" font-size=\"10\" fill=\"#666\">R{label}</text>\n"
+        ));
+    }
+
+    // Build a name→row index for stable ordering.
+    let name_to_row: HashMap<&str, usize> = trace
+        .node_names
+        .iter()
+        .enumerate()
+        .map(|(i, n)| (n.as_str(), i))
+        .collect();
+
+    for (r, round_snaps) in trace.snapshots_per_round.iter().enumerate() {
+        for (name, snap) in round_snaps {
+            if let Some(&row) = name_to_row.get(name.as_str()) {
+                let frac = if trace.total_keys > 0 {
+                    snap.entries.len() as f64 / trace.total_keys as f64
+                } else {
+                    0.0
+                };
+                let x = label_w + r as f64 * cell_w;
+                let y = header_h + row as f64 * cell_h;
+                let color = heatmap_color(frac);
+                html.push_str(&format!(
+                    "<rect x=\"{x:.1}\" y=\"{y:.1}\" width=\"{cell_w}\" height=\"{cell_h}\" fill=\"{color}\" stroke=\"#fff\" stroke-width=\"1\"/>\n"
+                ));
+                // Show count inside cell.
+                let text_color = if frac > 0.5 { "#fff" } else { "#333" };
+                let tx = x + cell_w / 2.0;
+                let ty = y + cell_h / 2.0 + 3.0;
+                let count = snap.entries.len();
+                html.push_str(&format!(
+                    "<text x=\"{tx:.1}\" y=\"{ty:.1}\" text-anchor=\"middle\" font-size=\"10\" fill=\"{text_color}\">{count}</text>\n"
+                ));
+            }
+        }
+    }
+
+    // Row labels.
+    for (i, name) in trace.node_names.iter().enumerate() {
+        let y = header_h + i as f64 * cell_h + cell_h / 2.0 + 4.0;
+        html.push_str(&format!(
+            "<text x=\"{x}\" y=\"{y:.1}\" font-size=\"11\" fill=\"#333\">{name}</text>\n",
+            x = 4.0,
+        ));
+    }
+
+    html.push_str("</svg>\n");
+}
+
+fn heatmap_color(frac: f64) -> String {
+    // Interpolate from light (#e8f4e8) to deep green (#1a7a1a).
+    let f = frac.clamp(0.0, 1.0);
+    let r = (232.0 + f * (26.0 - 232.0)) as u8;
+    let g = (244.0 + f * (122.0 - 244.0)) as u8;
+    let b = (232.0 + f * (26.0 - 232.0)) as u8;
+    format!("#{r:02x}{g:02x}{b:02x}")
+}
+
+// ── Convergence curve ────────────────────────────────────────────────────
+
+fn render_convergence_svg(html: &mut String, trace: &SimulationTrace) {
+    html.push_str("<h2>Convergence Curve</h2>\n");
+    html.push_str("<p>Percentage of nodes that hold all keys vs. round number.</p>\n");
+
+    let rounds = trace.snapshots_per_round.len();
+    if rounds == 0 {
+        html.push_str("<p>No data.</p>\n");
+        return;
+    }
+
+    let chart_w = 600.0_f64;
+    let chart_h = 300.0_f64;
+    let margin_l = 50.0_f64;
+    let margin_b = 40.0_f64;
+    let margin_t = 20.0_f64;
+    let margin_r = 20.0_f64;
+    let w = chart_w + margin_l + margin_r;
+    let h = chart_h + margin_t + margin_b;
+
+    html.push_str(&format!(
+        "<svg width=\"{w:.0}\" height=\"{h:.0}\" viewBox=\"0 0 {w:.0} {h:.0}\">\n"
+    ));
+
+    // Axes.
+    html.push_str(&format!(
+        "<line x1=\"{margin_l}\" y1=\"{margin_t}\" x2=\"{margin_l}\" y2=\"{}\" stroke=\"#333\" stroke-width=\"1\"/>\n",
+        margin_t + chart_h,
+    ));
+    html.push_str(&format!(
+        "<line x1=\"{margin_l}\" y1=\"{}\" x2=\"{}\" y2=\"{}\" stroke=\"#333\" stroke-width=\"1\"/>\n",
+        margin_t + chart_h,
+        margin_l + chart_w,
+        margin_t + chart_h,
+    ));
+
+    // Y-axis labels.
+    for pct in [0, 25, 50, 75, 100] {
+        let y = margin_t + chart_h - (pct as f64 / 100.0) * chart_h;
+        html.push_str(&format!(
+            "<text x=\"{}\" y=\"{:.1}\" text-anchor=\"end\" font-size=\"10\" fill=\"#666\">{pct}%</text>\n",
+            margin_l - 6.0, y + 3.0,
+        ));
+        html.push_str(&format!(
+            "<line x1=\"{margin_l}\" y1=\"{y:.1}\" x2=\"{}\" y2=\"{y:.1}\" stroke=\"#eee\" stroke-width=\"1\"/>\n",
+            margin_l + chart_w,
+        ));
+    }
+
+    // X-axis labels.
+    let step = (rounds / 10).max(1);
+    for r in (0..rounds).step_by(step) {
+        let x = margin_l + (r as f64 + 0.5) / rounds as f64 * chart_w;
+        html.push_str(&format!(
+            "<text x=\"{x:.1}\" y=\"{}\" text-anchor=\"middle\" font-size=\"10\" fill=\"#666\">{}</text>\n",
+            margin_t + chart_h + 16.0, r + 1,
+        ));
+    }
+    // X-axis title.
+    html.push_str(&format!(
+        "<text x=\"{}\" y=\"{}\" text-anchor=\"middle\" font-size=\"11\" fill=\"#444\">Round</text>\n",
+        margin_l + chart_w / 2.0,
+        margin_t + chart_h + 34.0,
+    ));
+
+    // Compute data points.
+    let n = trace.node_names.len();
+    let mut points = Vec::with_capacity(rounds);
+    for round_snaps in &trace.snapshots_per_round {
+        let converged = round_snaps
+            .iter()
+            .filter(|(_, snap)| snap.entries.len() >= trace.total_keys && trace.total_keys > 0)
+            .count();
+        let pct = if n > 0 {
+            converged as f64 / n as f64 * 100.0
+        } else {
+            0.0
+        };
+        points.push(pct);
+    }
+
+    // Draw line.
+    let mut path = String::new();
+    for (i, &pct) in points.iter().enumerate() {
+        let x = margin_l + (i as f64 + 0.5) / rounds as f64 * chart_w;
+        let y = margin_t + chart_h - (pct / 100.0) * chart_h;
+        if i == 0 {
+            path.push_str(&format!("M{x:.1},{y:.1}"));
+        } else {
+            path.push_str(&format!(" L{x:.1},{y:.1}"));
+        }
+    }
+    html.push_str(&format!(
+        "<path d=\"{path}\" fill=\"none\" stroke=\"#4a90d9\" stroke-width=\"2\"/>\n"
+    ));
+
+    // Draw dots.
+    for (i, &pct) in points.iter().enumerate() {
+        let x = margin_l + (i as f64 + 0.5) / rounds as f64 * chart_w;
+        let y = margin_t + chart_h - (pct / 100.0) * chart_h;
+        html.push_str(&format!(
+            "<circle cx=\"{x:.1}\" cy=\"{y:.1}\" r=\"3\" fill=\"#4a90d9\"/>\n"
+        ));
+    }
+
+    html.push_str("</svg>\n");
+}
+
+// ── Message flow timeline ────────────────────────────────────────────────
+
+fn render_message_flow_svg(html: &mut String, trace: &SimulationTrace) {
+    html.push_str("<h2>Message Flow Timeline</h2>\n");
+    html.push_str("<p>Arrows show Push messages from sender to receiver, grouped by round.</p>\n");
+
+    let n = trace.node_names.len();
+    let rounds = trace.num_rounds;
+    if n == 0 || rounds == 0 {
+        html.push_str("<p>No data.</p>\n");
+        return;
+    }
+
+    let name_to_col: HashMap<&str, usize> = trace
+        .node_names
+        .iter()
+        .enumerate()
+        .map(|(i, n)| (n.as_str(), i))
+        .collect();
+
+    // Collect message arrows grouped by round.
+    let mut arrows_per_round: Vec<Vec<(usize, usize)>> = vec![Vec::new(); rounds];
+    for event in &trace.events {
+        if let GossipEventKind::PushReceived { ref from_name, .. } = event.kind {
+            let round_idx = event.tick.saturating_sub(1) as usize;
+            if round_idx < rounds {
+                if let (Some(&from_col), Some(&to_col)) = (
+                    name_to_col.get(from_name.as_str()),
+                    name_to_col.get(event.node_name.as_str()),
+                ) {
+                    arrows_per_round[round_idx].push((from_col, to_col));
+                }
+            }
+        }
+    }
+
+    let col_w = 80.0_f64;
+    let row_h = 40.0_f64;
+    let header_h = 30.0_f64;
+    let label_h = 24.0_f64;
+    let svg_w = col_w * n as f64 + 40.0;
+    let svg_h = header_h + label_h + row_h * rounds as f64 + 20.0;
+
+    html.push_str(&format!(
+        "<svg width=\"{svg_w:.0}\" height=\"{svg_h:.0}\" viewBox=\"0 0 {svg_w:.0} {svg_h:.0}\">\n"
+    ));
+    html.push_str("<defs><marker id=\"flow-arrow\" markerWidth=\"8\" markerHeight=\"6\" refX=\"8\" refY=\"3\" orient=\"auto\"><path d=\"M0,0 L8,3 L0,6\" fill=\"#d94a4a\"/></marker></defs>\n");
+
+    // Column headers (node names).
+    for (i, name) in trace.node_names.iter().enumerate() {
+        let x = 20.0 + i as f64 * col_w + col_w / 2.0;
+        html.push_str(&format!(
+            "<text x=\"{x:.1}\" y=\"{label_h:.0}\" text-anchor=\"middle\" font-size=\"11\" font-weight=\"bold\" fill=\"#333\">{name}</text>\n"
+        ));
+        // Vertical lifeline.
+        let y_start = header_h + label_h;
+        let y_end = header_h + label_h + row_h * rounds as f64;
+        html.push_str(&format!(
+            "<line x1=\"{x:.1}\" y1=\"{y_start:.0}\" x2=\"{x:.1}\" y2=\"{y_end:.0}\" stroke=\"#ddd\" stroke-width=\"1\" stroke-dasharray=\"4,3\"/>\n"
+        ));
+    }
+
+    // Round labels and arrows.
+    for (r, arrows) in arrows_per_round.iter().enumerate() {
+        let y = header_h + label_h + r as f64 * row_h + row_h / 2.0;
+        // Round label on left.
+        html.push_str(&format!(
+            "<text x=\"4\" y=\"{y:.1}\" font-size=\"9\" fill=\"#999\">R{}</text>\n",
+            r + 1,
+        ));
+
+        for &(from_col, to_col) in arrows {
+            let x1 = 20.0 + from_col as f64 * col_w + col_w / 2.0;
+            let x2 = 20.0 + to_col as f64 * col_w + col_w / 2.0;
+            // Offset slightly so overlapping arrows are visible.
+            let offset = if from_col < to_col { -3.0 } else { 3.0 };
+            html.push_str(&format!(
+                "<line x1=\"{x1:.1}\" y1=\"{y1:.1}\" x2=\"{x2:.1}\" y2=\"{y2:.1}\" stroke=\"#d94a4a\" stroke-width=\"1.5\" marker-end=\"url(#flow-arrow)\"/>\n",
+                y1 = y + offset,
+                y2 = y + offset,
+            ));
+        }
+    }
+
+    html.push_str("</svg>\n");
+}
+
+// ── Event log table ──────────────────────────────────────────────────────
+
+fn render_event_table(html: &mut String, trace: &SimulationTrace) {
+    html.push_str("<h2>Event Log</h2>\n");
+
+    let max_rows = 500;
+    let events: Vec<_> = trace
+        .events
+        .iter()
+        .filter(|e| !matches!(e.kind, GossipEventKind::StateSnapshot { .. }))
+        .collect();
+
+    let total = events.len();
+    let display = events.iter().take(max_rows);
+
+    html.push_str("<table>\n<tr><th>Round</th><th>Node</th><th>Event</th><th>Details</th></tr>\n");
+    for event in display {
+        let (kind_str, detail) = format_event_kind(&event.kind);
+        html.push_str(&format!(
+            "<tr><td>{}</td><td>{}</td><td>{kind_str}</td><td>{detail}</td></tr>\n",
+            event.tick,
+            escape_html(&event.node_name),
+        ));
+    }
+    html.push_str("</table>\n");
+
+    if total > max_rows {
+        html.push_str(&format!(
+            "<p class=\"capped\">Showing {max_rows} of {total} events.</p>\n"
+        ));
+    }
+}
+
+fn format_event_kind(kind: &GossipEventKind) -> (&'static str, String) {
+    match kind {
+        GossipEventKind::LocalSet { key } => ("LocalSet", format!("key={}", escape_html(key))),
+        GossipEventKind::GossipRoundStarted { target_name } => {
+            ("GossipRound", format!("→ {}", escape_html(target_name)))
+        }
+        GossipEventKind::GossipRoundNoPeers => ("GossipRound", "no peers".into()),
+        GossipEventKind::PushReceived {
+            from_name,
+            keys_updated,
+        } => (
+            "PushReceived",
+            format!(
+                "from {} ({keys_updated} updated)",
+                escape_html(from_name)
+            ),
+        ),
+        GossipEventKind::QueryReceived { key } => {
+            ("Query", format!("key={}", escape_html(key)))
+        }
+        GossipEventKind::PeerAdded { peer_name } => {
+            ("PeerAdded", escape_html(peer_name))
+        }
+        GossipEventKind::PeerRemoved { peer_name } => {
+            ("PeerRemoved", escape_html(peer_name))
+        }
+        GossipEventKind::StateSnapshot { .. } => ("Snapshot", String::new()),
+    }
+}
+
+fn escape_html(s: &str) -> String {
+    s.replace('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+        .replace('"', "&quot;")
+}

crates/swactor-gossip/src/sim.rs

@@ -0,0 +1,236 @@
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+
+use swactor::actor::ActorAddress;
+use swactor::config::RuntimeConfig;
+use swactor::runtime::Runtime;
+
+use crate::actor::GossipActor;
+use crate::message::GossipMessage;
+use crate::trace::{
+    EventLog, GossipEventKind, NameRegistry, NodeSnapshot, SimulationTrace, TickCounter,
+};
+
+// ── Configuration ────────────────────────────────────────────────────────
+
+#[derive(Debug, Clone)]
+pub enum Topology {
+    /// Each node gossips to the next; last gossips to first.
+    Ring,
+    /// Node 0 is the hub; all others gossip to/from it.
+    Star,
+    /// Every node gossips to every other node.
+    FullMesh,
+    /// Unidirectional chain: 0→1→2→…→(n-1).
+    Chain,
+    /// Two halves with no cross-links (healed later via `heal_after_round`).
+    Partitioned,
+}
+
+#[derive(Debug, Clone)]
+pub struct SimConfig {
+    pub name: String,
+    pub topology: Topology,
+    pub num_nodes: usize,
+    /// `(key, value)` pairs to set on node 0 before gossip starts.
+    pub initial_data: Vec<(String, Vec<u8>)>,
+    pub num_rounds: usize,
+    pub ticks_per_round: usize,
+    /// If `Some(r)`, cross-partition links are added after round `r`.
+    pub heal_after_round: Option<usize>,
+}
+
+// ── Public entry point ───────────────────────────────────────────────────
+
+pub fn run_simulation(config: SimConfig) -> SimulationTrace {
+    let event_log: EventLog = Arc::new(Mutex::new(Vec::new()));
+    let tick_counter: TickCounter = Arc::new(AtomicU64::new(0));
+    let name_registry: NameRegistry = Arc::new(Mutex::new(HashMap::new()));
+
+    let rt = Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    });
+
+    // Spawn nodes.
+    let mut addrs = Vec::with_capacity(config.num_nodes);
+    let mut names = Vec::with_capacity(config.num_nodes);
+    for i in 0..config.num_nodes {
+        let name = format!("node-{i}");
+        let actor = GossipActor::traced(
+            Arc::clone(&event_log),
+            Arc::clone(&tick_counter),
+            Arc::clone(&name_registry),
+        );
+        let addr = rt.spawn(actor).unwrap();
+        name_registry.lock().unwrap().insert(addr, name.clone());
+        addrs.push(addr);
+        names.push(name);
+    }
+
+    // Wire topology.
+    let edges = wire_topology(&rt, &config.topology, &addrs, &names);
+    // Deliver AddPeer messages.
+    for _ in 0..3 {
+        rt.tick();
+    }
+
+    // Set initial data on node 0.
+    let total_keys = config.initial_data.len();
+    for (key, value) in &config.initial_data {
+        rt.send_to(
+            addrs[0],
+            GossipMessage::Set {
+                key: key.clone(),
+                value: value.clone(),
+            },
+        )
+        .unwrap();
+    }
+    rt.tick();
+
+    // Run gossip rounds.
+    let mut snapshots_per_round: Vec<Vec<(String, NodeSnapshot)>> = Vec::new();
+
+    for round in 0..config.num_rounds {
+        // Heal partition if needed.
+        if config.heal_after_round == Some(round) {
+            heal_partition(&rt, &config.topology, &addrs, &names);
+            for _ in 0..3 {
+                rt.tick();
+            }
+        }
+
+        tick_counter.store((round + 1) as u64, Ordering::Relaxed);
+
+        // Trigger gossip on all nodes.
+        for &addr in &addrs {
+            rt.send_to(addr, GossipMessage::DoGossipRound).unwrap();
+        }
+        for _ in 0..config.ticks_per_round {
+            rt.tick();
+        }
+
+        // Take snapshots.
+        for &addr in &addrs {
+            rt.send_to(addr, GossipMessage::TakeSnapshot).unwrap();
+        }
+        for _ in 0..3 {
+            rt.tick();
+        }
+
+        // Extract snapshots from event log.
+        let current_round_tick = (round + 1) as u64;
+        let log = event_log.lock().unwrap();
+        let mut round_snapshots: Vec<(String, NodeSnapshot)> = Vec::new();
+        for event in log.iter().rev() {
+            if event.tick != current_round_tick {
+                break;
+            }
+            if let GossipEventKind::StateSnapshot { ref snapshot } = event.kind {
+                round_snapshots.push((event.node_name.clone(), snapshot.clone()));
+            }
+        }
+        round_snapshots.reverse();
+        snapshots_per_round.push(round_snapshots);
+    }
+
+    let events = event_log.lock().unwrap().clone();
+
+    SimulationTrace {
+        name: config.name,
+        node_names: names,
+        node_addrs: addrs,
+        topology_edges: edges,
+        events,
+        snapshots_per_round,
+        num_rounds: config.num_rounds,
+        total_keys,
+    }
+}
+
+// ── Topology wiring ──────────────────────────────────────────────────────
+
+fn wire_topology(
+    rt: &Runtime,
+    topology: &Topology,
+    addrs: &[ActorAddress],
+    names: &[String],
+) -> Vec<(String, String)> {
+    let n = addrs.len();
+    let mut edges = Vec::new();
+
+    let mut add_edge = |from: usize, to: usize| {
+        rt.send_to(addrs[from], GossipMessage::AddPeer(addrs[to]))
+            .unwrap();
+        edges.push((names[from].clone(), names[to].clone()));
+    };
+
+    match topology {
+        Topology::Ring => {
+            for i in 0..n {
+                add_edge(i, (i + 1) % n);
+            }
+        }
+        Topology::Star => {
+            for i in 1..n {
+                add_edge(0, i);
+                add_edge(i, 0);
+            }
+        }
+        Topology::FullMesh => {
+            for i in 0..n {
+                for j in 0..n {
+                    if i != j {
+                        add_edge(i, j);
+                    }
+                }
+            }
+        }
+        Topology::Chain => {
+            for i in 0..n.saturating_sub(1) {
+                add_edge(i, i + 1);
+            }
+        }
+        Topology::Partitioned => {
+            let half = n / 2;
+            // Wire each half as a full mesh.
+            for i in 0..half {
+                for j in 0..half {
+                    if i != j {
+                        add_edge(i, j);
+                    }
+                }
+            }
+            for i in half..n {
+                for j in half..n {
+                    if i != j {
+                        add_edge(i, j);
+                    }
+                }
+            }
+        }
+    }
+    edges
+}
+
+fn heal_partition(
+    rt: &Runtime,
+    topology: &Topology,
+    addrs: &[ActorAddress],
+    _names: &[String],
+) {
+    if !matches!(topology, Topology::Partitioned) {
+        return;
+    }
+    let n = addrs.len();
+    let half = n / 2;
+    // Add bidirectional links between the two halves (bridge nodes).
+    if half > 0 && half < n {
+        rt.send_to(addrs[half - 1], GossipMessage::AddPeer(addrs[half]))
+            .unwrap();
+        rt.send_to(addrs[half], GossipMessage::AddPeer(addrs[half - 1]))
+            .unwrap();
+    }
+}

crates/swactor-gossip/src/state.rs

@@ -0,0 +1,69 @@
+use std::collections::HashMap;
+
+#[derive(Debug, Clone)]
+pub struct VersionedValue {
+    pub value: Vec<u8>,
+    pub version: u64,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct GossipState {
+    entries: HashMap<String, VersionedValue>,
+}
+
+impl GossipState {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Insert or update a key. Auto-increments the version for that key.
+    /// Returns the new version number.
+    pub fn set(&mut self, key: String, value: Vec<u8>) -> u64 {
+        let new_version = self
+            .entries
+            .get(&key)
+            .map_or(1, |existing| existing.version + 1);
+        self.entries.insert(
+            key,
+            VersionedValue {
+                value,
+                version: new_version,
+            },
+        );
+        new_version
+    }
+
+    pub fn get(&self, key: &str) -> Option<&VersionedValue> {
+        self.entries.get(key)
+    }
+
+    pub fn entries(&self) -> &HashMap<String, VersionedValue> {
+        &self.entries
+    }
+
+    pub fn len(&self) -> usize {
+        self.entries.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.entries.is_empty()
+    }
+
+    /// Merge a remote state into this one. For each key, keep the entry
+    /// with the higher version (last-writer-wins). Returns the number of
+    /// entries that were updated.
+    pub fn merge(&mut self, remote: &GossipState) -> usize {
+        let mut updated = 0;
+        for (key, remote_val) in &remote.entries {
+            let dominated = match self.entries.get(key) {
+                Some(local_val) => remote_val.version > local_val.version,
+                None => true,
+            };
+            if dominated {
+                self.entries.insert(key.clone(), remote_val.clone());
+                updated += 1;
+            }
+        }
+        updated
+    }
+}

crates/swactor-gossip/src/trace.rs

@@ -0,0 +1,90 @@
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+
+use swactor::actor::ActorAddress;
+
+use crate::state::VersionedValue;
+
+// ── Shared handles ───────────────────────────────────────────────────────
+
+/// Shared, append-only event log.
+pub type EventLog = Arc<Mutex<Vec<GossipEvent>>>;
+
+/// Shared tick counter — the simulation harness increments this.
+pub type TickCounter = Arc<AtomicU64>;
+
+/// Maps actor addresses to human-readable names like `"node-0"`.
+pub type NameRegistry = Arc<Mutex<HashMap<ActorAddress, String>>>;
+
+// ── Event types ──────────────────────────────────────────────────────────
+
+#[derive(Debug, Clone)]
+pub struct GossipEvent {
+    pub tick: u64,
+    pub node_name: String,
+    pub node_addr: ActorAddress,
+    pub kind: GossipEventKind,
+}
+
+#[derive(Debug, Clone)]
+pub enum GossipEventKind {
+    /// A local `Set { key, .. }` was processed.
+    LocalSet { key: String },
+    /// `DoGossipRound` chose a peer and sent a Push.
+    GossipRoundStarted { target_name: String },
+    /// `DoGossipRound` had no peers.
+    GossipRoundNoPeers,
+    /// Received a Push from another node.
+    PushReceived {
+        from_name: String,
+        keys_updated: usize,
+    },
+    /// Received a Query.
+    QueryReceived { key: String },
+    /// A peer was added.
+    PeerAdded { peer_name: String },
+    /// A peer was removed.
+    PeerRemoved { peer_name: String },
+    /// Full state snapshot (requested via `TakeSnapshot`).
+    StateSnapshot { snapshot: NodeSnapshot },
+}
+
+#[derive(Debug, Clone)]
+pub struct NodeSnapshot {
+    pub entries: HashMap<String, VersionedValue>,
+    pub peer_count: usize,
+}
+
+// ── Simulation trace (complete run output) ───────────────────────────────
+
+#[derive(Debug, Clone)]
+pub struct SimulationTrace {
+    pub name: String,
+    pub node_names: Vec<String>,
+    pub node_addrs: Vec<ActorAddress>,
+    pub topology_edges: Vec<(String, String)>,
+    pub events: Vec<GossipEvent>,
+    pub snapshots_per_round: Vec<Vec<(String, NodeSnapshot)>>,
+    pub num_rounds: usize,
+    pub total_keys: usize,
+}
+
+// ── Helpers ──────────────────────────────────────────────────────────────
+
+pub fn current_tick(counter: &TickCounter) -> u64 {
+    counter.load(Ordering::Relaxed)
+}
+
+pub fn resolve_name(registry: &NameRegistry, addr: ActorAddress) -> String {
+    registry
+        .lock()
+        .unwrap()
+        .get(&addr)
+        .cloned()
+        .unwrap_or_else(|| format!("{:?}", &addr.0[..4]))
+}
+
+pub fn record_event(log: &EventLog, event: GossipEvent) {
+    log.lock().unwrap().push(event);
+}

crates/swactor-gossip/tests/gossip_convergence.rs

@@ -0,0 +1,243 @@
+use swactor::config::RuntimeConfig;
+use swactor::runtime::Runtime;
+use swactor_gossip::{GossipActor, GossipMessage, GossipQueryResponse};
+
+fn single_thread_runtime() -> Runtime {
+    Runtime::new(RuntimeConfig {
+        num_threads: 1,
+        ..Default::default()
+    })
+}
+
+/// Drive ticks until the inbox receives a response, or panic after a limit.
+fn recv_query_response(
+    rt: &Runtime,
+    inbox: &swactor::runtime::Inbox<GossipQueryResponse>,
+    max_ticks: usize,
+) -> GossipQueryResponse {
+    for _ in 0..max_ticks {
+        rt.tick();
+        if let Some(resp) = inbox.try_recv() {
+            return resp;
+        }
+    }
+    panic!("no GossipQueryResponse after {max_ticks} ticks");
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// Test 1: Value propagates through a chain A → B → C
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn value_propagates_through_chain() {
+    // Given: three gossip nodes wired A→B→C (each only gossips to the next)
+    let rt = single_thread_runtime();
+    let a = rt.spawn(GossipActor::new()).unwrap();
+    let b = rt.spawn(GossipActor::new()).unwrap();
+    let c = rt.spawn(GossipActor::new()).unwrap();
+
+    rt.send_to(a, GossipMessage::AddPeer(b)).unwrap();
+    rt.send_to(b, GossipMessage::AddPeer(c)).unwrap();
+    rt.tick(); // deliver AddPeer messages
+
+    // When: we set a value on A and trigger gossip hops
+    rt.send_to(a, GossipMessage::Set {
+        key: "color".into(),
+        value: b"blue".to_vec(),
+    })
+    .unwrap();
+    rt.tick(); // A processes Set
+
+    rt.send_to(a, GossipMessage::DoGossipRound).unwrap();
+    rt.tick(); // A pushes to B
+    rt.tick(); // B processes Push
+
+    rt.send_to(b, GossipMessage::DoGossipRound).unwrap();
+    rt.tick(); // B pushes to C
+    rt.tick(); // C processes Push
+
+    // Then: querying C returns the value that originated at A
+    let inbox = rt.new_inbox::<GossipQueryResponse>().unwrap();
+    rt.send_to(c, GossipMessage::Query {
+        key: "color".into(),
+        reply_to: *inbox.addr(),
+    })
+    .unwrap();
+
+    let resp = recv_query_response(&rt, &inbox, 10);
+    assert_eq!(resp.key, "color");
+    assert_eq!(resp.value.as_deref(), Some(b"blue".as_slice()));
+    assert_eq!(resp.version, Some(1));
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// Test 2: Higher version wins during merge
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn higher_version_wins() {
+    // Given: two nodes A and B, each set the same key at different versions
+    let rt = single_thread_runtime();
+    let a = rt.spawn(GossipActor::new()).unwrap();
+    let b = rt.spawn(GossipActor::new()).unwrap();
+
+    rt.send_to(a, GossipMessage::AddPeer(b)).unwrap();
+    rt.tick();
+
+    // A sets "x" once (version 1)
+    rt.send_to(a, GossipMessage::Set {
+        key: "x".into(),
+        value: b"old".to_vec(),
+    })
+    .unwrap();
+    rt.tick();
+
+    // B sets "x" three times (version 3)
+    for val in [b"v1".as_slice(), b"v2", b"new"] {
+        rt.send_to(b, GossipMessage::Set {
+            key: "x".into(),
+            value: val.to_vec(),
+        })
+        .unwrap();
+    }
+    rt.tick();
+
+    // When: A pushes its lower-version state to B
+    rt.send_to(a, GossipMessage::DoGossipRound).unwrap();
+    rt.tick(); // A sends Push
+    rt.tick(); // B receives Push
+
+    // Then: B still has the higher-version value
+    let inbox = rt.new_inbox::<GossipQueryResponse>().unwrap();
+    rt.send_to(b, GossipMessage::Query {
+        key: "x".into(),
+        reply_to: *inbox.addr(),
+    })
+    .unwrap();
+
+    let resp = recv_query_response(&rt, &inbox, 10);
+    assert_eq!(resp.value.as_deref(), Some(b"new".as_slice()));
+    assert_eq!(resp.version, Some(3));
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// Test 3: Disjoint keys merge — both nodes end up with both keys
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn disjoint_keys_merge() {
+    // Given: A owns key "a", B owns key "b", they are mutual peers
+    let rt = single_thread_runtime();
+    let a = rt.spawn(GossipActor::new()).unwrap();
+    let b = rt.spawn(GossipActor::new()).unwrap();
+
+    rt.send_to(a, GossipMessage::AddPeer(b)).unwrap();
+    rt.send_to(b, GossipMessage::AddPeer(a)).unwrap();
+    rt.tick();
+
+    rt.send_to(a, GossipMessage::Set {
+        key: "a".into(),
+        value: b"from-a".to_vec(),
+    })
+    .unwrap();
+    rt.send_to(b, GossipMessage::Set {
+        key: "b".into(),
+        value: b"from-b".to_vec(),
+    })
+    .unwrap();
+    rt.tick();
+
+    // When: both gossip to each other
+    rt.send_to(a, GossipMessage::DoGossipRound).unwrap();
+    rt.send_to(b, GossipMessage::DoGossipRound).unwrap();
+    rt.tick(); // send Pushes
+    rt.tick(); // receive Pushes
+
+    // Then: A has key "b" and B has key "a"
+    let inbox = rt.new_inbox::<GossipQueryResponse>().unwrap();
+
+    rt.send_to(a, GossipMessage::Query {
+        key: "b".into(),
+        reply_to: *inbox.addr(),
+    })
+    .unwrap();
+    let resp = recv_query_response(&rt, &inbox, 10);
+    assert_eq!(resp.key, "b");
+    assert_eq!(resp.value.as_deref(), Some(b"from-b".as_slice()));
+
+    rt.send_to(b, GossipMessage::Query {
+        key: "a".into(),
+        reply_to: *inbox.addr(),
+    })
+    .unwrap();
+    let resp = recv_query_response(&rt, &inbox, 10);
+    assert_eq!(resp.key, "a");
+    assert_eq!(resp.value.as_deref(), Some(b"from-a".as_slice()));
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// Test 4: Query for nonexistent key returns None
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn query_nonexistent_key_returns_none() {
+    // Given: a gossip node with no data
+    let rt = single_thread_runtime();
+    let a = rt.spawn(GossipActor::new()).unwrap();
+    rt.tick();
+
+    // When: we query a key that was never set
+    let inbox = rt.new_inbox::<GossipQueryResponse>().unwrap();
+    rt.send_to(a, GossipMessage::Query {
+        key: "ghost".into(),
+        reply_to: *inbox.addr(),
+    })
+    .unwrap();
+
+    // Then: response has None value and None version
+    let resp = recv_query_response(&rt, &inbox, 10);
+    assert_eq!(resp.key, "ghost");
+    assert!(resp.value.is_none());
+    assert!(resp.version.is_none());
+}
+
+// ────────────────────────────────────────────────────────────────────────────
+// Test 5: Idempotent push — double-push doesn't bump versions
+// ────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn idempotent_push() {
+    // Given: A has a key set, B is its peer
+    let rt = single_thread_runtime();
+    let a = rt.spawn(GossipActor::new()).unwrap();
+    let b = rt.spawn(GossipActor::new()).unwrap();
+
+    rt.send_to(a, GossipMessage::AddPeer(b)).unwrap();
+    rt.tick();
+
+    rt.send_to(a, GossipMessage::Set {
+        key: "k".into(),
+        value: b"val".to_vec(),
+    })
+    .unwrap();
+    rt.tick();
+
+    // When: A gossips to B twice (same state, same version)
+    for _ in 0..2 {
+        rt.send_to(a, GossipMessage::DoGossipRound).unwrap();
+        rt.tick(); // send Push
+        rt.tick(); // receive Push
+    }
+
+    // Then: B's version is still 1 (merge is idempotent, not additive)
+    let inbox = rt.new_inbox::<GossipQueryResponse>().unwrap();
+    rt.send_to(b, GossipMessage::Query {
+        key: "k".into(),
+        reply_to: *inbox.addr(),
+    })
+    .unwrap();
+
+    let resp = recv_query_response(&rt, &inbox, 10);
+    assert_eq!(resp.version, Some(1));
+    assert_eq!(resp.value.as_deref(), Some(b"val".as_slice()));
+}