Add connect/disconnect control commands and maelstrom chaos scenario

Add runtime peer management to the FIPS daemon via control socket
commands, and a new chaos simulation scenario that exercises dynamic
topology mutation with ephemeral node identities.

Daemon (connect/disconnect commands):
- Extend control socket Request with optional params field
- Add commands.rs module for mutating command dispatch, separate from
  read-only queries
- Add api_connect() on Node: builds ephemeral PeerConfig (no auto-
  reconnect), pre-seeds identity cache, reuses initiate_peer_connection
- Add api_disconnect() on Node: calls remove_active_peer(), clears
  retry_pending to suppress reconnection
- Route non-show_* commands to async command dispatch in rx_loop

fipsctl CLI:
- Add Connect and Disconnect subcommands accepting npub or hostname
- Resolve hostnames from /etc/fips/hosts before sending to daemon
- Refactor socket I/O into reusable send_request helper

Chaos simulator (maelstrom scenario):
- Add PeerChurnManager: periodically disconnects a random active link
  and connects a random unconnected node pair via control socket
- Add send_command() to control.py using base64-encoded JSON payloads
  to avoid shell quoting issues in docker exec
- Add PeerChurnConfig to scenario with interval and ephemeral_fraction
- Ephemeral identity support: nodes configured without nsec generate
  fresh keypairs on restart; simulator queries show_status for new
  npub and updates its cache via on_node_restart callback
- Add maelstrom.yaml: all chaos dimensions (netem, link flaps, node
  churn, peer topology churn, traffic) with 50% ephemeral identity

src/bin/fipsctl.rs

@@ -1,10 +1,11 @@
 //! fipsctl — FIPS control client
 //!
-//! Connects to the FIPS daemon's Unix domain control socket, sends a
-//! query command, and pretty-prints the JSON response.
+//! Connects to the FIPS daemon's Unix domain control socket, sends
+//! commands, and pretty-prints the JSON response.
 
 use clap::{Parser, Subcommand};
 use fips::config::{write_key_file, write_pub_file};
+use fips::upper::hosts::HostMap;
 use fips::version;
 use fips::{encode_nsec, Identity};
 use std::io::{BufRead, BufReader, Write};
@@ -18,7 +19,7 @@ use std::time::Duration;
     name = "fipsctl",
     version = version::short_version(),
     long_version = version::long_version(),
-    about = "Query a running FIPS daemon"
+    about = "Control a running FIPS daemon"
 )]
 struct Cli {
     /// Control socket path override
@@ -48,6 +49,20 @@ enum Commands {
         #[arg(short = 's', long = "stdout")]
         stdout: bool,
     },
+    /// Connect to a peer
+    Connect {
+        /// Peer identifier: npub (bech32) or hostname from /etc/fips/hosts
+        peer: String,
+        /// Transport address (e.g., "192.168.1.1:2121")
+        address: String,
+        /// Transport type: udp, tcp, tor, ethernet
+        transport: String,
+    },
+    /// Disconnect a peer
+    Disconnect {
+        /// Peer identifier: npub (bech32) or hostname from /etc/fips/hosts
+        peer: String,
+    },
 }
 
 #[derive(Subcommand, Debug)]
@@ -108,123 +123,58 @@ fn default_socket_path() -> PathBuf {
     }
 }
 
-fn main() {
-    let cli = Cli::parse();
-
-    // Commands that don't require a running daemon
-    match &cli.command {
-        Commands::Keygen {
-            dir,
-            force,
-            stdout,
-        } => {
-            let identity = Identity::generate();
-            let nsec = encode_nsec(&identity.keypair().secret_key());
-            let npub = identity.npub();
-
-            if *stdout {
-                println!("{}", nsec);
-                println!("{}", npub);
-                return;
-            }
-
-            let key_path = dir.join("fips.key");
-            let pub_path = dir.join("fips.pub");
-
-            if key_path.exists() && !force {
-                eprintln!("error: key file already exists: {}", key_path.display());
-                eprintln!("Use --force to overwrite.");
-                std::process::exit(1);
-            }
-
-            if let Err(e) = std::fs::create_dir_all(dir) {
-                eprintln!("error: cannot create directory {}: {}", dir.display(), e);
-                std::process::exit(1);
-            }
-
-            if let Err(e) = write_key_file(&key_path, &nsec) {
-                eprintln!("error: failed to write key file: {}", e);
-                std::process::exit(1);
-            }
-
-            if let Err(e) = write_pub_file(&pub_path, &npub) {
-                eprintln!("error: failed to write pub file: {}", e);
-                std::process::exit(1);
-            }
-
-            eprintln!("{}", npub);
-            eprintln!("Key files written to: {}/", dir.display());
-            eprintln!();
-            eprintln!("NOTE: Set 'node.identity.persistent: true' in fips.yaml");
-            eprintln!("      or these keys will be overwritten on next daemon start.");
-            return;
-        }
-        Commands::Show { .. } => {}
-    }
-
-    let socket_path = cli.socket.unwrap_or_else(default_socket_path);
-    let command_name = match &cli.command {
-        Commands::Show { what } => what.command_name(),
-        Commands::Keygen { .. } => unreachable!(),
-    };
-
-    // Connect to the control socket
-    let mut stream = match UnixStream::connect(&socket_path) {
-        Ok(s) => s,
-        Err(e) => {
-            eprintln!(
-                "error: cannot connect to {}: {}",
+/// Send a JSON request to the control socket and return the response.
+fn send_request(socket_path: &Path, request_json: &str) -> Result<serde_json::Value, String> {
+    let mut stream = UnixStream::connect(socket_path).map_err(|e| {
+        if e.kind() == std::io::ErrorKind::PermissionDenied {
+            format!(
+                "cannot connect to {}: {}\n\
+                 Hint: add your user to the 'fips' group: sudo usermod -aG fips $USER\n\
+                 Then log out and back in for the change to take effect.",
                 socket_path.display(),
                 e
-            );
-            if e.kind() == std::io::ErrorKind::PermissionDenied {
-                eprintln!(
-                    "Hint: add your user to the 'fips' group: sudo usermod -aG fips $USER"
-                );
-                eprintln!("Then log out and back in for the change to take effect.");
-            } else {
-                eprintln!("Is the FIPS daemon running?");
-            }
-            std::process::exit(1);
+            )
+        } else {
+            format!(
+                "cannot connect to {}: {}\nIs the FIPS daemon running?",
+                socket_path.display(),
+                e
+            )
         }
-    };
+    })?;
 
-    // Set timeouts
-    let timeout = Duration::from_secs(2);
+    let timeout = Duration::from_secs(5);
     let _ = stream.set_read_timeout(Some(timeout));
     let _ = stream.set_write_timeout(Some(timeout));
 
-    // Send request
-    let request = format!("{{\"command\":\"{}\"}}\n", command_name);
-    if let Err(e) = stream.write_all(request.as_bytes()) {
-        eprintln!("error: failed to send request: {}", e);
-        std::process::exit(1);
-    }
-
-    // Shutdown write half to signal end of request
+    stream
+        .write_all(request_json.as_bytes())
+        .map_err(|e| format!("failed to send request: {e}"))?;
     let _ = stream.shutdown(std::net::Shutdown::Write);
 
-    // Read response
     let reader = BufReader::new(&stream);
-    let response_line = match reader.lines().next() {
-        Some(Ok(line)) => line,
-        Some(Err(e)) => {
-            eprintln!("error: failed to read response: {}", e);
-            std::process::exit(1);
-        }
-        None => {
-            eprintln!("error: no response from daemon");
-            std::process::exit(1);
-        }
-    };
+    let line = reader
+        .lines()
+        .next()
+        .ok_or("no response from daemon")?
+        .map_err(|e| format!("failed to read response: {e}"))?;
 
-    // Parse and pretty-print
-    let Ok(value) = serde_json::from_str::<serde_json::Value>(&response_line) else {
-        // Not JSON, print raw
-        println!("{}", response_line);
-        return;
-    };
+    serde_json::from_str(&line).map_err(|e| format!("invalid response JSON: {e}"))
+}
 
+/// Build a request JSON string for a simple command (no params).
+fn build_query(command: &str) -> String {
+    format!("{{\"command\":\"{command}\"}}\n")
+}
+
+/// Build a request JSON string for a command with params.
+fn build_command(command: &str, params: serde_json::Value) -> String {
+    let req = serde_json::json!({"command": command, "params": params});
+    format!("{}\n", serde_json::to_string(&req).unwrap())
+}
+
+/// Print a control socket response, handling error status.
+fn print_response(value: &serde_json::Value) {
     let status = value
         .get("status")
         .and_then(|v| v.as_str())
@@ -235,15 +185,121 @@ fn main() {
             .get("message")
             .and_then(|v| v.as_str())
             .unwrap_or("unknown error");
-        eprintln!("error: {}", msg);
+        eprintln!("error: {msg}");
         std::process::exit(1);
     }
 
-    // Pretty-print the data field (or whole response if no data field)
     let output = if let Some(data) = value.get("data") {
-        serde_json::to_string_pretty(data).unwrap_or(response_line)
+        serde_json::to_string_pretty(data)
     } else {
-        serde_json::to_string_pretty(&value).unwrap_or(response_line)
+        serde_json::to_string_pretty(value)
     };
-    println!("{}", output);
+    println!("{}", output.unwrap_or_else(|_| format!("{value}")));
+}
+
+/// Resolve a peer identifier to an npub.
+///
+/// If the identifier starts with "npub1", it's returned as-is.
+/// Otherwise, it's looked up as a hostname in /etc/fips/hosts.
+fn resolve_peer(peer: &str) -> String {
+    if peer.starts_with("npub1") {
+        return peer.to_string();
+    }
+
+    let hosts = HostMap::load_hosts_file(Path::new(fips::upper::hosts::DEFAULT_HOSTS_PATH));
+    match hosts.lookup_npub(peer) {
+        Some(npub) => npub.to_string(),
+        None => {
+            eprintln!("error: unknown host '{peer}'");
+            eprintln!("Not found in /etc/fips/hosts and not an npub.");
+            std::process::exit(1);
+        }
+    }
+}
+
+fn main() {
+    let cli = Cli::parse();
+
+    // Commands that don't require a running daemon
+    if let Commands::Keygen {
+        dir,
+        force,
+        stdout,
+    } = &cli.command
+    {
+        let identity = Identity::generate();
+        let nsec = encode_nsec(&identity.keypair().secret_key());
+        let npub = identity.npub();
+
+        if *stdout {
+            println!("{nsec}");
+            println!("{npub}");
+            return;
+        }
+
+        let key_path = dir.join("fips.key");
+        let pub_path = dir.join("fips.pub");
+
+        if key_path.exists() && !force {
+            eprintln!("error: key file already exists: {}", key_path.display());
+            eprintln!("Use --force to overwrite.");
+            std::process::exit(1);
+        }
+
+        if let Err(e) = std::fs::create_dir_all(dir) {
+            eprintln!("error: cannot create directory {}: {e}", dir.display());
+            std::process::exit(1);
+        }
+
+        if let Err(e) = write_key_file(&key_path, &nsec) {
+            eprintln!("error: failed to write key file: {e}");
+            std::process::exit(1);
+        }
+
+        if let Err(e) = write_pub_file(&pub_path, &npub) {
+            eprintln!("error: failed to write pub file: {e}");
+            std::process::exit(1);
+        }
+
+        eprintln!("{npub}");
+        eprintln!("Key files written to: {}/", dir.display());
+        eprintln!();
+        eprintln!("NOTE: Set 'node.identity.persistent: true' in fips.yaml");
+        eprintln!("      or these keys will be overwritten on next daemon start.");
+        return;
+    }
+
+    let socket_path = cli.socket.unwrap_or_else(default_socket_path);
+
+    let request = match &cli.command {
+        Commands::Show { what } => build_query(what.command_name()),
+        Commands::Connect {
+            peer,
+            address,
+            transport,
+        } => {
+            let npub = resolve_peer(peer);
+            build_command(
+                "connect",
+                serde_json::json!({
+                    "npub": npub,
+                    "address": address,
+                    "transport": transport,
+                }),
+            )
+        }
+        Commands::Disconnect { peer } => {
+            let npub = resolve_peer(peer);
+            build_command("disconnect", serde_json::json!({"npub": npub}))
+        }
+        Commands::Keygen { .. } => unreachable!(),
+    };
+
+    match send_request(&socket_path, &request) {
+        Ok(value) => print_response(&value),
+        Err(e) => {
+            eprintln!("error: {e}");
+            std::process::exit(1);
+        }
+    }
 }

src/control/commands.rs

@@ -0,0 +1,68 @@
+//! Mutating control socket commands.
+//!
+//! Commands that modify node state (connect, disconnect) are handled here,
+//! separate from read-only queries in `queries.rs`.
+
+use super::protocol::Response;
+use crate::node::Node;
+use serde_json::Value;
+use tracing::info;
+
+/// Dispatch a mutating command to the appropriate handler.
+pub async fn dispatch(node: &mut Node, command: &str, params: Option<&Value>) -> Response {
+    match command {
+        "connect" => connect(node, params).await,
+        "disconnect" => disconnect(node, params),
+        _ => Response::error(format!("unknown command: {command}")),
+    }
+}
+
+/// Connect to a peer.
+///
+/// Params: `{"npub": "npub1...", "address": "host:port", "transport": "udp"}`
+async fn connect(node: &mut Node, params: Option<&Value>) -> Response {
+    let Some(params) = params else {
+        return Response::error("missing params for connect");
+    };
+
+    let npub = match params.get("npub").and_then(|v| v.as_str()) {
+        Some(v) => v,
+        None => return Response::error("missing 'npub' parameter"),
+    };
+    let address = match params.get("address").and_then(|v| v.as_str()) {
+        Some(v) => v,
+        None => return Response::error("missing 'address' parameter"),
+    };
+    let transport = match params.get("transport").and_then(|v| v.as_str()) {
+        Some(v) => v,
+        None => return Response::error("missing 'transport' parameter"),
+    };
+
+    info!(npub = %npub, address = %address, transport = %transport, "API connect requested");
+
+    match node.api_connect(npub, address, transport).await {
+        Ok(data) => Response::ok(data),
+        Err(msg) => Response::error(msg),
+    }
+}
+
+/// Disconnect a peer.
+///
+/// Params: `{"npub": "npub1..."}`
+fn disconnect(node: &mut Node, params: Option<&Value>) -> Response {
+    let Some(params) = params else {
+        return Response::error("missing params for disconnect");
+    };
+
+    let npub = match params.get("npub").and_then(|v| v.as_str()) {
+        Some(v) => v,
+        None => return Response::error("missing 'npub' parameter"),
+    };
+
+    info!(npub = %npub, "API disconnect requested");
+
+    match node.api_disconnect(npub) {
+        Ok(data) => Response::ok(data),
+        Err(msg) => Response::error(msg),
+    }
+}

src/control/mod.rs

@@ -1,9 +1,10 @@
-//! Control socket for runtime observability.
+//! Control socket for runtime management and observability.
 //!
-//! Provides a Unix domain socket that accepts query commands and returns
-//! structured JSON data about the node's current state. Read-only queries
-//! only — no state mutation through this channel.
+//! Provides a Unix domain socket that accepts commands and returns
+//! structured JSON responses. Supports both read-only queries (show_*)
+//! and mutating commands (connect, disconnect).
 
+pub mod commands;
 pub mod protocol;
 pub mod queries;
 

src/control/protocol.rs

@@ -8,8 +8,11 @@ use serde::{Deserialize, Serialize};
 /// A control request from a client.
 #[derive(Debug, Deserialize)]
 pub struct Request {
-    /// The command to execute (e.g., "show_status", "show_peers").
+    /// The command to execute (e.g., "show_status", "connect").
     pub command: String,
+    /// Optional parameters for mutating commands.
+    #[serde(default)]
+    pub params: Option<serde_json::Value>,
 }
 
 /// A control response to a client.
@@ -81,6 +84,24 @@ mod tests {
         assert_eq!(req.command, "show_peers");
     }
 
+    #[test]
+    fn test_deserialize_request_with_params() {
+        let json = r#"{"command": "connect", "params": {"npub": "npub1abc", "address": "1.2.3.4:2121", "transport": "udp"}}"#;
+        let req: Request = serde_json::from_str(json).unwrap();
+        assert_eq!(req.command, "connect");
+        let params = req.params.unwrap();
+        assert_eq!(params["npub"], "npub1abc");
+        assert_eq!(params["transport"], "udp");
+    }
+
+    #[test]
+    fn test_deserialize_request_without_params() {
+        let json = r#"{"command": "show_status"}"#;
+        let req: Request = serde_json::from_str(json).unwrap();
+        assert_eq!(req.command, "show_status");
+        assert!(req.params.is_none());
+    }
+
     #[test]
     fn test_deserialize_malformed_request() {
         let json = r#"{"not_command": "foo"}"#;

src/node/handlers/rx_loop.rs

@@ -1,6 +1,6 @@
 //! RX event loop and packet dispatch.
 
-use crate::control::ControlSocket;
+use crate::control::{commands, ControlSocket};
 use crate::control::queries;
 use crate::node::{Node, NodeError};
 use crate::transport::ReceivedPacket;
@@ -98,7 +98,15 @@ impl Node {
                     self.register_identity(identity.node_addr, identity.pubkey);
                 }
                 Some((request, response_tx)) = control_rx.recv() => {
-                    let response = queries::dispatch(self, &request.command);
+                    let response = if request.command.starts_with("show_") {
+                        queries::dispatch(self, &request.command)
+                    } else {
+                        commands::dispatch(
+                            self,
+                            &request.command,
+                            request.params.as_ref(),
+                        ).await
+                    };
                     let _ = response_tx.send(response);
                 }
                 _ = tick.tick() => {

src/node/lifecycle.rs

@@ -739,4 +739,76 @@ impl Node {
 
         info!(sent, total = peer_addrs.len(), reason = %reason, "Sent disconnect notifications");
     }
+
+    // === Control API methods ===
+
+    /// Connect to a peer via the control API.
+    ///
+    /// Creates an ephemeral peer connection (not persisted to config, no
+    /// auto-reconnect). Reuses the same connection path as auto-connect
+    /// peers. Returns JSON data on success or an error message.
+    pub(crate) async fn api_connect(
+        &mut self,
+        npub: &str,
+        address: &str,
+        transport: &str,
+    ) -> Result<serde_json::Value, String> {
+        let peer_config = crate::config::PeerConfig {
+            npub: npub.to_string(),
+            alias: None,
+            addresses: vec![crate::config::PeerAddress::new(transport, address)],
+            connect_policy: crate::config::ConnectPolicy::Manual,
+            auto_reconnect: false,
+        };
+
+        // Pre-seed identity cache (same as initiate_peer_connections does)
+        if let Ok(identity) = PeerIdentity::from_npub(npub) {
+            self.peer_aliases
+                .insert(*identity.node_addr(), identity.short_npub());
+            self.register_identity(*identity.node_addr(), identity.pubkey_full());
+        }
+
+        self.initiate_peer_connection(&peer_config)
+            .await
+            .map(|()| {
+                info!(
+                    npub = %npub,
+                    address = %address,
+                    transport = %transport,
+                    "API connect initiated"
+                );
+                serde_json::json!({
+                    "npub": npub,
+                    "address": address,
+                    "transport": transport,
+                })
+            })
+            .map_err(|e| e.to_string())
+    }
+
+    /// Disconnect a peer via the control API.
+    ///
+    /// Removes the peer and suppresses auto-reconnect.
+    pub(crate) fn api_disconnect(&mut self, npub: &str) -> Result<serde_json::Value, String> {
+        let peer_identity = PeerIdentity::from_npub(npub)
+            .map_err(|e| format!("invalid npub '{npub}': {e}"))?;
+        let node_addr = *peer_identity.node_addr();
+
+        if !self.peers.contains_key(&node_addr) {
+            return Err(format!("peer not found: {npub}"));
+        }
+
+        // Remove the peer (full cleanup: sessions, indices, links, tree, bloom)
+        self.remove_active_peer(&node_addr);
+
+        // Suppress any pending auto-reconnect
+        self.retry_pending.remove(&node_addr);
+
+        info!(npub = %npub, "API disconnect completed");
+
+        Ok(serde_json::json!({
+            "npub": npub,
+            "disconnected": true,
+        }))
+    }
 }

testing/chaos/scenarios/maelstrom.yaml

@@ -0,0 +1,76 @@
+# Maelstrom: topology mutation + node churn + ephemeral identity
+#
+# Combines all chaos dimensions: netem impairments, link flaps, node
+# churn, AND peer-level topology mutation (connect/disconnect).
+# Half the nodes have ephemeral identities that change on restart.
+#
+# Default: 20 nodes. Override with --nodes flag.
+
+scenario:
+  name: "maelstrom"
+  seed: 42
+  duration_secs: 600
+
+topology:
+  num_nodes: 20
+  algorithm: erdos_renyi
+  params:
+    p: 0.3
+  ensure_connected: true
+  subnet: "172.20.0.0/16"
+  ip_start: 10
+  transport_mix:
+    udp: 0.8
+    tcp: 0.2
+
+netem:
+  enabled: true
+  default_policy:
+    delay_ms: { min: 5, max: 50 }
+    jitter_ms: { min: 1, max: 10 }
+    loss_pct: { min: 0, max: 2 }
+  mutation:
+    interval_secs: { min: 20, max: 45 }
+    fraction: 0.3
+    policies:
+      normal:
+        delay_ms: [5, 20]
+        loss_pct: [0, 1]
+      degraded:
+        delay_ms: [50, 100]
+        jitter_ms: [10, 30]
+        loss_pct: [3, 8]
+
+link_flaps:
+  enabled: true
+  interval_secs: { min: 30, max: 60 }
+  max_down_links: 3
+  down_duration_secs: { min: 10, max: 30 }
+  protect_connectivity: true
+
+traffic:
+  enabled: true
+  max_concurrent: 5
+  interval_secs: { min: 5, max: 30 }
+  duration_secs: { min: 5, max: 60 }
+  parallel_streams: 4
+
+node_churn:
+  enabled: true
+  interval_secs: { min: 60, max: 120 }
+  max_down_nodes: 3
+  down_duration_secs: { min: 30, max: 90 }
+  protect_connectivity: false
+
+peer_churn:
+  enabled: true
+  interval_secs: { min: 8, max: 12 }
+  ephemeral_fraction: 0.5
+
+bandwidth:
+  enabled: true
+  tiers_mbps: [1, 10, 100, 1000]
+
+logging:
+  rust_log: "debug"
+  output_dir: "./sim-results"

testing/chaos/sim/config_gen.py

@@ -107,8 +107,13 @@ def generate_node_config(
     node_id: str,
     outbound_peers: list[str],
     fips_overrides: dict | None = None,
+    ephemeral: bool = False,
 ) -> str:
-    """Generate a complete FIPS config YAML for one node."""
+    """Generate a complete FIPS config YAML for one node.
+
+    If ephemeral is True, the nsec is omitted from the config so the
+    daemon generates a fresh keypair on each restart.
+    """
     template = _load_template()
     node = topology.nodes[node_id]
     peers_yaml = generate_peers_block(topology, node_id, outbound_peers)
@@ -120,6 +125,13 @@ def generate_node_config(
     config = config.replace("{{NSEC}}", node.nsec)
     config = config.replace("{{PEERS}}", peers_yaml)
 
+    # Ephemeral nodes: remove nsec so daemon generates fresh keys on restart
+    if ephemeral:
+        parsed = yaml.safe_load(config)
+        identity = parsed.get("node", {}).get("identity", {})
+        identity.pop("nsec", None)
+        config = yaml.dump(parsed, default_flow_style=False, sort_keys=False)
+
     # Determine which transports this node participates in
     eth_ifaces = topology.ethernet_interfaces(node_id)
     has_tcp = bool(topology.tcp_peers(node_id))
@@ -168,14 +180,17 @@ def write_configs(
     topology: SimTopology,
     output_dir: str,
     fips_overrides: dict | None = None,
+    ephemeral_nodes: set[str] | None = None,
 ):
     """Write all node configs and npubs.env to the output directory."""
     os.makedirs(output_dir, exist_ok=True)
+    ephemeral_nodes = ephemeral_nodes or set()
 
     outbound = topology.directed_outbound()
     for node_id in topology.nodes:
         config = generate_node_config(
-            topology, node_id, outbound[node_id], fips_overrides
+            topology, node_id, outbound[node_id], fips_overrides,
+            ephemeral=(node_id in ephemeral_nodes),
         )
         path = os.path.join(output_dir, f"{node_id}.yaml")
         with open(path, "w") as f:

testing/chaos/sim/control.py

@@ -56,6 +56,52 @@ def query_node(container: str, command: str, timeout: int = 10) -> dict | None:
     return response.get("data", {})
 
 
+def send_command(
+    container: str, command: str, params: dict, timeout: int = 10
+) -> dict | None:
+    """Send a mutating command with params to a node's control socket.
+
+    Returns the response data dict, or None on failure.
+    Uses base64-encoded JSON to avoid shell quoting issues with
+    embedded quotes in the payload.
+    """
+    import base64
+
+    payload = json.dumps({"command": command, "params": params})
+    b64 = base64.b64encode(payload.encode()).decode()
+    script = (
+        "import socket,json,sys,base64; "
+        "s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM); "
+        f"s.connect('{CONTROL_SOCKET}'); "
+        f"s.sendall(base64.b64decode('{b64}')+b'\\n'); "
+        "s.shutdown(socket.SHUT_WR); "
+        "chunks=[]; "
+        "[chunks.append(d) for d in iter(lambda:s.recv(65536),b'')]; "
+        "print(b''.join(chunks).decode())"
+    )
+    stdout = docker_exec_quiet(container, f"python3 -c \"{script}\"", timeout=timeout)
+    if stdout is None:
+        return None
+
+    try:
+        response = json.loads(stdout.strip())
+    except json.JSONDecodeError as e:
+        log.warning("Invalid JSON from %s: %s", container, e)
+        return None
+
+    if response.get("status") != "ok":
+        msg = response.get("message", "unknown error")
+        log.debug("Command %s on %s: %s", command, container, msg)
+        return None
+
+    return response.get("data", {})
+
+
+def query_status(container: str) -> dict | None:
+    """Query a node's status (npub, ipv6, uptime, etc.)."""
+    return query_node(container, "show_status")
+
+
 def query_tree(container: str) -> dict | None:
     """Query a node's spanning tree state."""
     return query_node(container, "show_tree")

testing/chaos/sim/nodes.py

@@ -43,6 +43,7 @@ class NodeManager:
         netem_mgr=None,
         down_nodes: set[str] | None = None,
         veth_mgr=None,
+        on_node_restart=None,
     ):
         self.topology = topology
         self.config = config
@@ -50,6 +51,7 @@ class NodeManager:
         self.netem_mgr = netem_mgr
         self.veth_mgr = veth_mgr
         self.down_nodes = down_nodes or set()
+        self.on_node_restart = on_node_restart
         self.node_states: dict[str, NodeState] = {
             nid: NodeState(node_id=nid) for nid in topology.nodes
         }
@@ -162,6 +164,10 @@ class NodeManager:
                 time.sleep(1)
             self.netem_mgr.setup_node(node_id)
 
+        # Notify callback (e.g., refresh npub for ephemeral identity nodes)
+        if self.on_node_restart:
+            self.on_node_restart(node_id)
+
     def _would_disconnect(self, node_id: str) -> bool:
         """Check if removing this node (plus currently-down nodes) disconnects the graph.
 

testing/chaos/sim/peer_churn.py

@@ -0,0 +1,178 @@
+"""Peer-level topology churn via connect/disconnect commands.
+
+Unlike NodeManager (which stops/starts containers), this uses the
+fipsctl connect/disconnect API to dynamically add and remove individual
+peer connections while nodes stay running. The topology graph evolves
+over time: random links are disconnected and new random pairs connected.
+
+Supports ephemeral identity nodes — half the nodes (configurable) get
+new keypairs on each container restart, requiring the simulator to
+track current npubs via show_status queries.
+"""
+
+from __future__ import annotations
+
+import logging
+import random
+import time
+
+from .control import query_status, send_command
+from .scenario import PeerChurnConfig
+from .topology import SimTopology
+
+log = logging.getLogger(__name__)
+
+
+class PeerChurnManager:
+    """Manages peer-level topology churn using connect/disconnect commands."""
+
+    def __init__(
+        self,
+        topology: SimTopology,
+        config: PeerChurnConfig,
+        rng: random.Random,
+        down_nodes: set[str],
+        ephemeral_nodes: set[str] | None = None,
+    ):
+        self.topology = topology
+        self.config = config
+        self.rng = rng
+        self.down_nodes = down_nodes
+        self.ephemeral_nodes = ephemeral_nodes or set()
+
+        # Current npub for each node (populated during init)
+        self.npub_cache: dict[str, str] = {}
+
+        # Currently active edges (start with topology edges)
+        self.active_edges: set[tuple[str, str]] = set()
+        for a, b in topology.edges:
+            self.active_edges.add(self._canonical(a, b))
+
+        # Edges disconnected by peer churn (not yet reconnected)
+        self.churned_count = 0
+
+    @staticmethod
+    def _canonical(a: str, b: str) -> tuple[str, str]:
+        """Canonical edge ordering (sorted)."""
+        return (min(a, b), max(a, b))
+
+    @property
+    def churn_count(self) -> int:
+        return self.churned_count
+
+    def refresh_npub(self, node_id: str) -> str | None:
+        """Query a node's current npub and update the cache.
+
+        Returns the npub or None if the query failed.
+        """
+        container = self.topology.container_name(node_id)
+        status = query_status(container)
+        if status and "npub" in status:
+            npub = status["npub"]
+            self.npub_cache[node_id] = npub
+            return npub
+        return None
+
+    def refresh_all_npubs(self):
+        """Populate npub cache for all nodes."""
+        for node_id in self.topology.nodes:
+            if node_id not in self.down_nodes:
+                self.refresh_npub(node_id)
+        log.info(
+            "Cached npubs for %d/%d nodes",
+            len(self.npub_cache),
+            len(self.topology.nodes),
+        )
+
+    def maybe_churn(self):
+        """Disconnect a random active link, then connect a random new pair."""
+        # Skip if too many nodes are down
+        up_nodes = [n for n in self.topology.nodes if n not in self.down_nodes]
+        if len(up_nodes) < 3:
+            return
+
+        # Phase 1: Disconnect a random active link between up nodes
+        candidates = [
+            (a, b)
+            for a, b in self.active_edges
+            if a not in self.down_nodes and b not in self.down_nodes
+        ]
+        if candidates:
+            edge = self.rng.choice(candidates)
+            if self._disconnect_edge(edge[0], edge[1]):
+                self.active_edges.discard(self._canonical(edge[0], edge[1]))
+                self.churned_count += 1
+
+        # Phase 2: Connect a random pair that isn't currently connected
+        non_edges = []
+        for i, a in enumerate(up_nodes):
+            for b in up_nodes[i + 1 :]:
+                if self._canonical(a, b) not in self.active_edges:
+                    non_edges.append((a, b))
+
+        if non_edges:
+            a, b = self.rng.choice(non_edges)
+            if self._connect_edge(a, b):
+                self.active_edges.add(self._canonical(a, b))
+
+    def _disconnect_edge(self, a: str, b: str) -> bool:
+        """Disconnect both sides of a link."""
+        npub_a = self.npub_cache.get(a)
+        npub_b = self.npub_cache.get(b)
+        if not npub_a or not npub_b:
+            log.debug("Missing npub for %s or %s, skipping disconnect", a, b)
+            return False
+
+        container_a = self.topology.container_name(a)
+        container_b = self.topology.container_name(b)
+
+        ok_a = send_command(container_a, "disconnect", {"npub": npub_b})
+        ok_b = send_command(container_b, "disconnect", {"npub": npub_a})
+
+        if ok_a is not None or ok_b is not None:
+            log.info("Peer DISCONNECT: %s -- %s", a, b)
+            return True
+
+        log.debug("Disconnect failed for %s -- %s", a, b)
+        return False
+
+    def _connect_edge(self, a: str, b: str) -> bool:
+        """Connect both sides of a new link (mutual outbound)."""
+        npub_a = self.npub_cache.get(a)
+        npub_b = self.npub_cache.get(b)
+        if not npub_a or not npub_b:
+            log.debug("Missing npub for %s or %s, skipping connect", a, b)
+            return False
+
+        # Use UDP transport with the node's Docker IP
+        ip_a = self.topology.nodes[a].docker_ip
+        ip_b = self.topology.nodes[b].docker_ip
+        port = 2121  # Default UDP port
+
+        container_a = self.topology.container_name(a)
+        container_b = self.topology.container_name(b)
+
+        # Node A connects to B
+        ok_a = send_command(
+            container_a,
+            "connect",
+            {"npub": npub_b, "address": f"{ip_b}:{port}", "transport": "udp"},
+        )
+
+        # Node B connects to A
+        ok_b = send_command(
+            container_b,
+            "connect",
+            {"npub": npub_a, "address": f"{ip_a}:{port}", "transport": "udp"},
+        )
+
+        if ok_a is not None or ok_b is not None:
+            log.info("Peer CONNECT: %s -- %s (udp)", a, b)
+            return True
+
+        log.debug("Connect failed for %s -- %s", a, b)
+        return False
+
+    def restore_all(self):
+        """No-op for teardown — peer connections are ephemeral."""
+        pass

testing/chaos/sim/runner.py

@@ -19,6 +19,7 @@ from .links import LinkManager
 from .logs import AnalysisResult, analyze_logs, collect_logs, write_sim_metadata
 from .netem import NetemManager
 from .nodes import NodeManager
+from .peer_churn import PeerChurnManager
 from .scenario import Scenario
 from .topology import SimTopology, generate_topology
 from .traffic import TrafficManager
@@ -46,6 +47,7 @@ class SimRunner:
         self.link_mgr: LinkManager | None = None
         self.traffic_mgr: TrafficManager | None = None
         self.node_mgr: NodeManager | None = None
+        self.peer_churn_mgr: PeerChurnManager | None = None
 
     def run(self) -> AnalysisResult | None:
         """Run the full simulation lifecycle."""
@@ -113,7 +115,23 @@ class SimRunner:
         config_dir = os.path.normpath(
             os.path.join(docker_network_dir, "generated-configs", "sim")
         )
-        write_configs(self.topology, config_dir, self.scenario.fips_overrides)
+        # Select ephemeral identity nodes (if peer churn enabled)
+        self._ephemeral_nodes: set[str] = set()
+        if s.peer_churn.enabled and s.peer_churn.ephemeral_fraction > 0:
+            all_nodes = sorted(self.topology.nodes.keys())
+            count = int(len(all_nodes) * s.peer_churn.ephemeral_fraction)
+            self._ephemeral_nodes = set(self.rng.sample(all_nodes, count))
+            log.info(
+                "Ephemeral identity nodes (%d/%d): %s",
+                len(self._ephemeral_nodes),
+                len(all_nodes),
+                ", ".join(sorted(self._ephemeral_nodes)),
+            )
+
+        write_configs(
+            self.topology, config_dir, self.scenario.fips_overrides,
+            ephemeral_nodes=self._ephemeral_nodes,
+        )
         log.info("Wrote node configs to %s", config_dir)
 
         # 3. Generate docker-compose.yml
@@ -167,6 +185,14 @@ class SimRunner:
                 self.topology, s.node_churn, self.rng,
                 netem_mgr=self.netem_mgr, down_nodes=self._down_nodes,
                 veth_mgr=self.veth_mgr,
+                on_node_restart=self._handle_node_restart,
+            )
+
+        if s.peer_churn.enabled:
+            self.peer_churn_mgr = PeerChurnManager(
+                self.topology, s.peer_churn, self.rng,
+                down_nodes=self._down_nodes,
+                ephemeral_nodes=self._ephemeral_nodes,
             )
 
     def _warmup(self):
@@ -177,6 +203,30 @@ class SimRunner:
         self._sleep(wait)
         self._take_snapshot("warmup")
 
+        # Populate npub cache after convergence (nodes must be running)
+        if self.peer_churn_mgr:
+            self.peer_churn_mgr.refresh_all_npubs()
+
+    def _handle_node_restart(self, node_id: str):
+        """Called after a node container is restarted.
+
+        For ephemeral identity nodes, waits briefly for the daemon to
+        start, then queries its new npub and updates the peer churn
+        manager's cache.
+        """
+        if not self.peer_churn_mgr:
+            return
+        if node_id not in self.peer_churn_mgr.ephemeral_nodes:
+            return
+
+        # Brief delay for daemon startup before querying control socket
+        time.sleep(2)
+        new_npub = self.peer_churn_mgr.refresh_npub(node_id)
+        if new_npub:
+            log.info("Ephemeral node %s new identity: %s...%s", node_id, new_npub[:12], new_npub[-6:])
+        else:
+            log.warning("Failed to refresh npub for ephemeral node %s", node_id)
+
     def _simulation_loop(self):
         """Main event loop driving stochastic behavior."""
         start = time.time()
@@ -189,6 +239,7 @@ class SimRunner:
         next_flap = self._schedule_next(start, s.link_flaps.interval_secs) if self.link_mgr else float("inf")
         next_traffic = self._schedule_next(start, s.traffic.interval_secs) if self.traffic_mgr else float("inf")
         next_churn = self._schedule_next(start, s.node_churn.interval_secs) if self.node_mgr else float("inf")
+        next_peer_churn = self._schedule_next(start, s.peer_churn.interval_secs) if self.peer_churn_mgr else float("inf")
 
         while not self._interrupted:
             now = time.time()
@@ -222,17 +273,26 @@ class SimRunner:
                     next_churn = self._schedule_next(now, s.node_churn.interval_secs)
                 self.node_mgr.restore_expired()
 
+            # Peer churn (topology mutation)
+            if self.peer_churn_mgr:
+                if now >= next_peer_churn:
+                    self.peer_churn_mgr.maybe_churn()
+                    next_peer_churn = self._schedule_next(now, s.peer_churn.interval_secs)
+
             # Status line
             down_links = self.link_mgr.down_count if self.link_mgr else 0
             down_nodes = self.node_mgr.down_count if self.node_mgr else 0
             active = self.traffic_mgr.active_count if self.traffic_mgr else 0
+            peer_churns = self.peer_churn_mgr.churn_count if self.peer_churn_mgr else 0
+            status_extra = f" peer_churns={peer_churns}" if self.peer_churn_mgr else ""
             print(
                 f"\r  [{elapsed:.0f}s/{duration}s] "
                 f"nodes={len(self.topology.nodes)} "
                 f"edges={len(self.topology.edges)} "
                 f"links_down={down_links} "
                 f"nodes_down={down_nodes} "
-                f"traffic={active}   ",
+                f"traffic={active}"
+                f"{status_extra}   ",
                 end="",
                 flush=True,
             )

testing/chaos/sim/scenario.py

@@ -106,6 +106,20 @@ class NodeChurnConfig:
     protect_connectivity: bool = True
 
 
+@dataclass
+class PeerChurnConfig:
+    """Peer-level topology churn via connect/disconnect commands.
+
+    When enabled, periodically disconnects a random active link and
+    connects a random unconnected node pair, causing the mesh topology
+    to evolve over time.
+    """
+
+    enabled: bool = False
+    interval_secs: Range = field(default_factory=lambda: Range(8, 12))
+    ephemeral_fraction: float = 0.0
+
+
 @dataclass
 class BandwidthConfig:
     """Per-link bandwidth pacing via HTB rate limiting.
@@ -152,6 +166,7 @@ class Scenario:
     link_flaps: LinkFlapsConfig = field(default_factory=LinkFlapsConfig)
     traffic: TrafficConfig = field(default_factory=TrafficConfig)
     node_churn: NodeChurnConfig = field(default_factory=NodeChurnConfig)
+    peer_churn: PeerChurnConfig = field(default_factory=PeerChurnConfig)
     bandwidth: BandwidthConfig = field(default_factory=BandwidthConfig)
     ingress: IngressConfig = field(default_factory=IngressConfig)
     logging: LoggingConfig = field(default_factory=LoggingConfig)
@@ -279,6 +294,13 @@ def load_scenario(path: str) -> Scenario:
         )
     s.node_churn.protect_connectivity = nc2.get("protect_connectivity", True)
 
+    # Peer churn section
+    pc = raw.get("peer_churn", {})
+    s.peer_churn.enabled = pc.get("enabled", False)
+    if "interval_secs" in pc:
+        s.peer_churn.interval_secs = _parse_range(pc["interval_secs"], "peer_churn.interval_secs")
+    s.peer_churn.ephemeral_fraction = float(pc.get("ephemeral_fraction", 0.0))
+
     # Bandwidth section
     bw = raw.get("bandwidth", {})
     s.bandwidth.enabled = bw.get("enabled", False)
@@ -399,6 +421,10 @@ def _validate(s: Scenario):
         s.node_churn.down_duration_secs.validate("node_churn.down_duration_secs")
         if s.node_churn.max_down_nodes >= s.topology.num_nodes:
             raise ValueError("node_churn.max_down_nodes must be < topology.num_nodes")
+    if s.peer_churn.enabled:
+        s.peer_churn.interval_secs.validate("peer_churn.interval_secs")
+        if not 0.0 <= s.peer_churn.ephemeral_fraction <= 1.0:
+            raise ValueError("peer_churn.ephemeral_fraction must be between 0.0 and 1.0")
     if s.bandwidth.enabled:
         for tier in s.bandwidth.tiers_mbps:
             if tier <= 0: