feat(app): wire the core node and implement the run command

crates/app/Cargo.toml

@@ -12,8 +12,15 @@ axum.workspace = true
 backon.workspace = true
 chrono.workspace = true
 futures.workspace = true
+libp2p.workspace = true
 pluto-core.workspace = true
+pluto-consensus.workspace = true
+pluto-eth1wrap.workspace = true
+pluto-featureset.workspace = true
 pluto-eth2api.workspace = true
+pluto-p2p.workspace = true
+pluto-parsigex.workspace = true
+pluto-peerinfo.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 vise.workspace = true
@@ -38,7 +45,6 @@ pluto-cluster.workspace = true
 pluto-k1util.workspace = true
 pluto-crypto.workspace = true
 pluto-ssz.workspace = true
-pluto-p2p.workspace = true
 
 [build-dependencies]
 pluto-build-proto.workspace = true
@@ -49,6 +55,7 @@ pluto-testutil.workspace = true
 wiremock.workspace = true
 test-case.workspace = true
 tower = { workspace = true, features = ["util"] }
+rand.workspace = true
 
 [lints]
 workspace = true

crates/app/src/lib.rs

@@ -40,3 +40,8 @@ pub mod utils;
 /// fixed set of checks over a rolling window, and publishes per-check pass/fail
 /// state as the `app_health_checks` gauge.
 pub mod health;
+
+/// Distributed-validator node wiring: constructs and connects the core duty
+/// workflow components (scheduler, fetcher, consensus, dutydb, validatorapi,
+/// parsigdb, parsigex, sigagg, aggsigdb, broadcaster) into a runnable node.
+pub mod node;

crates/app/src/node/behaviour.rs

@@ -0,0 +1,166 @@
+//! Composed P2P [`NetworkBehaviour`] for the core duty workflow.
+//!
+//! This is the Rust analog of Charon's `wireP2P`: it composes the relay
+//! transport behaviours (relay client, [`RelayManager`], force-direct) with
+//! the three core protocol behaviours (partial-signature exchange, QBFT
+//! consensus, peerinfo) into a single libp2p behaviour and builds a [`Node`]
+//! driving it. Relay reservation + routing is the cluster's only
+//! peer-discovery path (lock ENRs carry no addresses), mirroring Charon's
+//! `NewRelays` / `NewRelayReserver` / `NewRelayRouter` /
+//! `ForceDirectConnections` (`app.go:341-400`).
+//!
+//! Routing is push-based inside the individual behaviours (the QBFT p2p
+//! [`Handler`](pluto_consensus::qbft::p2p) holds an `Arc<Consensus>` and calls
+//! `consensus.handle()`; parsigex's [`Handle::subscribe`] dispatches inbound
+//! partial signatures), so the swarm drive loop body can be empty for
+//! correctness — see [`crate::node::drive_network`].
+
+use std::{collections::HashMap, sync::Arc};
+
+use libp2p::{relay, swarm::NetworkBehaviour};
+use pluto_consensus::qbft;
+use pluto_core::{gater::DutyGaterFn, types::PubKey};
+use pluto_crypto::types::PublicKey;
+use pluto_eth2api::EthBeaconNodeApiClient;
+use pluto_p2p::{
+    bootnode,
+    force_direct::ForceDirectBehaviour,
+    gater,
+    p2p::{Node, NodeType},
+    p2p_context::P2PContext,
+    peer::{self, Peer},
+    relay::RelayManager,
+};
+use pluto_parsigex as parsigex;
+use pluto_peerinfo::{self as peerinfo, LocalPeerInfo};
+use tokio_util::sync::CancellationToken;
+
+use crate::node::AppError;
+
+/// Composed network behaviour for the core duty workflow.
+#[derive(NetworkBehaviour)]
+pub(crate) struct CoreBehaviour {
+    /// Relay client transport (circuit reservations and relayed dials).
+    pub relay: relay::client::Behaviour,
+    /// Relay reservation lifecycle + relay-circuit peer routing (Charon's
+    /// `NewRelayReserver` + `NewRelayRouter`).
+    pub relay_manager: RelayManager,
+    /// Upgrades relay-routed connections to direct ones (Charon's
+    /// `ForceDirectConnections`).
+    pub force_direct: ForceDirectBehaviour,
+    /// Partial signature exchange between cluster peers.
+    pub parsigex: parsigex::Behaviour,
+    /// QBFT consensus message transport.
+    pub consensus: qbft::p2p::Behaviour,
+    /// Peer metadata exchange.
+    pub peerinfo: peerinfo::Behaviour,
+}
+
+/// Async handles for driving the composed behaviour from the core workflow.
+pub struct CoreHandles {
+    /// Outbound partial-signature broadcast + inbound subscription handle.
+    pub parsigex: parsigex::Handle,
+    /// Outbound QBFT broadcast handle.
+    pub consensus: qbft::p2p::Handle,
+}
+
+/// Composes the core behaviours and builds the libp2p [`Node`].
+// TODO(#402 part B): QUIC transport (featureset-gated off at v1.7.1) and
+// bandwidth metrics.
+#[allow(
+    clippy::too_many_arguments,
+    reason = "wireP2P aggregates independent inputs; a config struct is deferred to part B when priority inputs are added"
+)]
+pub(crate) async fn wire_p2p(
+    key: k256::SecretKey,
+    p2p_config: pluto_p2p::config::P2PConfig,
+    peers: Vec<Peer>,
+    consensus: Arc<qbft::Consensus>,
+    duty_gater: DutyGaterFn,
+    eth2_cl: EthBeaconNodeApiClient,
+    pub_shares_by_key: HashMap<PubKey, HashMap<u64, PublicKey>>,
+    lock_hash: Vec<u8>,
+    builder_enabled: bool,
+    nickname: String,
+    cancellation: CancellationToken,
+) -> Result<(Node<CoreBehaviour>, CoreHandles), AppError> {
+    let peer_ids = peers.iter().map(|peer| peer.id).collect::<Vec<_>>();
+    let local_peer_id = peer::peer_id_from_key(key.public_key())?;
+
+    // Relay endpoints resolve in the background; the `Charon-Cluster` header
+    // carries the lock hash hex (Charon v1.7.1 `p2p.NewRelays`). Post-#4130
+    // Charon additionally sends a `Cluster-Uuid` header for relay-side load
+    // balancing — a pending follow-up in pluto-p2p's `new_relays`.
+    let relay_addrs = bootnode::relay_addrs_for_resolution(&p2p_config.relays);
+    let relays =
+        bootnode::new_relays(cancellation.clone(), &relay_addrs, &hex::encode(&lock_hash)).await?;
+
+    // Closed gater: only cluster peers and the resolved relays may connect.
+    let conn_gater = gater::ConnGater::new_conn_gater(peer_ids.clone(), relays.clone());
+
+    let p2p_context = P2PContext::new(peer_ids.clone());
+    p2p_context.set_local_peer_id(local_peer_id);
+
+    // Keeps one circuit reservation alive per relay and continuously routes
+    // cluster peers via relay circuit addresses; force-direct then upgrades
+    // relayed connections to direct ones.
+    let relay_manager = RelayManager::new(relays, p2p_context.clone());
+    let force_direct = ForceDirectBehaviour::new(p2p_context.clone(), local_peer_id);
+
+    // Partial signature exchange. Inbound partial signatures are verified
+    // against the sender's public share for the duty via the eth2 verifier
+    // (Charon `parsigex.NewEth2Verifier`).
+    let parsigex_config = parsigex::Config::new(
+        local_peer_id,
+        p2p_context.clone(),
+        parsigex::new_eth2_verifier(eth2_cl, pub_shares_by_key),
+        duty_gater,
+    );
+    let (parsigex_comp, parsigex_handle) = parsigex::Behaviour::new(parsigex_config);
+
+    // QBFT consensus transport. `Behaviour::new` errors if the local peer id is
+    // not present in the configured cluster peer list.
+    let (consensus_comp, consensus_handle) = qbft::p2p::Behaviour::new(qbft::p2p::Config {
+        consensus,
+        p2p_context: p2p_context.clone(),
+        local_peer_id,
+        cancellation,
+    })?;
+
+    // Peer metadata exchange.
+    let (git_hash, _) = pluto_core::version::git_commit();
+    let peerinfo_config = peerinfo::Config::new(LocalPeerInfo::new(
+        pluto_core::version::VERSION.to_string(),
+        lock_hash,
+        git_hash,
+        builder_enabled,
+        nickname,
+    ))
+    .with_peers(peer_ids.clone());
+    let peerinfo_comp = peerinfo::Behaviour::new(local_peer_id, peerinfo_config);
+
+    let node = Node::new(
+        p2p_config,
+        key,
+        NodeType::TCP,
+        false,
+        p2p_context,
+        |builder, _keypair, relay_client| {
+            builder.with_gater(conn_gater).with_inner(CoreBehaviour {
+                relay: relay_client,
+                relay_manager,
+                force_direct,
+                parsigex: parsigex_comp,
+                consensus: consensus_comp,
+                peerinfo: peerinfo_comp,
+            })
+        },
+    )?;
+
+    let handles = CoreHandles {
+        parsigex: parsigex_handle,
+        consensus: consensus_handle,
+    };
+
+    Ok((node, handles))
+}

crates/app/src/node/config.rs

@@ -0,0 +1,74 @@
+//! Configuration for the distributed-validator node.
+
+use std::{net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
+
+use pluto_featureset::FeatureSet;
+use pluto_p2p::config::P2PConfig;
+
+/// Application configuration for running a distributed-validator node.
+///
+/// This is the Rust analog of Charon's `app.Config` (`app/app.go`), reduced to
+/// the minimal set required to wire and run the core duty workflow.
+/// Observability (monitoring/debug API, tracing/OTLP) and simnet/mock-only
+/// fields are intentionally omitted for the minimal-runnable wiring.
+// TODO(#402 part B): add monitoring/debug addrs, OTLP/Jaeger tracing config,
+// simnet (beacon/validator mock) and `TestConfig`-style overrides.
+#[derive(Debug, Clone)]
+pub struct AppConfig {
+    /// P2P networking configuration (listen/advertise addresses, relays, ...).
+    pub p2p: P2PConfig,
+
+    /// Path to the cluster lock file (`cluster-lock.json`).
+    pub lock_file: PathBuf,
+
+    /// Path to the node's secp256k1 P2P private key.
+    pub priv_key_file: PathBuf,
+
+    /// Enable private-key file locking. When set, a
+    /// `<priv_key_file>.lock`sentinel is maintained for the node's lifetime to
+    /// detect a second node started against the same key.
+    pub priv_key_locking: bool,
+
+    /// Beacon node API endpoints. The first reachable endpoint is used;
+    /// multiple addresses enable fallback.
+    pub beacon_node_addrs: Vec<String>,
+
+    /// Timeout for general beacon node requests.
+    pub beacon_node_timeout: Duration,
+
+    /// Timeout for beacon node submission (broadcast) requests.
+    pub beacon_node_submit_timeout: Duration,
+
+    /// Address the validator API HTTP server binds to.
+    pub validator_api_addr: SocketAddr,
+
+    /// Whether the builder API (MEV-boost) is enabled.
+    pub builder_api: bool,
+
+    /// Human-readable node nickname, surfaced via the peerinfo protocol.
+    pub nickname: String,
+
+    /// Skip cluster lock hash + signature verification.
+    pub no_verify: bool,
+
+    /// Execution-layer (eth1) JSON-RPC endpoint, used to verify operator
+    /// signatures (including EIP-1271 smart-contract signatures) in the cluster
+    /// lock. When `None`, lock verification runs without eth1 and such operator
+    /// signatures are not checked. Mirrors Charon's
+    /// `--execution-client-rpc-endpoint`.
+    pub eth1_endpoint: Option<String>,
+
+    /// Graffiti included in proposed blocks. `None` gives every validator the
+    /// default (client) graffiti; a single value applies to all validators; one
+    /// value per validator otherwise. Mirrors Charon's `--graffiti`.
+    pub graffiti: Option<Vec<String>>,
+
+    /// Disable appending the client version/codex to graffiti. Mirrors Charon's
+    /// `--graffiti-disable-client-append`.
+    pub graffiti_disable_client_append: bool,
+
+    /// Feature set controlling optional/alpha behaviors (e.g.
+    /// `FetchOnlyCommIdx0`, `ChainSplitHalt`). Resolved from the CLI
+    /// feature flags (out of scope here).
+    pub feature_set: Arc<FeatureSet>,
+}