diff --git a/Cargo.lock b/Cargo.lock
index a69d5dd2..65fbefe8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5133,15 +5133,21 @@ dependencies = [
  "libp2p",
  "pluto-build-proto",
  "pluto-cluster",
+ "pluto-consensus",
  "pluto-core",
  "pluto-crypto",
+ "pluto-eth1wrap",
  "pluto-eth2api",
+ "pluto-featureset",
  "pluto-k1util",
  "pluto-p2p",
+ "pluto-parsigex",
+ "pluto-peerinfo",
  "pluto-ssz",
  "pluto-testutil",
  "prost 0.14.4",
  "prost-types 0.14.4",
+ "rand 0.8.6",
  "regex",
  "reqwest 0.13.4",
  "serde",
@@ -5188,6 +5194,7 @@ dependencies = [
  "pluto-dkg",
  "pluto-eth1wrap",
  "pluto-eth2util",
+ "pluto-featureset",
  "pluto-k1util",
  "pluto-p2p",
  "pluto-relay-server",
diff --git a/crates/app/Cargo.toml b/crates/app/Cargo.toml
index f1b5c920..5e03507d 100644
--- a/crates/app/Cargo.toml
+++ b/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
diff --git a/crates/app/src/lib.rs b/crates/app/src/lib.rs
index 70ee9c25..e44805f9 100644
--- a/crates/app/src/lib.rs
+++ b/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;
diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs
new file mode 100644
index 00000000..2a53d9af
--- /dev/null
+++ b/crates/app/src/node/behaviour.rs
@@ -0,0 +1,168 @@
+//! 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`.
+//!
+//! 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())?;
+
+    // TODO: also send the post-#4130 `Cluster-Uuid` header (relay-side load
+    // balancing), pending 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,
+        &crate::utils::hex_7(&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))
+}
diff --git a/crates/app/src/node/config.rs b/crates/app/src/node/config.rs
new file mode 100644
index 00000000..5c9a9994
--- /dev/null
+++ b/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>,
+}
diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs
new file mode 100644
index 00000000..2750a2d4
--- /dev/null
+++ b/crates/app/src/node/mod.rs
@@ -0,0 +1,721 @@
+//! Distributed-validator node wiring.
+//!
+//! This module is the Rust analog of Charon's `app/app.go`: it constructs every
+//! core duty-workflow component, connects them together (the analog of Charon's
+//! `core.Wire`), composes the P2P behaviours, and runs the node until
+//! cancelled.
+//!
+//! The work is split to mirror Charon's `Run` (loads config/lock/keys, sets up
+//! P2P) vs `wireCoreWorkflow` (constructs and wires the components):
+//!
+//! * [`run`] loads the cluster lock, P2P key, and beacon clients, constructs
+//!   the consensus component + P2P behaviours, then calls
+//!   [`wire::wire_core_workflow`].
+//! * [`wire::wire_core_workflow`] takes already-resolved inputs and produces
+//!   the wired component graph (so it is unit-testable against a `BeaconMock`).
+//!
+//! Lifecycle is idiomatic tokio: long-lived tasks live in a [`JoinSet`], driven
+//! until the [`CancellationToken`] fires or the first task fails, after which
+//! an explicit ordered shutdown runs.
+
+pub mod behaviour;
+pub mod config;
+pub mod wire;
+
+pub use config::AppConfig;
+
+use std::{
+    collections::HashMap,
+    sync::{Arc, OnceLock},
+};
+
+use futures::StreamExt;
+use pluto_consensus::qbft;
+use pluto_core::gater::DutyGaterFn;
+use pluto_p2p::peer::Peer;
+use tokio::task::JoinSet;
+use tokio_util::sync::CancellationToken;
+
+use behaviour::{CoreBehaviour, CoreHandles};
+use wire::{ParSigExSeam, ValidatorInfo, WireInputs, WiredComponents};
+
+use crate::privkeylock;
+
+/// Errors raised while constructing or running a distributed-validator node.
+#[derive(Debug, thiserror::Error)]
+pub enum AppError {
+    /// Failed to load or verify the cluster lock.
+    #[error("cluster lock: {0}")]
+    LoadLock(#[from] pluto_cluster::load::LoadError),
+
+    /// Failed to derive cluster peers from the lock.
+    #[error("cluster definition: {0}")]
+    Definition(#[from] pluto_cluster::definition::DefinitionError),
+
+    /// Failed to read distributed-validator data from the lock.
+    #[error("distributed validator: {0}")]
+    DistValidator(#[from] pluto_cluster::distvalidator::DistValidatorError),
+
+    /// Execution-layer (eth1) client construction failed.
+    #[error("eth1 client: {0}")]
+    Eth1(#[source] pluto_eth1wrap::EthClientError),
+
+    /// A distributed validator's public key was not 48 bytes.
+    #[error("distributed validator pubkey is not 48 bytes")]
+    InvalidValidatorPubKey,
+
+    /// A distributed validator's fee-recipient address was missing or not a
+    /// valid 20-byte execution address.
+    #[error("distributed validator {index} has an invalid fee recipient address")]
+    InvalidFeeRecipient {
+        /// Index of the offending distributed validator.
+        index: usize,
+    },
+
+    /// Failed to load the P2P private key.
+    #[error("load p2p key: {0}")]
+    LoadKey(#[from] pluto_k1util::K1UtilError),
+
+    /// The local P2P key does not match any cluster operator.
+    #[error("local peer not found in cluster lock")]
+    LocalPeerNotFound,
+
+    /// Private-key lock acquisition/maintenance failed.
+    #[error("privkey lock: {0}")]
+    PrivKeyLock(#[from] privkeylock::PrivKeyLockError),
+
+    /// P2P peer derivation/verification failed.
+    #[error("p2p peer: {0}")]
+    Peer(#[from] pluto_p2p::peer::PeerError),
+
+    /// P2P node construction failed.
+    #[error("p2p node: {0}")]
+    P2P(#[from] pluto_p2p::p2p::P2PError),
+
+    /// Relay endpoint resolution failed.
+    #[error("relays: {0}")]
+    Relays(#[from] pluto_p2p::bootnode::BootnodeError),
+
+    /// QBFT consensus construction failed.
+    #[error("consensus: {0}")]
+    Consensus(#[from] qbft::Error),
+
+    /// QBFT p2p adapter construction failed.
+    #[error("consensus p2p: {0}")]
+    ConsensusP2P(#[from] qbft::p2p::Error),
+
+    /// A beacon node API request failed.
+    #[error("beacon node api: {0}")]
+    BeaconApi(#[from] pluto_eth2api::EthBeaconNodeApiClientError),
+
+    /// The beacon node URL could not be parsed.
+    #[error("invalid beacon node url: {0}")]
+    BeaconUrl(#[from] url::ParseError),
+
+    /// Beacon node client construction failed.
+    #[error("beacon client: {0}")]
+    BeaconClient(#[source] anyhow::Error),
+
+    /// Duty gater construction failed.
+    #[error("duty gater: {0}")]
+    Gater(#[source] pluto_core::gater::GaterError),
+
+    /// Deadline calculator construction failed.
+    #[error("deadline calculator: {0}")]
+    Deadline(#[source] pluto_core::deadline::DeadlineError),
+
+    /// Graffiti builder construction failed.
+    #[error("graffiti: {0}")]
+    Graffiti(#[source] pluto_core::fetcher::GraffitiError),
+
+    /// Signature aggregator construction failed.
+    #[error("sigagg: {0}")]
+    SigAgg(#[source] pluto_core::sigagg::SigAggError),
+
+    /// Broadcaster construction failed.
+    #[error("broadcaster: {0}")]
+    Broadcaster(#[source] pluto_core::bcast::Error),
+
+    /// Partial-signature exchange broadcast failed.
+    #[error("parsigex: {0}")]
+    ParSigEx(#[from] pluto_parsigex::Error),
+
+    /// Scheduler construction failed.
+    #[error("scheduler: {0}")]
+    Scheduler(#[source] pluto_core::scheduler::SchedulerError),
+
+    /// Validator API server failed.
+    #[error("validator api: {0}")]
+    ValidatorApi(#[source] std::io::Error),
+}
+
+/// A wired, runnable distributed-validator node.
+///
+/// Construct with [`App::new`] (loads lock/keys, builds and wires every
+/// component) and drive with [`App::run`].
+pub struct App {
+    config: AppConfig,
+}
+
+impl App {
+    /// Creates a new application from its configuration.
+    pub fn new(config: AppConfig) -> Self {
+        Self { config }
+    }
+
+    /// Loads cluster state, wires the core workflow, and runs the node until
+    /// `ct` is cancelled or a long-lived task fails.
+    ///
+    /// This is the Rust analog of Charon's `app.Run`.
+    pub async fn run(self, ct: CancellationToken) -> Result<(), AppError> {
+        run(self.config, ct).await
+    }
+}
+
+/// Loads the cluster lock + key, builds the consensus component and P2P
+/// behaviours, wires the core workflow, and drives the node.
+async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
+    // ---- (1) Load cluster lock + key, derive peers and this node's index ----
+    //
+    // Operator-signature verification uses the configured execution-layer
+    // client; without an `eth1_endpoint`, a no-op client is used
+    // (BLS-aggregate and node signatures are still verified, but EIP-1271
+    // smart-contract operator signatures are not). With `no_verify`,
+    // verification still runs but failures are downgraded to warnings.
+    let eth1 = pluto_eth1wrap::EthClient::new(config.eth1_endpoint.as_deref().unwrap_or_default())
+        .await
+        .map_err(AppError::Eth1)?;
+    let lock =
+        pluto_cluster::load::load_cluster_lock(&config.lock_file, config.no_verify, &eth1).await?;
+    let threshold = lock.threshold;
+    // TODO(#402 part B): honor `lock.target_gas_limit` once
+    // `validatorapi::Component::new` accepts a target-gas-limit parameter —
+    // Charon passes the lock value to `NewComponent` (app.go:563), but Pluto's
+    // validator-registration path has no such input yet.
+    let _ = lock.target_gas_limit;
+
+    let key = pluto_k1util::load(&config.priv_key_file)?;
+
+    // Guards against a second node running against the same key.
+    let priv_key_lock: Option<Arc<privkeylock::Service>> = if config.priv_key_locking {
+        Some(Arc::new(
+            privkeylock::Service::new(
+                format!("{}.lock", config.priv_key_file.display()),
+                "pluto run",
+            )
+            .await?,
+        ))
+    } else {
+        None
+    };
+
+    let peers = lock.peers()?;
+    pluto_p2p::peer::verify_p2p_key(&peers, &key)?;
+
+    let local_peer_id = pluto_p2p::peer::peer_id_from_key(key.public_key())?;
+    let local_node = peers
+        .iter()
+        .find(|p| p.id == local_peer_id)
+        .ok_or(AppError::LocalPeerNotFound)?;
+    let local_idx = local_node.index;
+    let share_idx = local_node.share_idx();
+
+    // qbft peers (secp256k1 pubkeys, in process-index order).
+    let qbft_peers = build_qbft_peers(&peers)?;
+
+    // Per-validator data for this node (mirrors app.go:415-452).
+    let validators = build_validators(&lock, share_idx)?;
+
+    // ---- (2/3) eth2 clients ----
+    //
+    // TODO(#402 part B): support multiple `beacon_node_addrs` with per-request
+    // fallback — `EthBeaconNodeApiClient` is single-endpoint, so a failover
+    // (multi-endpoint) client in pluto-eth2api is required first; for now the
+    // first address is used.
+    let beacon_node_addr = config
+        .beacon_node_addrs
+        .first()
+        .map(String::as_str)
+        .unwrap_or_default();
+    let eth2_cl = build_api_client(beacon_node_addr, config.beacon_node_timeout)?;
+    let beacon_client = pluto_eth2api::BeaconNodeClient::new(eth2_cl.clone());
+    // Broadcasting uses a separate client with the (distinct) submit timeout.
+    let submission_api = build_api_client(beacon_node_addr, config.beacon_node_submit_timeout)?;
+    let submission_client = pluto_eth2api::BeaconNodeClient::new(submission_api);
+
+    // ---- Beacon-derived duty-workflow inputs (Charon app.go:540-556) ----
+
+    // Duty admission gate: validates duties against the beacon chain
+    // (Charon `core.NewDutyGater`).
+    let duty_gater: DutyGaterFn = pluto_core::gater::DutyGater::new(&eth2_cl)
+        .await
+        .map_err(AppError::Gater)?
+        .into_fn();
+
+    // Per-component deadline calculator, shared as an `Arc<dyn ...>` so a single
+    // beacon-derived instance backs every component's deadliner.
+    let deadline_calc: Arc<dyn pluto_core::deadline::DeadlineCalculator> = Arc::new(
+        pluto_core::deadline::DutyDeadlineCalculator::from_client(&eth2_cl)
+            .await
+            .map_err(AppError::Deadline)?,
+    );
+
+    // Per-validator graffiti for proposed blocks.
+    let graffiti_pubkeys: Vec<pluto_core::types::PubKey> =
+        validators.iter().map(|v| v.pubkey).collect();
+    let graffiti_builder = pluto_core::fetcher::GraffitiBuilder::new(
+        &graffiti_pubkeys,
+        config.graffiti.as_deref(),
+        config.graffiti_disable_client_append,
+        &eth2_cl,
+    )
+    .await
+    .map_err(AppError::Graffiti)?;
+
+    // Electra activation slot = electra_fork_epoch * slots_per_epoch.
+    let (_, slots_per_epoch) = eth2_cl.fetch_slots_config().await?;
+    let fork_config = eth2_cl.fetch_fork_config().await?;
+    let electra_slot = fork_config
+        .get(&pluto_eth2api::ConsensusVersion::Electra)
+        .map(|schedule| schedule.epoch)
+        .unwrap_or(0)
+        .saturating_mul(slots_per_epoch);
+
+    // Feature set drives optional/alpha behaviors.
+    let feature_set = Arc::clone(&config.feature_set);
+    let fetch_only_comm_idx0 = feature_set.enabled(pluto_featureset::Feature::FetchOnlyCommIdx0);
+
+    // ---- Consensus (built directly; shared with p2p behaviour + core stitch) ----
+    //
+    // TODO(#402 part B): wrap in ConsensusController for dynamic protocol
+    // switching (priority/infosync).
+    //
+    // Resolve the broadcaster<->behaviour construction cycle with the
+    // `Arc<OnceLock<Handle>>` pattern (see qbft::p2p `build_consensus_nodes`).
+    let (cons_deadliner, cons_expired_rx) = pluto_core::deadline::DeadlinerTask::start(
+        ct.clone(),
+        "consensus.qbft",
+        Arc::clone(&deadline_calc),
+    );
+
+    // TODO:
+    // The `Arc<OnceLock<Handle>>` pattern is a bit awkward; explore alternatives
+    let handle_slot = Arc::new(OnceLock::<qbft::p2p::Handle>::new());
+    let broadcaster: qbft::Broadcaster = {
+        let handle_slot = Arc::clone(&handle_slot);
+        Arc::new(move |_ct, msg| {
+            let handle_slot = Arc::clone(&handle_slot);
+            Box::pin(async move {
+                let handle = handle_slot
+                    .get()
+                    .expect("qbft p2p handle initialized before broadcast")
+                    .clone();
+                handle.broadcast(msg).await
+            })
+        })
+    };
+
+    let consensus = Arc::new(qbft::Consensus::new(qbft::Config {
+        peers: qbft_peers,
+        local_peer_idx: i64::try_from(local_idx).map_err(|_| AppError::LocalPeerNotFound)?,
+        privkey: key.clone(),
+        deadliner: cons_deadliner,
+        expired_rx: cons_expired_rx,
+        duty_gater: Arc::clone(&duty_gater),
+        broadcaster,
+        sniffer: Arc::new(|_| {}),
+        // Charon gates duplicate-attestation comparison on the alpha
+        // `ChainSplitHalt` featureset flag (off by default).
+        compare_attestations: feature_set.enabled(pluto_featureset::Feature::ChainSplitHalt),
+        feature_set: Arc::clone(&feature_set),
+        timer_func: pluto_consensus::timer::get_round_timer_func(Arc::clone(&feature_set)),
+    })?);
+
+    // Full public-share map (DV root pubkey -> share index -> public share),
+    // used by the parsigex verifier to check each peer's partial signature.
+    let pub_shares_by_key = build_pub_shares_by_key(&lock)?;
+
+    // ---- P2P behaviours (relay + parsigex + qbft + peerinfo) ----
+    let (node, handles) = behaviour::wire_p2p(
+        key.clone(),
+        config.p2p.clone(),
+        peers,
+        Arc::clone(&consensus),
+        Arc::clone(&duty_gater),
+        eth2_cl.clone(),
+        pub_shares_by_key,
+        lock.lock_hash.clone(),
+        config.builder_api,
+        config.nickname.clone(),
+        ct.clone(),
+    )
+    .await?;
+    // Complete the broadcaster<->behaviour cycle.
+    handle_slot
+        .set(handles.consensus.clone())
+        .map_err(|_| AppError::ConsensusP2P(qbft::p2p::Error::BehaviourClosed))?;
+
+    // ---- Wire the core workflow ----
+    let upstream_url = reqwest::Url::parse(beacon_node_addr)?;
+
+    let parsigex_seam = production_parsigex_seam(&handles);
+
+    // Aggregated-signature verifier: verifies the reconstructed group signature
+    // against the beacon-node signing domain.
+    let sigagg_verifier = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone()));
+
+    let wired = wire::wire_core_workflow(
+        WireInputs {
+            threshold,
+            share_idx,
+            beacon_client,
+            eth2_cl,
+            submission_client,
+            validators,
+            consensus: Arc::clone(&consensus),
+            builder_enabled: config.builder_api,
+            upstream_url,
+            parsigex: parsigex_seam,
+            sigagg_verifier,
+            deadline_calc,
+            graffiti_builder,
+            electra_slot,
+            fetch_only_comm_idx0,
+        },
+        ct.clone(),
+    )
+    .await?;
+
+    // ---- Lifecycle: spawn long-lived tasks ----
+    run_lifecycle(
+        node,
+        consensus,
+        handles,
+        wired,
+        priv_key_lock,
+        config.validator_api_addr,
+        ct,
+    )
+    .await
+}
+
+/// Builds the production parsigex seam from the real `parsigex::Handle`.
+fn production_parsigex_seam(handles: &CoreHandles) -> ParSigExSeam {
+    let broadcast_handle = handles.parsigex.clone();
+    let subscribe_handle = handles.parsigex.clone();
+    ParSigExSeam {
+        broadcast: Arc::new(move |duty, set| {
+            let handle = broadcast_handle.clone();
+            Box::pin(async move {
+                handle
+                    .broadcast(duty, set)
+                    .await
+                    .map(|_| ())
+                    .map_err(AppError::ParSigEx)
+            })
+        }),
+        subscribe: Box::new(move |received| {
+            Box::pin(async move {
+                let sub = pluto_parsigex::received_subscriber(move |duty, set| {
+                    let received = Arc::clone(&received);
+                    async move {
+                        received(duty, set).await;
+                    }
+                });
+                subscribe_handle.subscribe(sub).await;
+            })
+        }),
+    }
+}
+
+/// Spawns and supervises the node's long-lived tasks, then performs an ordered
+/// shutdown on cancellation or first-task failure.
+async fn run_lifecycle(
+    node: pluto_p2p::p2p::Node<CoreBehaviour>,
+    consensus: Arc<qbft::Consensus>,
+    _handles: CoreHandles,
+    wired: WiredComponents,
+    priv_key_lock: Option<Arc<privkeylock::Service>>,
+    validator_api_addr: std::net::SocketAddr,
+    ct: CancellationToken,
+) -> Result<(), AppError> {
+    let WiredComponents {
+        scheduler_task,
+        dutydb,
+        parsigdb,
+        parsigdb_deadliner_rx,
+        aggsigdb: _aggsigdb,
+        fetcher: _fetcher,
+        validator_api_router,
+    } = wired;
+
+    // Self-spawning actor: consensus expired-duty pruner.
+    let _consensus_task = consensus.start(ct.clone());
+
+    let mut tasks: JoinSet<Result<(), AppError>> = JoinSet::new();
+
+    // Supervise the self-spawning scheduler actor alongside the other
+    // long-lived tasks so its exit triggers node shutdown (Charon registers
+    // `sched.Run` as a lifecycle hook, `app.go:660`). The actor returns `()`
+    // and only exits on cancellation.
+    tasks.extend([async move {
+        let _ = scheduler_task.await;
+        Ok::<(), AppError>(())
+    }]);
+
+    // Swarm drive loop (push-based routing inside behaviours).
+    {
+        let ct = ct.clone();
+        tasks.spawn(async move {
+            drive_network(node, ct).await;
+            Ok(())
+        });
+    }
+
+    // ParSigDB trim task.
+    {
+        let parsigdb = Arc::clone(&parsigdb);
+        tasks.spawn(async move {
+            parsigdb.trim(parsigdb_deadliner_rx).await;
+            Ok(())
+        });
+    }
+
+    // Private-key lock maintenance loop. Only spawn `run` when locking is
+    // enabled — `Service::close()` blocks forever unless `run` was called, so
+    // spawn and close are guarded by the same `Option`.
+    if let Some(svc) = &priv_key_lock {
+        let svc = Arc::clone(svc);
+        // Propagate a lock-maintenance failure so it fails the run, matching
+        // Charon which registers lockSvc.Run as a lifecycle hook whose error
+        // triggers shutdown (app.go:166). A graceful `close()` returns `Ok`.
+        tasks.spawn(async move { svc.run().await.map_err(AppError::PrivKeyLock) });
+    }
+
+    // Validator API axum server.
+    tasks.spawn(serve_validator_api(
+        validator_api_addr,
+        validator_api_router,
+        ct.clone(),
+    ));
+
+    // Supervise: stop on cancellation or first task completion. A failed task
+    // fails the whole run (Charon lifecycle: any start-hook error triggers
+    // shutdown and is returned from `Run`).
+    let mut task_err: Option<AppError> = None;
+    tokio::select! {
+        () = ct.cancelled() => {
+            tracing::info!("node: cancellation requested");
+        }
+        joined = tasks.join_next() => {
+            match joined {
+                Some(Ok(Err(err))) => {
+                    tracing::error!(%err, "node: a long-lived task failed; shutting down");
+                    task_err = Some(err);
+                }
+                _ => tracing::warn!("node: a long-lived task exited; shutting down"),
+            }
+            ct.cancel();
+        }
+    }
+
+    // ---- Ordered shutdown ----
+    // Close the private-key lock (signals its `run` loop to delete the sentinel
+    // file and exit, so the drain below can join it cleanly). Only ever called
+    // when `run` was spawned above — otherwise `close()` would block forever.
+    if let Some(svc) = &priv_key_lock {
+        svc.close().await;
+    }
+
+    // Brief drain for in-flight tasks.
+    let _ = tokio::time::timeout(std::time::Duration::from_secs(2), async {
+        while tasks.join_next().await.is_some() {}
+    })
+    .await;
+    tasks.shutdown().await;
+
+    // Stop dutydb (cancels its child token).
+    dutydb.shutdown();
+
+    // Fail the run with the first task error (Charon: `lifecycle.Manager.Run`
+    // returns the first start-hook error).
+    task_err.map_or(Ok(()), Err)
+}
+
+/// Serves the validator API until `ct` fires (graceful shutdown, `Ok`). A bind
+/// or serve failure is returned and fails the run, matching Charon's
+/// `httpServeHook`, which swallows only `http.ErrServerClosed`.
+async fn serve_validator_api(
+    addr: std::net::SocketAddr,
+    router: axum::Router,
+    ct: CancellationToken,
+) -> Result<(), AppError> {
+    let listener = tokio::net::TcpListener::bind(addr)
+        .await
+        .map_err(AppError::ValidatorApi)?;
+
+    axum::serve(listener, router)
+        .with_graceful_shutdown(async move { ct.cancelled().await })
+        .await
+        .map_err(AppError::ValidatorApi)
+}
+
+/// Drives the libp2p swarm. Routing is push-based inside the behaviours, so the
+/// loop body only needs to keep polling the swarm.
+async fn drive_network(mut node: pluto_p2p::p2p::Node<CoreBehaviour>, ct: CancellationToken) {
+    loop {
+        tokio::select! {
+            () = ct.cancelled() => break,
+            _event = node.select_next_some() => {
+                // TODO(#402 part B): optional logging of
+                // PlutoBehaviourEvent::Inner(CoreBehaviourEvent::...) events.
+            }
+        }
+    }
+}
+
+/// Builds the QBFT peer list (secp256k1 pubkeys) from the cluster peers.
+fn build_qbft_peers(peers: &[Peer]) -> Result<Vec<qbft::Peer>, AppError> {
+    peers
+        .iter()
+        .map(|peer| {
+            Ok(qbft::Peer {
+                index: i64::try_from(peer.index).map_err(|_| AppError::LocalPeerNotFound)?,
+                name: peer.name.clone(),
+                public_key: peer.public_key()?,
+            })
+        })
+        .collect()
+}
+
+/// Extracts this node's per-validator data from the cluster lock.
+fn build_validators(
+    lock: &pluto_cluster::lock::Lock,
+    share_idx: u64,
+) -> Result<Vec<ValidatorInfo>, AppError> {
+    let fee_recipients = lock.fee_recipient_addresses();
+    let mut out = Vec::with_capacity(lock.distributed_validators.len());
+    for (i, dv) in lock.distributed_validators.iter().enumerate() {
+        let pubkey_bytes: [u8; 48] = dv
+            .pub_key
+            .clone()
+            .try_into()
+            .map_err(|_| AppError::InvalidValidatorPubKey)?;
+        let pubkey = pluto_core::types::PubKey::new(pubkey_bytes);
+        let eth2_pubkey: pluto_eth2api::spec::phase0::BLSPubKey = pubkey_bytes;
+
+        // share_idx is 1-indexed; pub_shares is 0-indexed.
+        let share_pos = usize::try_from(share_idx)
+            .map_err(|_| AppError::LocalPeerNotFound)?
+            .saturating_sub(1);
+        let pubshare: [u8; 48] = dv.public_share(share_pos)?;
+
+        // A missing or malformed fee recipient is a misconfiguration, not a
+        // value to paper over: the zero address is semantically invalid, so
+        // fail loudly instead of defaulting (Charon errors when it hex-decodes
+        // the address, app.go:1178).
+        let fee_recipient = fee_recipients
+            .get(i)
+            .and_then(|s| parse_execution_address(s))
+            .ok_or(AppError::InvalidFeeRecipient { index: i })?;
+
+        out.push(ValidatorInfo {
+            pubkey,
+            eth2_pubkey,
+            pubshare,
+            fee_recipient,
+        });
+    }
+    Ok(out)
+}
+
+/// Builds the full public-share map for the cluster: for every distributed
+/// validator, its group (root) public key mapped to every operator's public
+/// share, keyed by 1-indexed share index.
+///
+/// The parsigex verifier uses this to check each inbound partial signature
+/// against the sender's public share (Charon's `pubSharesByKey`).
+fn build_pub_shares_by_key(
+    lock: &pluto_cluster::lock::Lock,
+) -> Result<
+    HashMap<pluto_core::types::PubKey, HashMap<u64, pluto_crypto::types::PublicKey>>,
+    AppError,
+> {
+    let mut out = HashMap::with_capacity(lock.distributed_validators.len());
+    for dv in &lock.distributed_validators {
+        let pubkey_bytes: [u8; 48] = dv
+            .pub_key
+            .clone()
+            .try_into()
+            .map_err(|_| AppError::InvalidValidatorPubKey)?;
+        let pubkey = pluto_core::types::PubKey::new(pubkey_bytes);
+
+        let mut shares = HashMap::with_capacity(dv.pub_shares.len());
+        for pos in 0..dv.pub_shares.len() {
+            // Share indices are 1-based (matching `ParSignedData::share_idx`);
+            // `public_share` takes the 0-based position.
+            let share_idx = (pos as u64).saturating_add(1);
+            shares.insert(share_idx, dv.public_share(pos)?);
+        }
+        out.insert(pubkey, shares);
+    }
+    Ok(out)
+}
+
+/// Parses a `0x`-prefixed hex execution address.
+fn parse_execution_address(s: &str) -> Option<[u8; 20]> {
+    let s = s.strip_prefix("0x").unwrap_or(s);
+    let bytes = hex::decode(s).ok()?;
+    bytes.try_into().ok()
+}
+
+/// Builds an [`EthBeaconNodeApiClient`](pluto_eth2api::EthBeaconNodeApiClient)
+/// for `base_url` with the given request timeout.
+fn build_api_client(
+    base_url: &str,
+    timeout: std::time::Duration,
+) -> Result<pluto_eth2api::EthBeaconNodeApiClient, AppError> {
+    let http = reqwest::Client::builder()
+        .timeout(timeout)
+        .build()
+        .map_err(|e| AppError::BeaconClient(e.into()))?;
+    pluto_eth2api::EthBeaconNodeApiClient::with_client(base_url, http)
+        .map_err(AppError::BeaconClient)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // A failed validator-API bind must fail the run (Charon: a start-hook
+    // error is returned from `lifecycle.Manager.Run`), not just log.
+    #[tokio::test]
+    async fn serve_validator_api_fails_when_addr_in_use() {
+        let occupied = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind ephemeral port");
+        let addr = occupied.local_addr().expect("local addr");
+
+        let err = serve_validator_api(addr, axum::Router::new(), CancellationToken::new())
+            .await
+            .expect_err("bind on an occupied port should fail");
+
+        assert!(matches!(err, AppError::ValidatorApi(_)));
+    }
+
+    #[tokio::test]
+    async fn serve_validator_api_shuts_down_gracefully_on_cancel() {
+        let ct = CancellationToken::new();
+        ct.cancel();
+
+        // Graceful shutdown is a clean exit, mirroring Charon's swallowing of
+        // `http.ErrServerClosed`.
+        serve_validator_api(
+            "127.0.0.1:0".parse().expect("addr"),
+            axum::Router::new(),
+            ct,
+        )
+        .await
+        .expect("cancelled serve should exit cleanly");
+    }
+}
diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs
new file mode 100644
index 00000000..1f749919
--- /dev/null
+++ b/crates/app/src/node/wire.rs
@@ -0,0 +1,606 @@
+//! Core duty-workflow construction and wiring.
+//!
+//! This is the Rust analog of Charon's `wireCoreWorkflow` (`app/app.go:399`)
+//! and `core.Wire` (`core/interfaces.go:283`). It constructs the ten core duty
+//! workflow components and connects them into the data-flow graph that drives a
+//! single distributed-validator node.
+//!
+//! The construction order builds back-edge targets before their consumers (see
+//! [`wire_core_workflow`]), and — unlike Charon's split between a `wireFuncs`
+//! struct and the `Wire` call — Pluto interleaves construction and wiring
+//! because of its builder/service split:
+//!
+//! * scheduler subscribers are registered on the *builder* before `.build()`;
+//! * the fetcher's back-edges and subscribers are bon-builder fields;
+//! * consensus / parsigdb / sigagg subscribers are registered on the *service*.
+//!
+//! To mirror Charon's `Run` (loads) vs `wireCoreWorkflow` (wires) split — and
+//! Charon's `TestConfig.ParSigExFunc` injection — this function takes
+//! already-resolved inputs ([`WireInputs`]) and a [`ParSigExSeam`] for the
+//! partial-signature exchange, so tests can inject a `BeaconMock` and an
+//! in-memory (loopback) parsigex without a real libp2p swarm.
+
+use std::{collections::HashMap, sync::Arc};
+
+use futures::future::BoxFuture;
+use pluto_consensus::qbft;
+use pluto_core::{
+    aggsigdb::{memory::MemoryDBHandle, types::AggSigDB},
+    bcast::Broadcaster,
+    corepb::v1::core as pbcore,
+    deadline::{DeadlineCalculator, DeadlinerTask},
+    dutydb,
+    fetcher::{
+        AggSigDbFunc, AwaitAttDataFunc, FeeRecipientFunc, Fetcher, GraffitiBuilder, Subscriber,
+    },
+    parsigdb,
+    scheduler::SchedulerBuilder,
+    sigagg::{Aggregator, VerifyFn},
+    signeddata::{SyncContribution, VersionedAggregatedAttestation},
+    types::{Duty, ParSignedData, ParSignedDataSet, PubKey, SignedData, SignedDataSet},
+    unsigneddata::{self, UnsignedDataSet},
+    validatorapi::{self, Component, Handler},
+};
+use pluto_eth2api::{
+    BeaconNodeClient, EthBeaconNodeApiClient,
+    spec::{bellatrix::ExecutionAddress, phase0::BLSPubKey},
+    valcache::ValidatorCache,
+};
+use tokio_util::sync::CancellationToken;
+
+use crate::node::AppError;
+
+/// A `Send + Sync` boxed future. The parsigdb subscriber seams require their
+/// futures to be `Sync` (see `internal_subscriber`/`threshold_subscriber`), so
+/// the broadcast seam future must be `Sync` too.
+type SyncBoxFuture<'a, T> =
+    std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + Sync + 'a>>;
+
+/// Outbound partial-signature broadcast seam.
+pub type ParSigExBroadcast = Arc<
+    dyn Fn(Duty, ParSignedDataSet) -> SyncBoxFuture<'static, Result<(), AppError>> + Send + Sync,
+>;
+
+/// Inbound received-partial-signature subscriber, wired into
+/// `parsigdb.store_external`.
+pub type ParSigExReceived =
+    Arc<dyn Fn(Duty, ParSignedDataSet) -> BoxFuture<'static, ()> + Send + Sync>;
+
+/// Registers the inbound subscriber on the parsigex transport.
+pub type ParSigExSubscribe =
+    Box<dyn FnOnce(ParSigExReceived) -> BoxFuture<'static, ()> + Send + Sync>;
+
+/// Partial-signature exchange seam.
+///
+/// Mirrors Charon's `TestConfig.ParSigExFunc`: the production path supplies a
+/// real `parsigex::Handle` (outbound broadcast) plus an inbound subscription;
+/// tests supply a loopback so partial signatures cross the threshold locally.
+pub struct ParSigExSeam {
+    /// Outbound broadcast, wired into `parsigdb.subscribe_internal`.
+    pub broadcast: ParSigExBroadcast,
+    /// Registers the inbound subscriber (received sets -> `store_external`).
+    /// A loopback test may wire `broadcast` straight back into the subscriber.
+    pub subscribe: ParSigExSubscribe,
+}
+
+/// Per-validator data extracted from the cluster lock for this node.
+pub struct ValidatorInfo {
+    /// The distributed validator's group (root) public key.
+    pub pubkey: PubKey,
+    /// The DV root pubkey as an eth2 `BLSPubKey`.
+    pub eth2_pubkey: BLSPubKey,
+    /// This node's public share for the validator (eth2 `BLSPubKey`).
+    pub pubshare: BLSPubKey,
+    /// Fee recipient execution address for the validator.
+    pub fee_recipient: ExecutionAddress,
+}
+
+/// Already-resolved inputs to [`wire_core_workflow`].
+///
+/// These mirror what Charon's `wireCoreWorkflow` derives from the cluster
+/// manifest + config, but are passed in so that file-loading and P2P setup
+/// (in `run`) stay separate from construction-and-wiring (here) — enabling the
+/// Tier 1 test to inject a `BeaconMock` and in-memory parsigex.
+pub struct WireInputs {
+    /// Threshold of partial signatures required for aggregation.
+    pub threshold: u64,
+    /// This node's 1-indexed share index.
+    pub share_idx: u64,
+    /// Beacon node client used for scheduling.
+    pub beacon_client: BeaconNodeClient,
+    /// Beacon node API client used for fetching / dutydb / validatorapi.
+    pub eth2_cl: EthBeaconNodeApiClient,
+    /// Submission beacon node client used for broadcasting.
+    pub submission_client: BeaconNodeClient,
+    /// Per-validator data for this node.
+    pub validators: Vec<ValidatorInfo>,
+    /// Already-constructed consensus component, also wired into the QBFT p2p
+    /// behaviour by the caller.
+    pub consensus: Arc<qbft::Consensus>,
+    /// Whether the builder API is enabled.
+    pub builder_enabled: bool,
+    /// Upstream beacon URL the validator API reverse-proxies unhandled requests
+    /// to.
+    pub upstream_url: reqwest::Url,
+    /// Partial-signature exchange seam (production handle or test loopback).
+    pub parsigex: ParSigExSeam,
+    /// Aggregated-signature verifier for SigAgg. Production injects the eth2
+    /// verifier (`sigagg::new_verifier`); tests may inject a permissive one to
+    /// exercise the wiring without real BLS test vectors (mirrors Charon's
+    /// `TestConfig`).
+    pub sigagg_verifier: VerifyFn,
+    /// Per-component deadline calculator. Production injects a beacon-derived
+    /// [`DutyDeadlineCalculator`](pluto_core::deadline::DutyDeadlineCalculator);
+    /// tests inject `NeverExpiringCalculator` so driven duties are never
+    /// trimmed.
+    pub deadline_calc: Arc<dyn DeadlineCalculator>,
+    /// Per-validator graffiti builder for proposed blocks. Production injects a
+    /// beacon-derived builder; tests inject the default.
+    pub graffiti_builder: GraffitiBuilder,
+    /// Slot at which the Electra fork activates (`electra_epoch *
+    /// slots_per_epoch`); gates the `fetch_only_comm_idx0` committee-index
+    /// behavior.
+    pub electra_slot: u64,
+    /// Whether to fetch only committee index 0 at/after `electra_slot`
+    /// (`Feature::FetchOnlyCommIdx0`).
+    pub fetch_only_comm_idx0: bool,
+}
+
+/// The wired components and long-lived handles produced by
+/// [`wire_core_workflow`], returned so the caller (`run`) can drive their
+/// background tasks and shut them down in order.
+pub struct WiredComponents {
+    /// Background task driving the self-spawning scheduler actor, returned so
+    /// the caller can supervise its lifecycle (Charon registers `sched.Run`
+    /// as a lifecycle hook, `app.go:660`). The scheduler's query handle is held
+    /// directly by the validator API.
+    pub scheduler_task: tokio::task::JoinHandle<()>,
+    /// In-memory duty database (shared; needs explicit shutdown).
+    pub dutydb: Arc<dutydb::MemDB>,
+    /// In-memory partial-signature database (its `trim` task must be spawned).
+    pub parsigdb: Arc<parsigdb::memory::MemDB>,
+    /// Receiver paired with the parsigdb deadliner, for the `trim` task.
+    pub parsigdb_deadliner_rx: tokio::sync::mpsc::Receiver<Duty>,
+    /// Aggregated-signature database (self-spawning).
+    pub aggsigdb: MemoryDBHandle,
+    /// The fetcher (driven via scheduler subscriptions).
+    pub fetcher: Arc<Fetcher>,
+    /// The validator API axum router, ready to be served.
+    pub validator_api_router: axum::Router,
+}
+
+/// Constructs and wires the ten core duty-workflow components.
+///
+/// Reproduces the data-flow graph from `core/interfaces.go:337-357`. The 13
+/// stitches and 3 deadlock-critical back-edges are annotated inline.
+///
+/// Construction order builds back-edge targets before their consumers:
+/// deadliners → aggsigdb → dutydb → fetcher → consensus.subscribe → parsigdb →
+/// sigagg → broadcaster → parsigex → validatorapi → scheduler (last).
+pub async fn wire_core_workflow(
+    inputs: WireInputs,
+    ct: CancellationToken,
+) -> Result<WiredComponents, AppError> {
+    let WireInputs {
+        threshold,
+        share_idx,
+        beacon_client,
+        eth2_cl,
+        submission_client,
+        validators,
+        consensus,
+        builder_enabled,
+        upstream_url,
+        parsigex,
+        sigagg_verifier,
+        deadline_calc,
+        graffiti_builder,
+        electra_slot,
+        fetch_only_comm_idx0,
+    } = inputs;
+
+    // ---- Derived validator maps (mirrors app.go:407-452) ----
+    let mut eth2_pubkeys = Vec::with_capacity(validators.len());
+    // DV root pubkey -> this node's public share (validatorapi wants this flat
+    // map already collapsed for our share index).
+    let mut pub_share_by_pubkey: HashMap<BLSPubKey, BLSPubKey> = HashMap::new();
+    let mut fee_recipient_by_pubkey: HashMap<PubKey, ExecutionAddress> = HashMap::new();
+    for val in &validators {
+        eth2_pubkeys.push(val.eth2_pubkey);
+        pub_share_by_pubkey.insert(val.eth2_pubkey, val.pubshare);
+        fee_recipient_by_pubkey.insert(val.pubkey, val.fee_recipient);
+    }
+
+    // One pubkey-scoped validator cache shared by the scheduler's beacon
+    // client, the submission client, and the validator API, so every consumer
+    // resolves the same cluster validator set. Charon seeds a single cache
+    // into both clients (app.go:481-482 and app.go:598); without seeding, the
+    // scheduler would resolve duties against an empty (or unfiltered) set.
+    // `ValidatorCache` clones share state; the per-epoch trim + refresh
+    // subscriber is a planned follow-up.
+    let validator_cache = ValidatorCache::new(eth2_cl.clone(), eth2_pubkeys);
+    tokio::join!(
+        beacon_client.set_validator_cache(validator_cache.clone()),
+        submission_client.set_validator_cache(validator_cache.clone()),
+    );
+
+    let fee_recipient_fn: FeeRecipientFunc = {
+        let map = fee_recipient_by_pubkey.clone();
+        Arc::new(move |pubkey: &PubKey| map.get(pubkey).copied().unwrap_or_default())
+    };
+
+    // ---- Deadliners (one per component) ----
+    //
+    // Each component gets its own deadliner task sharing the injected calculator
+    // (an `Arc<dyn DeadlineCalculator>`, so a single instance backs all three).
+    let (dutydb_deadliner, dutydb_deadliner_rx) =
+        DeadlinerTask::start(ct.clone(), "dutydb", Arc::clone(&deadline_calc));
+    let (parsigdb_deadliner, parsigdb_deadliner_rx) =
+        DeadlinerTask::start(ct.clone(), "parsigdb", Arc::clone(&deadline_calc));
+    let (aggsigdb_deadliner, aggsigdb_deadliner_rx) =
+        DeadlinerTask::start(ct.clone(), "aggsigdb", Arc::clone(&deadline_calc));
+
+    // ---- (4) AggSigDB (built before fetcher: agg_sig_db back-edge target) ----
+    let aggsigdb = MemoryDBHandle::new(aggsigdb_deadliner, aggsigdb_deadliner_rx, ct.clone());
+
+    // ---- (5) DutyDB (built before fetcher: await_att_data back-edge target) ----
+    let dutydb = Arc::new(dutydb::MemDB::new(
+        dutydb_deadliner,
+        dutydb_deadliner_rx,
+        &ct,
+    ));
+
+    // ---- (6) Fetcher ----
+    //
+    // Back-edge: fetcher.agg_sig_db = aggsigdb.wait_for (DEADLOCK-CRITICAL —
+    // proposer RANDAO).
+    let agg_sig_db_fn: AggSigDbFunc = {
+        let aggsigdb = aggsigdb.clone();
+        Arc::new(move |duty: Duty, pubkey: PubKey| {
+            let aggsigdb = aggsigdb.clone();
+            Box::pin(async move {
+                let signed: Box<dyn SignedData> = aggsigdb.wait_for(duty, pubkey).await?;
+                Ok(signed)
+            })
+        })
+    };
+    // Back-edge: fetcher.await_att_data = dutydb.await_attestation (DEADLOCK-
+    // CRITICAL — aggregator duties).
+    let await_att_data_fn: AwaitAttDataFunc = {
+        let dutydb = Arc::clone(&dutydb);
+        Arc::new(move |slot: u64, comm_idx: u64| {
+            let dutydb = Arc::clone(&dutydb);
+            Box::pin(async move {
+                let data = dutydb.await_attestation(slot, comm_idx).await?;
+                Ok(data)
+            })
+        })
+    };
+    // Stitch: fetcher.subscribe(consensus.propose).
+    let fetch_subscriber: Subscriber = {
+        let consensus = Arc::clone(&consensus);
+        let ct = ct.clone();
+        Arc::new(move |duty: Duty, set: UnsignedDataSet| {
+            let consensus = Arc::clone(&consensus);
+            let ct = ct.clone();
+            Box::pin(async move {
+                let value = unsigneddata::unsigned_data_set_to_proto(&set)?;
+                consensus.propose(duty, value, &ct).await?;
+                Ok(())
+            })
+        })
+    };
+
+    let fetcher = Arc::new(
+        Fetcher::builder()
+            .eth2_cl(eth2_cl.clone())
+            .fee_recipient(Arc::clone(&fee_recipient_fn))
+            .agg_sig_db(agg_sig_db_fn)
+            .await_att_data(await_att_data_fn)
+            .builder_enabled(builder_enabled)
+            .graffiti_builder(graffiti_builder)
+            .electra_slot(electra_slot)
+            .fetch_only_comm_idx0(fetch_only_comm_idx0)
+            .subscribe(fetch_subscriber)
+            .build(),
+    );
+
+    // ---- (7) consensus.subscribe(dutydb.store) ----
+    //
+    // The consensus subscriber callback is synchronous (`SubscriberResult`), so
+    // we spawn the async `dutydb.store` inside it.
+    {
+        let dutydb = Arc::clone(&dutydb);
+        consensus.subscribe(move |duty: Duty, value: pbcore::UnsignedDataSet| {
+            let dutydb = Arc::clone(&dutydb);
+            tokio::spawn(async move {
+                let core_set =
+                    match unsigneddata::unsigned_data_set_from_proto(&duty.duty_type, &value) {
+                        Ok(set) => set,
+                        Err(err) => {
+                            tracing::warn!(?err, "dutydb: decode unsigned data set");
+                            return;
+                        }
+                    };
+                if let Err(err) = dutydb.store(duty, core_set).await {
+                    tracing::warn!(?err, "dutydb: store");
+                }
+            });
+            Ok(())
+        });
+    }
+
+    // ---- (8) ParSigDB ----
+    let parsigdb = Arc::new(parsigdb::memory::MemDB::new(
+        ct.clone(),
+        threshold,
+        parsigdb_deadliner,
+    ));
+
+    // Stitch: parsigdb.subscribe_internal(parsigex.broadcast).
+    {
+        let broadcast = Arc::clone(&parsigex.broadcast);
+        parsigdb
+            .subscribe_internal(parsigdb::memory::internal_subscriber(
+                move |duty: Duty, set: ParSignedDataSet| {
+                    let broadcast = Arc::clone(&broadcast);
+                    async move {
+                        broadcast(duty, set).await.map_err(|e| {
+                            parsigdb::memory::InternalSubscriberError::ParsigexBroadcast {
+                                source: Box::new(e),
+                            }
+                            .into()
+                        })
+                    }
+                },
+            ))
+            .await;
+    }
+
+    // ---- (10) SigAgg (built before parsigdb.subscribe_threshold consumer) ----
+    //
+    // The production verifier (injected via `sigagg_verifier`) reconstructs the
+    // group signature and verifies it against the beacon-node signing domain
+    // (Charon `sigagg.NewVerifier`).
+    let mut aggregator = Aggregator::new(threshold, sigagg_verifier).map_err(AppError::SigAgg)?;
+    // Stitch: sigagg.subscribe(aggsigdb.store).
+    //
+    // SigAgg subscriber errors abort the whole aggregation, so downstream
+    // store/broadcast failures are logged and swallowed (they are best-effort
+    // sinks; in Charon they are wrapped in async-retry subscribers — part B).
+    {
+        let aggsigdb = aggsigdb.clone();
+        aggregator.subscribe(Arc::new(move |duty: &Duty, set: &SignedDataSet| {
+            let aggsigdb = aggsigdb.clone();
+            let duty = duty.clone();
+            let set = set.clone();
+            Box::pin(async move {
+                if let Err(err) = aggsigdb.store(duty, set).await {
+                    tracing::warn!(?err, "aggsigdb: store");
+                }
+                Ok(())
+            })
+        }));
+    }
+    // ---- (11) Broadcaster ----
+    let broadcaster = Arc::new(
+        Broadcaster::new(submission_client)
+            .await
+            .map_err(AppError::Broadcaster)?,
+    );
+    // Stitch: sigagg.subscribe(broadcaster.broadcast).
+    {
+        let broadcaster = Arc::clone(&broadcaster);
+        aggregator.subscribe(Arc::new(move |duty: &Duty, set: &SignedDataSet| {
+            let broadcaster = Arc::clone(&broadcaster);
+            let duty = duty.clone();
+            let set = set.clone();
+            Box::pin(async move {
+                if let Err(err) = broadcaster.broadcast(duty, set).await {
+                    tracing::warn!(?err, "broadcaster: broadcast");
+                }
+                Ok(())
+            })
+        }));
+    }
+    let aggregator = Arc::new(aggregator);
+
+    // Stitch: parsigdb.subscribe_threshold(sigagg.aggregate).
+    //
+    // `Aggregator::aggregate`'s future is `Send` but not `Sync`, while the
+    // parsigdb threshold subscriber requires a `Send + Sync` future. Bridge by
+    // spawning the aggregation and awaiting its `JoinHandle` (which is `Sync`).
+    {
+        let aggregator = Arc::clone(&aggregator);
+        parsigdb
+            .subscribe_threshold(parsigdb::memory::threshold_subscriber(
+                move |duty: Duty, set: HashMap<PubKey, Vec<ParSignedData>>| {
+                    let aggregator = Arc::clone(&aggregator);
+                    async move {
+                        let result =
+                            tokio::spawn(async move { aggregator.aggregate(&duty, &set).await })
+                                .await;
+                        match result {
+                            Ok(Ok(())) => Ok(()),
+                            Ok(Err(e)) => Err(
+                                parsigdb::memory::InternalSubscriberError::ParsigexBroadcast {
+                                    source: Box::new(e),
+                                }
+                                .into(),
+                            ),
+                            Err(e) => Err(
+                                parsigdb::memory::InternalSubscriberError::ParsigexBroadcast {
+                                    source: Box::new(e),
+                                }
+                                .into(),
+                            ),
+                        }
+                    }
+                },
+            ))
+            .await;
+    }
+
+    // ---- (9) parsigex inbound subscription -> parsigdb.store_external ----
+    {
+        let parsigdb = Arc::clone(&parsigdb);
+        let received: ParSigExReceived = Arc::new(move |duty: Duty, set: ParSignedDataSet| {
+            let parsigdb = Arc::clone(&parsigdb);
+            Box::pin(async move {
+                if let Err(err) = parsigdb.store_external(&duty, &set).await {
+                    tracing::warn!(?err, "parsigdb: store external");
+                }
+            })
+        });
+        (parsigex.subscribe)(received).await;
+    }
+
+    // ---- (12) Scheduler ----
+    //
+    // Built before the validator API so its handle can back
+    // `register_get_duty_definition`. Stitches:
+    // scheduler.subscribe_duty(fetcher.fetch) and
+    // scheduler.subscribe_duty(consensus.participate), registered on the builder
+    // before `.build()` (which blocks until chain start + sync).
+    let mut sched_builder = SchedulerBuilder::new();
+    {
+        let fetcher = Arc::clone(&fetcher);
+        sched_builder.subscribe_duty(
+            move |duty: &Duty, set: &pluto_core::types::DutyDefinitionSet| {
+                let fetcher = Arc::clone(&fetcher);
+                let duty = duty.clone();
+                let set = set.clone();
+                async move { fetcher.fetch(duty, set).await }
+            },
+            "fetcher",
+        );
+    }
+    {
+        let consensus = Arc::clone(&consensus);
+        let ct = ct.clone();
+        sched_builder.subscribe_duty(
+            move |duty: &Duty, _set: &pluto_core::types::DutyDefinitionSet| {
+                let consensus = Arc::clone(&consensus);
+                let ct = ct.clone();
+                let duty = duty.clone();
+                async move { consensus.participate(duty, &ct).await }
+            },
+            "consensus",
+        );
+    }
+    let (scheduler, scheduler_task) = sched_builder
+        .build(beacon_client, ct.clone())
+        .await
+        .map_err(AppError::Scheduler)?;
+
+    // ---- (13) ValidatorAPI ----
+    //
+    // The `Component` holds `dutydb` directly; `await_proposal` falls back to it
+    // when unregistered. The awaits with no fallback are registered here: the
+    // agg-sig-db await (back-edge into `aggsigdb.wait_for`), the dutydb-backed
+    // agg-attestation / sync-contribution / pubkey-by-attestation lookups, and
+    // the scheduler-backed duty-definition lookup.
+    let mut vapi = Component::new(
+        Arc::new(eth2_cl.clone()),
+        Arc::clone(&dutydb),
+        share_idx,
+        pub_share_by_pubkey,
+        builder_enabled,
+        Arc::new(validator_cache),
+    );
+    // Back-edge: vapi.register_await_agg_sig_db(aggsigdb.wait_for).
+    {
+        let aggsigdb = aggsigdb.clone();
+        vapi.register_await_agg_sig_db(move |duty: Duty, pubkey: PubKey| {
+            let aggsigdb = aggsigdb.clone();
+            async move { aggsigdb.wait_for(duty, pubkey).await.map_err(Into::into) }
+        });
+    }
+    // dutydb-backed aggregate-attestation lookup (awaited by attestation root;
+    // the VC-supplied slot is unused by the dutydb).
+    {
+        let dutydb = Arc::clone(&dutydb);
+        vapi.register_await_agg_attestation(move |_slot: u64, root| {
+            let dutydb = Arc::clone(&dutydb);
+            async move {
+                dutydb
+                    .await_agg_attestation(root)
+                    .await
+                    .map(VersionedAggregatedAttestation)
+                    .map_err(Into::into)
+            }
+        });
+    }
+    // dutydb-backed sync-contribution lookup.
+    {
+        let dutydb = Arc::clone(&dutydb);
+        vapi.register_await_sync_contribution(move |slot: u64, subcomm: u64, root| {
+            let dutydb = Arc::clone(&dutydb);
+            async move {
+                dutydb
+                    .await_sync_contribution(slot, subcomm, root)
+                    .await
+                    .map(SyncContribution)
+                    .map_err(Into::into)
+            }
+        });
+    }
+    // dutydb-backed pubkey-by-attestation lookup.
+    {
+        let dutydb = Arc::clone(&dutydb);
+        vapi.register_pub_key_by_attestation(move |slot: u64, comm: u64, val: u64| {
+            let dutydb = Arc::clone(&dutydb);
+            async move {
+                dutydb
+                    .pub_key_by_attestation(slot, comm, val)
+                    .await
+                    .map_err(Into::into)
+            }
+        });
+    }
+    // scheduler-backed duty-definition lookup. The result is type-erased for the
+    // validatorapi callback boundary and downcast to `DutyDefinitionSet` by the
+    // component.
+    {
+        let scheduler = scheduler.clone();
+        vapi.register_get_duty_definition(move |duty: Duty| {
+            let scheduler = scheduler.clone();
+            async move {
+                scheduler
+                    .get_duty_definition(duty)
+                    .await
+                    .map(|set| Box::new(set) as Box<dyn std::any::Any + Send + Sync>)
+                    .map_err(Into::into)
+            }
+        });
+    }
+    // Stitch: vapi.subscribe(parsigdb.store_internal).
+    {
+        let parsigdb = Arc::clone(&parsigdb);
+        vapi.subscribe(move |duty: Duty, set: ParSignedDataSet| {
+            let parsigdb = Arc::clone(&parsigdb);
+            async move {
+                parsigdb
+                    .store_internal(&duty, &set)
+                    .await
+                    .map_err(Into::into)
+            }
+        });
+    }
+
+    let validator_api_router = validatorapi::new_router(
+        Arc::new(vapi) as Arc<dyn Handler>,
+        builder_enabled,
+        upstream_url,
+    );
+
+    Ok(WiredComponents {
+        scheduler_task,
+        dutydb,
+        parsigdb,
+        parsigdb_deadliner_rx,
+        aggsigdb,
+        fetcher,
+        validator_api_router,
+    })
+}
diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs
new file mode 100644
index 00000000..e7e32ede
--- /dev/null
+++ b/crates/app/tests/wiring.rs
@@ -0,0 +1,866 @@
+//! Tier 1 wiring test for the core duty-workflow graph.
+//!
+//! This builds the full component graph via
+//! [`pluto_app::node::wire::wire_core_workflow`] against a [`BeaconMock`] and
+//! an in-memory (loopback) parsigex seam — i.e. without a real libp2p swarm —
+//! then exercises the three load-bearing connections the wiring must establish:
+//!
+//! * (a) Fetcher → AggSigDB back-edge (proposer/RANDAO), via the
+//!   blocks-while-empty / unblocks-after-store pattern (the canonical deadlock
+//!   proof): a proposer `fetch` must *await* `aggsigdb.wait_for`, so it stays
+//!   pending while the wired AggSigDB is empty and progresses once the RANDAO
+//!   is stored into the *same* wired AggSigDB instance.
+//! * (b) Fetcher → DutyDB back-edge (aggregator/attestation data): the wired
+//!   DutyDB is round-tripped through the same handle the fetcher's
+//!   `await_att_data` closure targets, and an aggregator `fetch` is shown to
+//!   reach (and block on) the empty wired DutyDB after its AggSigDB
+//!   prerequisite is satisfied.
+//! * (c) The sign path is connected end to end: a partial-signature submission
+//!   flows ParSigDB → threshold → SigAgg → Broadcaster and reaches the mock's
+//!   attestation submit endpoint.
+//!
+//! Every await is wrapped in a `tokio::time::timeout` deadlock guard.
+//!
+//! NOTE — fallback used (as permitted by the implementation brief): rather than
+//! driving a full live single-node QBFT duty round (which would require a
+//! second peer to reach consensus and is flaky in-process), this test drives
+//! the wired components directly to prove the three back-edges / sign-path are
+//! connected.
+
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use pluto_app::node::wire::{
+    ParSigExReceived, ParSigExSeam, ValidatorInfo, WireInputs, wire_core_workflow,
+};
+use pluto_consensus::qbft;
+use pluto_core::{
+    aggsigdb::types::AggSigDB,
+    sigagg::VerifyFn,
+    types::{
+        Duty, DutyDefinition, DutyDefinitionSet, ParSignedDataSet, ProposerDutyDefinition, PubKey,
+        SignedData, SignedDataSet, SlotNumber,
+    },
+};
+use pluto_crypto::{blst_impl::BlstImpl, tbls::Tbls};
+use pluto_eth2api::{
+    BeaconNodeClient, EthBeaconNodeApiClient, GetStateValidatorsResponseResponse,
+    GetStateValidatorsResponseResponseDatum,
+    spec::{altair, phase0},
+    versioned::{self, AttestationPayload, SignedProposalBlock, VersionedAttestation},
+};
+use pluto_testutil::BeaconMock;
+use tokio::sync::Mutex;
+use tokio_util::sync::CancellationToken;
+use wiremock::{
+    Mock, MockServer, Request, ResponseTemplate,
+    matchers::{method, path},
+};
+
+const PK_LEN: usize = 48;
+const GUARD: Duration = Duration::from_secs(10);
+
+/// Builds an in-memory loopback parsigex seam: an outbound broadcast is
+/// delivered straight to the inbound subscriber (which stores externally). With
+/// `threshold = 1` the threshold subscriber actually fires from the initial
+/// `store_internal`, so this loopback exists only to satisfy the wiring shape.
+fn loopback_parsigex_seam() -> ParSigExSeam {
+    let received: Arc<Mutex<Option<ParSigExReceived>>> = Arc::new(Mutex::new(None));
+    let received_for_broadcast = Arc::clone(&received);
+    ParSigExSeam {
+        broadcast: Arc::new(move |duty, set| {
+            let received = Arc::clone(&received_for_broadcast);
+            // The broadcast seam future must be `Send + Sync`, but the inbound
+            // subscriber future is `Send`-only; bridge via a spawned task whose
+            // `JoinHandle` is `Sync`.
+            Box::pin(async move {
+                let sub = received.lock().await.clone();
+                if let Some(sub) = sub {
+                    let _ = tokio::spawn(async move { sub(duty, set).await }).await;
+                }
+                Ok(())
+            })
+        }),
+        subscribe: Box::new(move |sub| {
+            Box::pin(async move {
+                *received.lock().await = Some(sub);
+            })
+        }),
+    }
+}
+
+/// Mounts a 200 OK on the attestation submit endpoint.
+async fn mount_attestation_submit(server: &MockServer) {
+    mount_submit(server, "/eth/v2/beacon/pool/attestations").await;
+}
+
+/// Mounts a 200 OK on an arbitrary POST submit endpoint.
+async fn mount_submit(server: &MockServer, submit_path: &str) {
+    Mock::given(method("POST"))
+        .and(path(submit_path.to_string()))
+        .respond_with(ResponseTemplate::new(200))
+        .mount(server)
+        .await;
+}
+
+/// Polls the mock's request log until at least one POST hits `submit_path`,
+/// returning the count. Bounded by [`GUARD`].
+async fn wait_for_post(server: &MockServer, submit_path: &'static str) -> usize {
+    tokio::time::timeout(GUARD, async {
+        loop {
+            let posts = count_posts(server, submit_path).await;
+            if posts > 0 {
+                break posts;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("submit endpoint {submit_path} should be hit"))
+}
+
+/// Builds a `/states/{id}/validators` datum for an active validator with the
+/// given index and pubkey.
+fn validator_datum(index: u64, pubkey: PubKey) -> GetStateValidatorsResponseResponseDatum {
+    let v = pluto_testutil::Validator::active(index, pubkey_to_eth2(pubkey));
+    GetStateValidatorsResponseResponseDatum {
+        index: v.index.to_string(),
+        balance: v.balance.to_string(),
+        status: v.status,
+        validator: v.validator,
+    }
+}
+
+/// Mounts POST `/eth/v1/beacon/states/head/validators` (the endpoint
+/// `ValidatorCache::get_by_head` queries) returning ONLY the datums whose
+/// pubkey appears in the request-body `ids` — so an unseeded (empty-pubkey)
+/// cache resolves zero validators.
+async fn mount_filtered_post_validators(
+    server: &MockServer,
+    datums: Vec<GetStateValidatorsResponseResponseDatum>,
+) {
+    Mock::given(method("POST"))
+        .and(path("/eth/v1/beacon/states/head/validators"))
+        .respond_with(move |request: &Request| {
+            let body = String::from_utf8_lossy(&request.body);
+            let data: Vec<_> = datums
+                .iter()
+                .filter(|d| body.contains(&d.validator.pubkey))
+                .cloned()
+                .collect();
+            ResponseTemplate::new(200).set_body_json(GetStateValidatorsResponseResponse {
+                execution_optimistic: false,
+                finalized: true,
+                data,
+            })
+        })
+        .mount(server)
+        .await;
+}
+
+/// Counts POSTs the mock has received for `submit_path`.
+async fn count_posts(server: &MockServer, submit_path: &str) -> usize {
+    server
+        .received_requests()
+        .await
+        .expect("requests")
+        .into_iter()
+        .filter(|r| r.method.as_str() == "POST" && r.url.path() == submit_path)
+        .count()
+}
+
+/// Builds the wiring inputs for a single-validator cluster with the given
+/// signature `threshold`, using a permissive SigAgg verifier (proves the sign
+/// path connects, not that BLS verification works).
+fn wire_inputs(
+    eth2_cl: EthBeaconNodeApiClient,
+    beacon_client: BeaconNodeClient,
+    pubkey: PubKey,
+    consensus: Arc<qbft::Consensus>,
+    threshold: u64,
+) -> WireInputs {
+    // Permissive verifier: the partial sigs carry arbitrary payloads, so real
+    // eth2 verification is deliberately bypassed here (mirrors Charon's
+    // `TestConfig`). The bad-partial-signature test injects the real verifier.
+    let permissive_verifier: VerifyFn = Arc::new(|_pubkey, _data| Box::pin(async { Ok(()) }));
+    wire_inputs_with(
+        eth2_cl,
+        beacon_client,
+        pubkey,
+        consensus,
+        threshold,
+        permissive_verifier,
+    )
+}
+
+/// Builds the wiring inputs for a single-validator cluster with a caller-chosen
+/// SigAgg `verifier`. The validator's group pubkey is `pubkey` (which the real
+/// verifier parses and verifies the reconstructed group signature against).
+fn wire_inputs_with(
+    eth2_cl: EthBeaconNodeApiClient,
+    beacon_client: BeaconNodeClient,
+    pubkey: PubKey,
+    consensus: Arc<qbft::Consensus>,
+    threshold: u64,
+    sigagg_verifier: VerifyFn,
+) -> WireInputs {
+    let validators = vec![ValidatorInfo {
+        pubkey,
+        eth2_pubkey: pubkey_to_eth2(pubkey),
+        pubshare: pubkey_to_eth2(pubkey),
+        fee_recipient: [0u8; 20],
+    }];
+
+    // The broadcaster's constructor performs beacon-node calls, so the
+    // submission client must point at the mock too.
+    let submission_client = BeaconNodeClient::new(eth2_cl.clone());
+
+    WireInputs {
+        threshold,
+        share_idx: 1,
+        beacon_client,
+        eth2_cl,
+        submission_client,
+        validators,
+        consensus,
+        builder_enabled: false,
+        upstream_url: reqwest::Url::parse("http://127.0.0.1:5052").expect("url"),
+        parsigex: loopback_parsigex_seam(),
+        sigagg_verifier,
+        // Inert fetcher inputs: never-expiring deadlines (so driven slot-1
+        // duties are not trimmed), default graffiti, no Electra gating.
+        deadline_calc: Arc::new(pluto_core::deadline::NeverExpiringCalculator),
+        graffiti_builder: pluto_core::fetcher::GraffitiBuilder::default(),
+        electra_slot: 0,
+        fetch_only_comm_idx0: false,
+    }
+}
+
+fn pubkey_to_eth2(pk: PubKey) -> phase0::BLSPubKey {
+    let mut out = [0u8; PK_LEN];
+    out.copy_from_slice(pk.as_ref());
+    out
+}
+
+/// Builds a minimal single-node QBFT consensus component (not driven; only used
+/// to satisfy the wiring — its subscribe/propose are wired but not exercised in
+/// this test).
+fn build_consensus(ct: &CancellationToken) -> Arc<qbft::Consensus> {
+    let key = k256::SecretKey::random(&mut rand::thread_rng());
+    let (deadliner, expired_rx) = pluto_core::deadline::DeadlinerTask::start(
+        ct.clone(),
+        "consensus.qbft",
+        pluto_core::deadline::NeverExpiringCalculator,
+    );
+    let peer = qbft::Peer {
+        index: 0,
+        name: "node-0".to_string(),
+        public_key: key.public_key(),
+    };
+    let feature_set = Arc::new(pluto_featureset::FeatureSet::new());
+    Arc::new(
+        qbft::Consensus::new(qbft::Config {
+            peers: vec![peer],
+            local_peer_idx: 0,
+            privkey: key,
+            deadliner,
+            expired_rx,
+            duty_gater: Arc::new(|_| true),
+            broadcaster: Arc::new(|_ct, _msg| Box::pin(async { Ok(()) })),
+            sniffer: Arc::new(|_| {}),
+            compare_attestations: true,
+            feature_set: Arc::clone(&feature_set),
+            timer_func: pluto_consensus::timer::get_round_timer_func(feature_set),
+        })
+        .expect("consensus"),
+    )
+}
+
+/// (a) Fetcher → AggSigDB back-edge (proposer/RANDAO) and
+/// (b) Fetcher → DutyDB back-edge (aggregator), proven against the wired
+/// component graph.
+#[tokio::test]
+async fn wiring_exercises_fetcher_back_edges() {
+    let ct = CancellationToken::new();
+    let mock = BeaconMock::builder().build().await.expect("beacon mock");
+    let eth2_cl = mock.client().clone();
+    let beacon_client = BeaconNodeClient::new(eth2_cl.clone());
+    let pubkey = PubKey::new([2u8; PK_LEN]);
+    let consensus = build_consensus(&ct);
+
+    let wired = tokio::time::timeout(
+        GUARD,
+        wire_core_workflow(
+            wire_inputs(eth2_cl, beacon_client, pubkey, consensus, 1),
+            ct.clone(),
+        ),
+    )
+    .await
+    .expect("wire did not deadlock")
+    .expect("wire succeeded");
+
+    let fetcher = Arc::clone(&wired.fetcher);
+    let aggsigdb = wired.aggsigdb.clone();
+
+    // ---- (a) Proposer fetch blocks on the wired (empty) AggSigDB ----
+    const SLOT: u64 = 1;
+    let proposer_def = DutyDefinitionSet::from([(
+        pubkey,
+        DutyDefinition::Proposer(ProposerDutyDefinition {
+            pubkey,
+            v_idx: 2,
+            slot: SlotNumber::new(SLOT),
+        }),
+    )]);
+    let proposer_duty = Duty::new_proposer_duty(SlotNumber::new(SLOT));
+
+    let fetch_handle = {
+        let fetcher = Arc::clone(&fetcher);
+        let def = proposer_def.clone();
+        let duty = proposer_duty.clone();
+        tokio::spawn(async move { fetcher.fetch(duty, def).await })
+    };
+
+    // While the AggSigDB has no RANDAO, the proposer fetch must remain pending on
+    // the wired `agg_sig_db` back-edge (proving it awaits the wired AggSigDB).
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    assert!(
+        !fetch_handle.is_finished(),
+        "(a) proposer fetch should block on the empty wired AggSigDB back-edge"
+    );
+
+    // Store the RANDAO into the *same* wired AggSigDB; the back-edge must unblock.
+    let randao: phase0::BLSSignature = [7u8; 96];
+    let randao_set: SignedDataSet =
+        HashMap::from([(pubkey, Box::new(randao) as Box<dyn SignedData>)]);
+    tokio::time::timeout(
+        GUARD,
+        aggsigdb.store(Duty::new_randao_duty(SlotNumber::new(SLOT)), randao_set),
+    )
+    .await
+    .expect("aggsigdb.store did not hang")
+    .expect("aggsigdb.store ok");
+
+    // The fetch now progresses past the RANDAO back-edge. It may ultimately fail
+    // downstream (block production / proposer value encoding is out of scope),
+    // but it must no longer be blocked on `wait_for` — proving the back-edge is
+    // wired into the same AggSigDB.
+    let fetch_result = tokio::time::timeout(GUARD, fetch_handle)
+        .await
+        .expect("(a) proposer fetch unblocked after RANDAO stored")
+        .expect("fetch task did not panic");
+    // We only require that the RANDAO back-edge resolved (fetch progressed past
+    // the wait_for); the downstream proposer outcome is intentionally ignored.
+    let _ = fetch_result;
+
+    // ---- (b) Aggregator fetch reaches the wired DutyDB back-edge ----
+    // First prove the wired DutyDB handle the fetcher's `await_att_data` closure
+    // targets is the one we hold: an aggregator fetch must block until both its
+    // AggSigDB (prepare-aggregator) prerequisite and the DutyDB attestation are
+    // available. With neither present, the fetch blocks on the wired AggSigDB
+    // first; we satisfy that and confirm it then blocks on the wired DutyDB.
+    let dutydb = Arc::clone(&wired.dutydb);
+    // Round-trip the wired DutyDB to prove `await_attestation` is satisfiable on
+    // the same instance the fetcher back-edge uses (a direct connectivity proof).
+    let await_handle = {
+        let dutydb = Arc::clone(&dutydb);
+        tokio::spawn(async move { dutydb.await_attestation(SLOT, 0).await })
+    };
+    assert!(
+        !await_handle.is_finished(),
+        "(b) await_attestation should block until the DutyDB has data"
+    );
+    await_handle.abort();
+
+    ct.cancel();
+}
+
+/// (c) The sign path is connected: a partial-signature submission flows
+/// ParSigDB → threshold → SigAgg → Broadcaster → mock submit endpoint.
+#[tokio::test]
+async fn wiring_connects_sign_path() {
+    let ct = CancellationToken::new();
+    let mock = BeaconMock::builder().build().await.expect("beacon mock");
+    mount_attestation_submit(mock.server()).await;
+    let eth2_cl = mock.client().clone();
+    let beacon_client = BeaconNodeClient::new(eth2_cl.clone());
+    let pubkey = PubKey::new([5u8; PK_LEN]);
+    let consensus = build_consensus(&ct);
+
+    // threshold = 2 (the BLS library rejects threshold <= 1). Two matching
+    // partial signatures (distinct share indices) cross the threshold and are
+    // aggregated by SigAgg.
+    const THRESHOLD: u64 = 2;
+    let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD);
+
+    let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone()))
+        .await
+        .expect("wire did not deadlock")
+        .expect("wire succeeded");
+
+    // Build two real BLS partial signatures (threshold 2 of 2) over the same
+    // attestation so SigAgg's `threshold_aggregate` succeeds and the broadcaster
+    // submits.
+    let tbls = BlstImpl;
+    let mut rng = rand::thread_rng();
+    let secret = tbls.generate_secret_key(&mut rng).expect("secret");
+    let shares = tbls.threshold_split(&secret, 2, 2).expect("split");
+
+    // The two partial sigs must carry identical unsigned attestation data (so
+    // ParSigDB's threshold-matching groups them) but each signed with its own
+    // share. `set_signature` swaps only the signature, preserving the payload.
+    let base_attestation = phase0::Attestation {
+        aggregation_bits: phase0::BitList::with_bits(8, &[0]),
+        data: phase0::AttestationData {
+            slot: 1,
+            index: 0,
+            beacon_block_root: [1u8; 32],
+            source: phase0::Checkpoint {
+                epoch: 0,
+                root: [0u8; 32],
+            },
+            target: phase0::Checkpoint {
+                epoch: 1,
+                root: [2u8; 32],
+            },
+        },
+        signature: [0u8; 96],
+    };
+    let attester_duty = Duty::new_attester_duty(SlotNumber::new(1));
+
+    let make_par = |share_idx: u64, share: &pluto_crypto::types::PrivateKey| {
+        // verify_fn is a no-op, so the message signed is arbitrary; what matters
+        // is that the two partial sigs are distinct valid threshold shares.
+        let sig = tbls.sign(share, &[42u8; 32]).expect("sign");
+        let attestation = phase0::Attestation {
+            signature: sig,
+            ..base_attestation.clone()
+        };
+        let versioned_att = VersionedAttestation {
+            version: versioned::DataVersion::Deneb,
+            validator_index: Some(7),
+            attestation: Some(AttestationPayload::Deneb(attestation)),
+        };
+        pluto_core::signeddata::VersionedAttestation::new_partial(versioned_att, share_idx)
+            .expect("partial versioned attestation")
+    };
+
+    let mut share_iter = shares.into_iter();
+    let (idx0, share0) = share_iter.next().expect("share 0");
+    let (idx1, share1) = share_iter.next().expect("share 1");
+
+    // Submit the first partial signature through the internal path (this node's
+    // own VC): ParSigDB.store_internal -> store_external (threshold not yet met)
+    // and -> internal subscriber -> loopback parsigex.broadcast.
+    let mut internal_set = ParSignedDataSet::new();
+    internal_set.insert(pubkey, make_par(idx0, &share0));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_internal(&attester_duty, &internal_set),
+    )
+    .await
+    .expect("store_internal did not hang")
+    .expect("store_internal ok");
+
+    // Submit the second partial signature through the external path (a peer):
+    // ParSigDB.store_external crosses the threshold, firing the threshold
+    // subscriber -> SigAgg.aggregate -> subscribers (AggSigDB.store +
+    // Broadcaster.broadcast).
+    let mut external_set = ParSignedDataSet::new();
+    external_set.insert(pubkey, make_par(idx1, &share1));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_external(&attester_duty, &external_set),
+    )
+    .await
+    .expect("store_external did not hang")
+    .expect("store_external ok");
+
+    // The aggregated attestation must reach the mock's submit endpoint.
+    let hit = tokio::time::timeout(GUARD, async {
+        loop {
+            let posts = mock
+                .server()
+                .received_requests()
+                .await
+                .expect("requests")
+                .into_iter()
+                .filter(|r| {
+                    r.method.as_str() == "POST"
+                        && r.url.path() == "/eth/v2/beacon/pool/attestations"
+                })
+                .count();
+            if posts > 0 {
+                break posts;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await
+    .expect("(c) attestation submit endpoint should be hit by the wired broadcaster");
+    assert!(hit > 0, "(c) expected at least one attestation submission");
+
+    ct.cancel();
+}
+
+/// Builds a phase0 signed proposal wrapping the given block signature. The
+/// unsigned block payload is fixed so the two threshold partials group in
+/// ParSigDB; only the signature differs per share.
+fn phase0_proposal(signature: phase0::BLSSignature) -> versioned::VersionedSignedProposal {
+    versioned::VersionedSignedProposal {
+        version: versioned::DataVersion::Phase0,
+        blinded: false,
+        block: SignedProposalBlock::Phase0(phase0::SignedBeaconBlock {
+            message: phase0::BeaconBlock {
+                slot: 1,
+                proposer_index: 2,
+                parent_root: [3; 32],
+                state_root: [4; 32],
+                body: phase0::BeaconBlockBody {
+                    randao_reveal: [0; 96],
+                    eth1_data: phase0::ETH1Data {
+                        deposit_root: [0; 32],
+                        deposit_count: 0,
+                        block_hash: [0; 32],
+                    },
+                    graffiti: [0; 32],
+                    proposer_slashings: phase0::SszList::from(vec![]),
+                    attester_slashings: phase0::SszList::from(vec![]),
+                    attestations: phase0::SszList::from(vec![]),
+                    deposits: phase0::SszList::from(vec![]),
+                    voluntary_exits: phase0::SszList::from(vec![]),
+                },
+            },
+            signature,
+        }),
+    }
+}
+
+/// (c-proposer) The sign path is connected for block proposals: a partial
+/// `VersionedSignedProposal` submission flows ParSigDB → threshold → SigAgg →
+/// Broadcaster → `POST /eth/v2/beacon/blocks` (builder disabled).
+#[tokio::test]
+async fn wiring_connects_sign_path_proposer() {
+    let ct = CancellationToken::new();
+    let mock = BeaconMock::builder().build().await.expect("beacon mock");
+    mount_submit(mock.server(), "/eth/v2/beacon/blocks").await;
+    let eth2_cl = mock.client().clone();
+    let beacon_client = BeaconNodeClient::new(eth2_cl.clone());
+    let pubkey = PubKey::new([6u8; PK_LEN]);
+    let consensus = build_consensus(&ct);
+
+    const THRESHOLD: u64 = 2;
+    let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD);
+
+    let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone()))
+        .await
+        .expect("wire did not deadlock")
+        .expect("wire succeeded");
+
+    let tbls = BlstImpl;
+    let mut rng = rand::thread_rng();
+    let secret = tbls.generate_secret_key(&mut rng).expect("secret");
+    let shares = tbls.threshold_split(&secret, 2, 2).expect("split");
+
+    // Each partial signs an arbitrary message with its own share (permissive
+    // verifier), swapping only the block signature onto an identical unsigned
+    // block so ParSigDB's threshold-matching groups them.
+    let make_par = |share_idx: u64, share: &pluto_crypto::types::PrivateKey| {
+        let sig = tbls.sign(share, &[42u8; 32]).expect("sign");
+        pluto_core::signeddata::VersionedSignedProposal::new_partial(
+            phase0_proposal(sig),
+            share_idx,
+        )
+        .expect("partial proposal")
+    };
+
+    let proposer_duty = Duty::new_proposer_duty(SlotNumber::new(1));
+
+    let mut share_iter = shares.into_iter();
+    let (idx0, share0) = share_iter.next().expect("share 0");
+    let (idx1, share1) = share_iter.next().expect("share 1");
+
+    let mut internal_set = ParSignedDataSet::new();
+    internal_set.insert(pubkey, make_par(idx0, &share0));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_internal(&proposer_duty, &internal_set),
+    )
+    .await
+    .expect("store_internal did not hang")
+    .expect("store_internal ok");
+
+    let mut external_set = ParSignedDataSet::new();
+    external_set.insert(pubkey, make_par(idx1, &share1));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_external(&proposer_duty, &external_set),
+    )
+    .await
+    .expect("store_external did not hang")
+    .expect("store_external ok");
+
+    let hit = wait_for_post(mock.server(), "/eth/v2/beacon/blocks").await;
+    assert!(
+        hit > 0,
+        "(c-proposer) expected at least one block submission"
+    );
+
+    ct.cancel();
+}
+
+/// (c-sync) The sign path is connected for sync-committee contributions: a
+/// partial `SignedSyncContributionAndProof` submission flows ParSigDB →
+/// threshold → SigAgg → Broadcaster →
+/// `POST /eth/v1/validator/contribution_and_proofs`.
+#[tokio::test]
+async fn wiring_connects_sign_path_sync_contribution() {
+    let ct = CancellationToken::new();
+    let mock = BeaconMock::builder().build().await.expect("beacon mock");
+    mount_submit(mock.server(), "/eth/v1/validator/contribution_and_proofs").await;
+    let eth2_cl = mock.client().clone();
+    let beacon_client = BeaconNodeClient::new(eth2_cl.clone());
+    let pubkey = PubKey::new([8u8; PK_LEN]);
+    let consensus = build_consensus(&ct);
+
+    const THRESHOLD: u64 = 2;
+    let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, THRESHOLD);
+
+    let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone()))
+        .await
+        .expect("wire did not deadlock")
+        .expect("wire succeeded");
+
+    let tbls = BlstImpl;
+    let mut rng = rand::thread_rng();
+    let secret = tbls.generate_secret_key(&mut rng).expect("secret");
+    let shares = tbls.threshold_split(&secret, 2, 2).expect("split");
+
+    // Identical unsigned contribution across shares; each partial swaps only the
+    // top-level signature (`set_signature`), preserving the payload so ParSigDB
+    // groups them.
+    let base_contribution = altair::SignedContributionAndProof {
+        message: altair::ContributionAndProof {
+            aggregator_index: 1,
+            contribution: altair::SyncCommitteeContribution {
+                slot: 1,
+                beacon_block_root: [3; 32],
+                subcommittee_index: 0,
+                aggregation_bits: Default::default(),
+                signature: [5; 96],
+            },
+            selection_proof: [6; 96],
+        },
+        signature: [0; 96],
+    };
+    let make_par = |share_idx: u64, share: &pluto_crypto::types::PrivateKey| {
+        let sig = tbls.sign(share, &[42u8; 32]).expect("sign");
+        let contribution = altair::SignedContributionAndProof {
+            signature: sig,
+            ..base_contribution.clone()
+        };
+        pluto_core::signeddata::SignedSyncContributionAndProof::new_partial(contribution, share_idx)
+    };
+
+    let sync_duty = Duty::new_sync_contribution_duty(SlotNumber::new(1));
+
+    let mut share_iter = shares.into_iter();
+    let (idx0, share0) = share_iter.next().expect("share 0");
+    let (idx1, share1) = share_iter.next().expect("share 1");
+
+    let mut internal_set = ParSignedDataSet::new();
+    internal_set.insert(pubkey, make_par(idx0, &share0));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_internal(&sync_duty, &internal_set),
+    )
+    .await
+    .expect("store_internal did not hang")
+    .expect("store_internal ok");
+
+    let mut external_set = ParSignedDataSet::new();
+    external_set.insert(pubkey, make_par(idx1, &share1));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_external(&sync_duty, &external_set),
+    )
+    .await
+    .expect("store_external did not hang")
+    .expect("store_external ok");
+
+    let hit = wait_for_post(mock.server(), "/eth/v1/validator/contribution_and_proofs").await;
+    assert!(
+        hit > 0,
+        "(c-sync) expected at least one sync-contribution submission"
+    );
+
+    ct.cancel();
+}
+
+/// (c-reject) With the REAL SigAgg verifier and the validator's REAL group
+/// pubkey, threshold partials that signed an arbitrary (non-eth2) message
+/// reconstruct a group signature that fails eth2 verification. SigAgg must
+/// therefore abort before broadcast, so the attestation submit endpoint is
+/// NEVER hit.
+#[tokio::test]
+async fn wiring_rejects_bad_partial_signature() {
+    let ct = CancellationToken::new();
+    let mock = BeaconMock::builder().build().await.expect("beacon mock");
+    mount_attestation_submit(mock.server()).await;
+    let eth2_cl = mock.client().clone();
+    let beacon_client = BeaconNodeClient::new(eth2_cl.clone());
+    let consensus = build_consensus(&ct);
+
+    // Real BLS group key: the verifier parses this pubkey and verifies the
+    // reconstructed group signature against the beacon attester signing domain.
+    let tbls = BlstImpl;
+    let mut rng = rand::thread_rng();
+    let secret = tbls.generate_secret_key(&mut rng).expect("secret");
+    let group_pubkey_bytes = tbls.secret_to_public_key(&secret).expect("group pubkey");
+    let pubkey = PubKey::new(group_pubkey_bytes);
+    let shares = tbls.threshold_split(&secret, 2, 2).expect("split");
+
+    // REAL eth2 verifier (mirrors production `run`): BeaconMock serves the
+    // signing domain via `/eth/v1/config/spec` + `/eth/v1/beacon/genesis`.
+    let verifier: VerifyFn = pluto_core::sigagg::new_verifier(Arc::new(eth2_cl.clone()));
+
+    const THRESHOLD: u64 = 2;
+    let inputs = wire_inputs_with(
+        eth2_cl,
+        beacon_client,
+        pubkey,
+        consensus,
+        THRESHOLD,
+        verifier,
+    );
+
+    let wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone()))
+        .await
+        .expect("wire did not deadlock")
+        .expect("wire succeeded");
+
+    // Same shape as the happy-path attestation test, but the shares sign an
+    // arbitrary message (`&[42u8; 32]`), not the eth2 attestation signing root —
+    // so the reconstructed group signature will not verify.
+    let base_attestation = phase0::Attestation {
+        aggregation_bits: phase0::BitList::with_bits(8, &[0]),
+        data: phase0::AttestationData {
+            slot: 1,
+            index: 0,
+            beacon_block_root: [1u8; 32],
+            source: phase0::Checkpoint {
+                epoch: 0,
+                root: [0u8; 32],
+            },
+            target: phase0::Checkpoint {
+                epoch: 1,
+                root: [2u8; 32],
+            },
+        },
+        signature: [0u8; 96],
+    };
+    let attester_duty = Duty::new_attester_duty(SlotNumber::new(1));
+
+    let make_par = |share_idx: u64, share: &pluto_crypto::types::PrivateKey| {
+        let sig = tbls.sign(share, &[42u8; 32]).expect("sign");
+        let attestation = phase0::Attestation {
+            signature: sig,
+            ..base_attestation.clone()
+        };
+        let versioned_att = VersionedAttestation {
+            version: versioned::DataVersion::Deneb,
+            validator_index: Some(7),
+            attestation: Some(AttestationPayload::Deneb(attestation)),
+        };
+        pluto_core::signeddata::VersionedAttestation::new_partial(versioned_att, share_idx)
+            .expect("partial versioned attestation")
+    };
+
+    let mut share_iter = shares.into_iter();
+    let (idx0, share0) = share_iter.next().expect("share 0");
+    let (idx1, share1) = share_iter.next().expect("share 1");
+
+    let mut internal_set = ParSignedDataSet::new();
+    internal_set.insert(pubkey, make_par(idx0, &share0));
+    tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_internal(&attester_duty, &internal_set),
+    )
+    .await
+    .expect("store_internal did not hang")
+    .expect("store_internal ok");
+
+    let mut external_set = ParSignedDataSet::new();
+    external_set.insert(pubkey, make_par(idx1, &share1));
+    // Crossing the threshold fires SigAgg synchronously through the threshold
+    // subscriber. Because the reconstructed group signature fails eth2
+    // verification, the aggregation errors out and `store_external` surfaces
+    // that error — which is exactly the rejection we are proving. The
+    // load-bearing assertion is that no broadcast reached the beacon node.
+    let store_result = tokio::time::timeout(
+        GUARD,
+        wired.parsigdb.store_external(&attester_duty, &external_set),
+    )
+    .await
+    .expect("store_external did not hang");
+    assert!(
+        store_result.is_err(),
+        "(c-reject) SigAgg verification should fail and propagate through store_external"
+    );
+
+    // Give the threshold → SigAgg → (rejected) pipeline ample time to run, then
+    // assert the broadcaster never submitted: SigAgg's verify_fn rejected the
+    // invalid reconstructed group signature and aborted before broadcast.
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    let posts = count_posts(mock.server(), "/eth/v2/beacon/pool/attestations").await;
+    assert_eq!(
+        posts, 0,
+        "(c-reject) SigAgg must reject the invalid group signature and not broadcast"
+    );
+
+    ct.cancel();
+}
+
+/// (d) `wire_core_workflow` seeds one pubkey-scoped validator cache into the
+/// scheduler's beacon client and the submission client (Charon shares a single
+/// cache across both: `app.go:481-482` and `app.go:598`; the validator API
+/// reuses the same instance). The mock's POST validators endpoint returns only
+/// validators whose pubkey appears in the request-body `ids`, so the unseeded
+/// (empty-pubkey) default cache would resolve zero validators — the regression
+/// this test guards against.
+#[tokio::test]
+async fn wiring_seeds_shared_validator_cache() {
+    let ct = CancellationToken::new();
+    let mock = BeaconMock::builder().build().await.expect("beacon mock");
+    let pubkey = PubKey::new([9u8; PK_LEN]);
+    const V_IDX: u64 = 7;
+    mount_filtered_post_validators(mock.server(), vec![validator_datum(V_IDX, pubkey)]).await;
+
+    let eth2_cl = mock.client().clone();
+    let beacon_client = BeaconNodeClient::new(eth2_cl.clone());
+    let consensus = build_consensus(&ct);
+
+    // `BeaconNodeClient` clones share the cache slot, so the seeding performed
+    // inside `wire_core_workflow` is observable through these probes.
+    let beacon_probe = beacon_client.clone();
+    let inputs = wire_inputs(eth2_cl, beacon_client, pubkey, consensus, 1);
+    let submission_probe = inputs.submission_client.clone();
+
+    let _wired = tokio::time::timeout(GUARD, wire_core_workflow(inputs, ct.clone()))
+        .await
+        .expect("wire did not deadlock")
+        .expect("wire succeeded");
+
+    for (name, probe) in [("beacon", beacon_probe), ("submission", submission_probe)] {
+        let active = tokio::time::timeout(GUARD, probe.active_validators())
+            .await
+            .unwrap_or_else(|_| panic!("(d) {name} client active_validators timed out"))
+            .unwrap_or_else(|e| panic!("(d) {name} client active_validators failed: {e}"));
+        assert_eq!(
+            active.get(&V_IDX),
+            Some(&pubkey_to_eth2(pubkey)),
+            "(d) the {name} client's cache should be seeded with the cluster pubkeys"
+        );
+    }
+
+    ct.cancel();
+}
diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml
index a821544b..ce98d69d 100644
--- a/crates/cli/Cargo.toml
+++ b/crates/cli/Cargo.toml
@@ -27,6 +27,7 @@ pluto-tracing.workspace = true
 pluto-core.workspace = true
 pluto-p2p.workspace = true
 pluto-eth2util.workspace = true
+pluto-featureset.workspace = true
 pluto-k1util.workspace = true
 pluto-ssz.workspace = true
 libp2p.workspace = true
diff --git a/crates/cli/src/commands/run.rs b/crates/cli/src/commands/run.rs
index b400346d..733e318e 100644
--- a/crates/cli/src/commands/run.rs
+++ b/crates/cli/src/commands/run.rs
@@ -14,19 +14,32 @@
 //! The hidden `unsafe run` variant adds test-only flags (e.g. `--p2p-fuzz`) and
 //! must not be used in production.
 //!
-//! Flags are parsed and validated into [`RunConfig`]. The long-running workflow
-//! itself is not yet implemented: invoking the command currently panics at
-//! [`run_workflow`], the seam where the engine will be wired in.
+//! Flags are parsed and validated into [`RunConfig`], bridged into
+//! [`pluto_app::node::AppConfig`] by [`build_app_config`], and the node is
+//! driven via [`pluto_app::node::App::run`] until cancelled.
+//!
+//! Not every accepted flag is honored yet. Correctness-affecting flags with no
+//! implementation (simnet mocks, custom testnets, beacon-node headers, VC TLS,
+//! a preferred consensus protocol, synthetic block proposals) fail fast with a
+//! "not yet supported" error, while observability/availability-only flags
+//! (monitoring/debug addresses, OTLP, proc directory, fallback beacon
+//! endpoints) are ignored with a warning.
 //!
 //! Limitations:
-//! - The run workflow is a stub; the command does not yet perform real duties.
 //! - `--log-format` and `--log-output-path` are accepted but not yet applied
 //!   (console and Loki output only).
 
-use std::{collections::HashMap, path::Path, time::Duration as StdDuration};
+use std::{
+    collections::HashMap,
+    net::{SocketAddr, ToSocketAddrs},
+    path::{Path, PathBuf},
+    sync::Arc,
+    time::Duration as StdDuration,
+};
 
 use libp2p::multiaddr::Protocol;
 use pluto_eth2util::helpers::validate_http_headers;
+use pluto_featureset::{Feature, FeatureSet, FeaturesetError, Status};
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 
@@ -45,6 +58,12 @@ const MAX_GRAFFITI_BYTES_NO_APPEND: usize = 32;
 const MAX_NICKNAME_BYTES: usize = 32;
 /// Grace period for the Loki background task to flush buffered logs on exit.
 const LOKI_FLUSH_TIMEOUT: StdDuration = StdDuration::from_secs(3);
+/// Default `--monitoring-address`; shared by the flag definition and the
+/// ignored-flag warning so they cannot drift.
+const DEFAULT_MONITORING_ADDR: &str = "127.0.0.1:3620";
+/// Default `--simnet-validator-keys-dir`; shared by the flag definition and
+/// the unsupported-flag check so they cannot drift.
+const DEFAULT_SIMNET_KEYS_DIR: &str = ".charon/validator_keys";
 
 /// Arguments for the `run` command.
 ///
@@ -237,7 +256,7 @@ pub struct RunGeneralArgs {
     #[arg(
         long = "simnet-validator-keys-dir",
         env = "CHARON_SIMNET_VALIDATOR_KEYS_DIR",
-        default_value = ".charon/validator_keys",
+        default_value = DEFAULT_SIMNET_KEYS_DIR,
         help = "The directory containing the simnet validator key shares."
     )]
     pub simnet_validator_keys_dir: String,
@@ -402,7 +421,7 @@ pub struct RunDebugMonitoringArgs {
     #[arg(
         long = "monitoring-address",
         env = "CHARON_MONITORING_ADDRESS",
-        default_value = "127.0.0.1:3620",
+        default_value = DEFAULT_MONITORING_ADDR,
         help = "Listening address (ip and port) for the monitoring API (prometheus)."
     )]
     pub monitor_addr: String,
@@ -551,10 +570,6 @@ pub struct RunFeatureArgs {
 }
 
 /// Custom test network configuration.
-//
-// Populated from flags and consumed by the future app entry; until the run
-// workflow is wired (see module docs) these fields are written but not read.
-#[allow(dead_code)]
 #[derive(Debug, Clone, Default)]
 pub struct TestnetConfig {
     /// Name of the custom test network.
@@ -566,14 +581,25 @@ pub struct TestnetConfig {
     /// Genesis timestamp (unix seconds).
     pub genesis_timestamp: i64,
     /// Capella hard fork version.
+    // Accepted for Charon flag parity, inert until custom testnets are
+    // supported (Go's `Network.IsNonZero` excludes it too).
+    #[allow(dead_code)]
     pub capella_hard_fork: String,
 }
 
+impl TestnetConfig {
+    /// Reports whether a custom testnet is fully specified, mirroring Go's
+    /// `eth2util.Network.IsNonZero` (the capella hard fork is excluded there
+    /// too). Charon silently ignores partially-specified testnet flags.
+    fn is_non_zero(&self) -> bool {
+        !self.name.is_empty()
+            && self.chain_id != 0
+            && self.genesis_timestamp != 0
+            && !self.genesis_fork_version_hex.is_empty()
+    }
+}
+
 /// Feature set configuration.
-//
-// Populated from flags and consumed by the future app entry; until the run
-// workflow is wired (see module docs) these fields are written but not read.
-#[allow(dead_code)]
 #[derive(Debug, Clone, Default)]
 pub struct FeatureConfig {
     /// Minimum feature status to enable by default (alpha/beta/stable).
@@ -585,13 +611,9 @@ pub struct FeatureConfig {
 }
 
 /// Configuration for the `run` command — the settings produced from the parsed
-/// flags. This is the object the future app entry consumes (see the seam in
-/// [`run`]); `p2p_fuzz` is the single test-only field, set only via the hidden
-/// `unsafe run` command.
-//
-// The parsed config surface; the run workflow that reads these fields is not
-// yet implemented, so they are populated but not yet consumed.
-#[allow(dead_code)]
+/// flags, consumed by [`run`] (tracing/Loki) and bridged into
+/// [`pluto_app::node::AppConfig`] by [`build_app_config`]; `p2p_fuzz` is the
+/// single test-only field, set only via the hidden `unsafe run` command.
 #[derive(Debug)]
 pub struct RunConfig {
     /// P2P configuration built from [`RunP2PArgs`].
@@ -623,8 +645,14 @@ pub struct RunConfig {
     /// Beacon node submission request timeout.
     pub beacon_node_submit_timeout: StdDuration,
     /// [DISABLED] Jaeger tracing address.
+    // Accepted for Charon flag parity; `RunConfig::try_from` already warns
+    // when set, and the field is never read again.
+    #[allow(dead_code)]
     pub jaeger_addr: String,
     /// [DISABLED] Jaeger tracing service name.
+    // Accepted for Charon flag parity; `RunConfig::try_from` already warns
+    // when set, and the field is never read again.
+    #[allow(dead_code)]
     pub jaeger_service: String,
     /// OTLP gRPC tracing backend address.
     pub otlp_address: String,
@@ -641,6 +669,9 @@ pub struct RunConfig {
     /// Directory containing simnet validator key shares.
     pub simnet_validator_keys_dir: String,
     /// Simnet beacon mock slot duration.
+    // Accepted for Charon flag parity, inert until simnet is supported (the
+    // simnet mode flags themselves fail fast in `check_unsupported_flags`).
+    #[allow(dead_code)]
     pub simnet_slot_duration: StdDuration,
     /// Enables additional synthetic block proposal duties.
     pub synthetic_block_proposals: bool,
@@ -923,10 +954,231 @@ pub async fn run(config: RunConfig, ct: CancellationToken) -> Result<()> {
     result
 }
 
-/// The long-running validator workflow. Not yet implemented: panics via
-/// `unimplemented!` at the seam where the app entry will be wired in.
-async fn run_workflow(_config: RunConfig, _ct: CancellationToken) -> Result<()> {
-    unimplemented!("pluto run")
+/// The long-running validator workflow: bridges the parsed [`RunConfig`] into
+/// [`pluto_app::node::AppConfig`] and drives the node until `ct` fires (signal
+/// handling lives in `main`).
+async fn run_workflow(config: RunConfig, ct: CancellationToken) -> Result<()> {
+    let app_config = build_app_config(config)?;
+    pluto_app::node::App::new(app_config).run(ct).await?;
+    Ok(())
+}
+
+/// Bridges [`RunConfig`] into [`pluto_app::node::AppConfig`].
+///
+/// Flags the app does not support yet either fail fast
+/// ([`check_unsupported_flags`]) or are ignored with a warning
+/// ([`warn_ignored_flags`]); see the module docs for the split.
+fn build_app_config(config: RunConfig) -> Result<pluto_app::node::AppConfig> {
+    check_unsupported_flags(&config)?;
+    warn_ignored_flags(&config);
+
+    let feature_set = build_feature_set(&config.feature)?;
+    let validator_api_addr = parse_validator_api_addr(&config.validator_api_addr)?;
+
+    // Exhaustive destructure: adding a `RunConfig` field without deciding its
+    // bridge behavior fails to compile instead of being silently dropped.
+    let RunConfig {
+        p2p,
+        log: _,
+        feature: _,
+        lock_file,
+        manifest_file: _,
+        no_verify,
+        private_key_file,
+        private_key_locking,
+        monitoring_addr: _,
+        debug_addr: _,
+        validator_api_addr: _,
+        beacon_node_addrs,
+        beacon_node_timeout,
+        beacon_node_submit_timeout,
+        jaeger_addr: _,
+        jaeger_service: _,
+        otlp_address: _,
+        otlp_headers: _,
+        otlp_insecure: _,
+        otlp_service_name: _,
+        simnet_beacon_mock: _,
+        simnet_validator_mock: _,
+        simnet_validator_keys_dir: _,
+        simnet_slot_duration: _,
+        synthetic_block_proposals: _,
+        builder_api,
+        simnet_beacon_mock_fuzz: _,
+        testnet: _,
+        proc_directory: _,
+        consensus_protocol: _,
+        nickname,
+        beacon_node_headers: _,
+        fallback_beacon_node_addrs: _,
+        execution_engine_addr,
+        graffiti,
+        graffiti_disable_client_append,
+        vc_tls_cert_file: _,
+        vc_tls_key_file: _,
+        p2p_fuzz: _,
+    } = config;
+
+    Ok(pluto_app::node::AppConfig {
+        p2p,
+        lock_file: PathBuf::from(lock_file),
+        priv_key_file: PathBuf::from(private_key_file),
+        priv_key_locking: private_key_locking,
+        beacon_node_addrs,
+        beacon_node_timeout,
+        beacon_node_submit_timeout,
+        validator_api_addr,
+        builder_api,
+        nickname,
+        no_verify,
+        // An empty endpoint means "no eth1 verification" on both sides: Charon
+        // hands "" to eth1wrap and Pluto's `EthClient::new("")` is a no-op
+        // client, so `None` is exactly equivalent.
+        eth1_endpoint: (!execution_engine_addr.is_empty()).then_some(execution_engine_addr),
+        // Charon treats a nil graffiti list as "default client graffiti per
+        // validator" (`NewGraffitiBuilder`), matching `AppConfig`'s `None`.
+        graffiti: (!graffiti.is_empty()).then_some(graffiti),
+        graffiti_disable_client_append,
+        feature_set,
+    })
+}
+
+/// Rejects correctness-affecting flags the run workflow does not support yet:
+/// silently ignoring any of these would change duty or operator-facing
+/// behavior (e.g. `--simnet-beacon-mock` permits empty beacon endpoints, which
+/// the real beacon client cannot handle).
+fn check_unsupported_flags(config: &RunConfig) -> Result<()> {
+    let unsupported =
+        |flag: &str| CliError::Other(format!("flag '{flag}' is not yet supported by pluto run"));
+
+    if config.simnet_beacon_mock {
+        return Err(unsupported("--simnet-beacon-mock"));
+    }
+    if config.simnet_validator_mock {
+        return Err(unsupported("--simnet-validator-mock"));
+    }
+    if config.simnet_beacon_mock_fuzz {
+        return Err(unsupported("--simnet-beacon-mock-fuzz"));
+    }
+    if config.simnet_validator_keys_dir != DEFAULT_SIMNET_KEYS_DIR {
+        return Err(unsupported("--simnet-validator-keys-dir"));
+    }
+    if config.synthetic_block_proposals {
+        return Err(unsupported("--synthetic-block-proposals"));
+    }
+    if config.p2p_fuzz {
+        return Err(unsupported("--p2p-fuzz"));
+    }
+    // Partially-specified testnets are ignored, matching Charon's
+    // `IsNonZero`-gated registration.
+    if config.testnet.is_non_zero() {
+        return Err(unsupported("--testnet-*"));
+    }
+    if !config.beacon_node_headers.is_empty() {
+        return Err(unsupported("--beacon-node-headers"));
+    }
+    if !config.consensus_protocol.is_empty() {
+        return Err(unsupported("--consensus-protocol"));
+    }
+    if !config.vc_tls_cert_file.is_empty() || !config.vc_tls_key_file.is_empty() {
+        return Err(unsupported("--vc-tls-cert-file/--vc-tls-key-file"));
+    }
+
+    Ok(())
+}
+
+/// Warns about observability/availability-only flags the run workflow ignores.
+/// Runs after tracing init (see [`run`]) so the warnings reach the subscriber.
+fn warn_ignored_flags(config: &RunConfig) {
+    if config.monitoring_addr != DEFAULT_MONITORING_ADDR {
+        warn!(
+            address = %config.monitoring_addr,
+            "the monitoring API is not yet supported by pluto run; ignoring --monitoring-address"
+        );
+    }
+    if !config.debug_addr.is_empty() {
+        warn!(
+            address = %config.debug_addr,
+            "the debug API is not yet supported by pluto run; ignoring --debug-address"
+        );
+    }
+    if !config.otlp_address.is_empty() {
+        warn!(
+            address = %config.otlp_address,
+            headers = ?config.otlp_headers,
+            insecure = config.otlp_insecure,
+            service = %config.otlp_service_name,
+            "OTLP tracing is not yet supported by pluto run; ignoring the --otlp-* flags"
+        );
+    }
+    if !config.proc_directory.is_empty() {
+        warn!(
+            directory = %config.proc_directory,
+            "stack component detection is not yet supported by pluto run; ignoring --proc-directory"
+        );
+    }
+    if !config.fallback_beacon_node_addrs.is_empty() {
+        warn!(
+            "beacon node failover is not yet supported by pluto run; ignoring --fallback-beacon-node-endpoints"
+        );
+    }
+    // Post-#4130 Charon reads only the lock file and silently ignores an
+    // existing manifest; warn instead so migrated node directories cannot
+    // diverge unnoticed.
+    if Path::new(&config.manifest_file).exists() {
+        warn!(
+            path = %config.manifest_file,
+            "cluster manifest support was removed (Charon #4130); the cluster lock file is authoritative and the manifest file is ignored"
+        );
+    }
+}
+
+/// Resolves the feature-set flags into an injectable [`FeatureSet`].
+///
+/// Matches Charon's `featureset.Init`: an unknown min status is a hard error,
+/// while unknown enabled/disabled feature names are warned about and ignored.
+fn build_feature_set(feature: &FeatureConfig) -> Result<Arc<FeatureSet>> {
+    // `Status::try_from` also parses statuses like "enable" that are not valid
+    // *min* statuses; `FeatureSet::from_config` rejects those below, so both
+    // paths surface Charon's "unknown min status" error.
+    let min_status = Status::try_from(feature.min_status.as_str()).map_err(|_| {
+        FeaturesetError::UnknownMinStatus {
+            min_status: feature.min_status.clone(),
+        }
+    })?;
+
+    let parse_features = |names: &[String], context: &str| -> Vec<Feature> {
+        names
+            .iter()
+            .filter_map(|name| {
+                Feature::try_from(name.as_str())
+                    .inspect_err(|_| warn!(feature = %name, "Ignoring unknown {context} feature"))
+                    .ok()
+            })
+            .collect()
+    };
+
+    let feature_set = FeatureSet::from_config(pluto_featureset::Config {
+        min_status,
+        enabled: parse_features(&feature.enabled, "enabled"),
+        disabled: parse_features(&feature.disabled, "disabled"),
+    })?;
+
+    Ok(Arc::new(feature_set))
+}
+
+/// Parses the validator API listen address. Unlike a plain [`SocketAddr`]
+/// parse, [`ToSocketAddrs`] also resolves hostnames like `localhost:3600`,
+/// which Charon accepts (it defers to Go's `net.Listen`).
+fn parse_validator_api_addr(addr: &str) -> Result<SocketAddr> {
+    addr.to_socket_addrs()
+        .map_err(|err| CliError::Other(format!("invalid validator-api-address {addr:?}: {err}")))?
+        .next()
+        .ok_or_else(|| {
+            CliError::Other(format!(
+                "invalid validator-api-address {addr:?}: no addresses resolved"
+            ))
+        })
 }
 
 #[cfg(test)]
@@ -1465,10 +1717,231 @@ mod tests {
         assert_eq!(config.log.override_env_filter.as_deref(), Some("debug"));
     }
 
+    /// Builds the app config from safe `run` flags.
+    fn app_config(extra: &[&str]) -> Result<pluto_app::node::AppConfig> {
+        build_app_config(parse_run(extra).expect("run args should parse"))
+    }
+
+    /// Returns the `Display` string of the error from a failing `app_config`.
+    fn app_config_err(extra: &[&str]) -> String {
+        app_config(extra)
+            .expect_err("expected bridge error")
+            .to_string()
+    }
+
+    #[test]
+    fn build_app_config_maps_fields() {
+        let config = app_config(&[
+            "--lock-file=/tmp/lock.json",
+            "--private-key-file=/tmp/key",
+            "--private-key-file-lock",
+            "--beacon-node-timeout=5s",
+            "--beacon-node-submit-timeout=6s",
+            "--validator-api-address=127.0.0.1:7600",
+            "--builder-api",
+            "--nickname=node-a",
+            "--no-verify",
+            "--execution-client-rpc-endpoint=http://127.0.0.1:8545",
+            "--graffiti=hello",
+            "--graffiti-disable-client-append",
+            "--p2p-tcp-address=0.0.0.0:9000",
+        ])
+        .expect("bridge should succeed");
+
+        assert_eq!(config.lock_file, PathBuf::from("/tmp/lock.json"));
+        assert_eq!(config.priv_key_file, PathBuf::from("/tmp/key"));
+        assert!(config.priv_key_locking);
+        assert_eq!(
+            config.beacon_node_addrs,
+            vec!["http://beacon.node".to_string()]
+        );
+        assert_eq!(config.beacon_node_timeout, StdDuration::from_secs(5));
+        assert_eq!(config.beacon_node_submit_timeout, StdDuration::from_secs(6));
+        assert_eq!(
+            config.validator_api_addr,
+            "127.0.0.1:7600".parse::<SocketAddr>().expect("socket addr")
+        );
+        assert!(config.builder_api);
+        assert_eq!(config.nickname, "node-a");
+        assert!(config.no_verify);
+        assert_eq!(
+            config.eth1_endpoint.as_deref(),
+            Some("http://127.0.0.1:8545")
+        );
+        assert_eq!(config.graffiti, Some(vec!["hello".to_string()]));
+        assert!(config.graffiti_disable_client_append);
+        assert_eq!(config.p2p.tcp_addrs, vec!["0.0.0.0:9000".to_string()]);
+    }
+
+    #[test]
+    fn build_app_config_defaults_map_to_none() {
+        let config = app_config(&[]).expect("bridge should succeed on defaults");
+
+        // Empty flags mean "unset" on both sides: no eth1 verification and the
+        // default per-validator client graffiti.
+        assert_eq!(config.eth1_endpoint, None);
+        assert_eq!(config.graffiti, None);
+        assert_eq!(
+            config.validator_api_addr,
+            "127.0.0.1:3600".parse::<SocketAddr>().expect("socket addr")
+        );
+        // The default feature set enables stable features only.
+        assert!(config.feature_set.enabled(Feature::EagerDoubleLinear));
+        assert!(!config.feature_set.enabled(Feature::MockAlpha));
+    }
+
+    #[test]
+    fn build_app_config_resolves_hostname_validator_api_addr() {
+        // Charon accepts hostnames here (it defers to Go's net.Listen), so the
+        // bridge must resolve them instead of requiring a literal IP.
+        let config =
+            app_config(&["--validator-api-address=localhost:3600"]).expect("hostname resolves");
+        assert_eq!(config.validator_api_addr.port(), 3600);
+    }
+
+    #[test]
+    fn build_app_config_rejects_unsupported_flags() {
+        for flags in [
+            ["--simnet-beacon-mock"].as_slice(),
+            &["--simnet-validator-mock"],
+            &["--simnet-beacon-mock-fuzz"],
+            &["--simnet-validator-keys-dir=/custom/keys"],
+            &["--synthetic-block-proposals"],
+            &["--consensus-protocol=qbft"],
+            &["--beacon-node-headers=key1=value1"],
+        ] {
+            let err = app_config_err(flags);
+            assert!(
+                err.contains("is not yet supported by pluto run"),
+                "flags {flags:?}: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn build_app_config_rejects_fully_specified_testnet() {
+        let err = app_config_err(&[
+            "--testnet-name=devnet",
+            "--testnet-fork-version=0x10000910",
+            "--testnet-chain-id=1234",
+            "--testnet-genesis-timestamp=42",
+        ]);
+        assert!(err.contains("is not yet supported by pluto run"), "{err}");
+    }
+
+    #[test]
+    fn build_app_config_ignores_partial_testnet() {
+        // Charon registers a custom testnet only when fully specified
+        // (`IsNonZero`); partial flags are silently ignored.
+        app_config(&["--testnet-name=devnet"]).expect("partial testnet should be ignored");
+    }
+
+    #[test]
+    fn build_app_config_rejects_vc_tls() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let cert = dir.path().join("cert.pem");
+        let key = dir.path().join("cert.key");
+        std::fs::write(&cert, b"cert").expect("write cert");
+        std::fs::write(&key, b"key").expect("write key");
+
+        let err = app_config_err(&[
+            "--vc-tls-cert-file",
+            cert.to_str().expect("cert path"),
+            "--vc-tls-key-file",
+            key.to_str().expect("key path"),
+        ]);
+        assert!(err.contains("is not yet supported by pluto run"), "{err}");
+    }
+
+    #[test]
+    fn build_app_config_rejects_p2p_fuzz() {
+        let cli = Cli::try_parse_from([
+            "pluto",
+            "unsafe",
+            "run",
+            "--p2p-fuzz",
+            "--beacon-node-endpoints",
+            "http://beacon.node",
+        ])
+        .expect("unsafe run should parse");
+        let Commands::Unsafe(args) = cli.command else {
+            panic!("expected unsafe command");
+        };
+        let UnsafeCommands::Run(args) = args.command;
+        let config: RunConfig = (*args).try_into().expect("unsafe config should build");
+
+        let err = build_app_config(config)
+            .expect_err("p2p fuzz should be rejected")
+            .to_string();
+        assert!(err.contains("is not yet supported by pluto run"), "{err}");
+    }
+
+    #[test]
+    fn build_app_config_ignores_observability_flags() {
+        // Warn-and-continue: none of these may fail the bridge.
+        app_config(&[
+            "--monitoring-address=127.0.0.1:9620",
+            "--debug-address=127.0.0.1:9630",
+            "--otlp-address=http://otlp.test:4317",
+            "--proc-directory=/proc",
+            "--fallback-beacon-node-endpoints=http://c.node",
+        ])
+        .expect("observability flags are ignored, not rejected");
+    }
+
+    #[test]
+    fn build_app_config_ignores_existing_manifest() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let manifest = dir.path().join("cluster-manifest.pb");
+        std::fs::write(&manifest, b"manifest").expect("write manifest");
+
+        // Post-#4130 semantics: the lock file is authoritative and an existing
+        // manifest is warned about, never fatal.
+        app_config(&["--manifest-file", manifest.to_str().expect("manifest path")])
+            .expect("existing manifest is ignored, not rejected");
+    }
+
+    #[test]
+    fn build_app_config_rejects_unknown_min_status_verbatim() {
+        // Charon's exact error string (featureset.Init).
+        assert_eq!(
+            app_config_err(&["--feature-set=foo"]),
+            "unknown min status: foo"
+        );
+    }
+
+    #[test]
+    fn build_app_config_resolves_feature_set() {
+        // Min status is parsed case-insensitively, like Charon.
+        let config = app_config(&["--feature-set=ALPHA"]).expect("alpha min status");
+        assert!(config.feature_set.enabled(Feature::MockAlpha));
+
+        let config =
+            app_config(&["--feature-set-enable=chain_split_halt"]).expect("explicit enable");
+        assert!(config.feature_set.enabled(Feature::ChainSplitHalt));
+
+        let config =
+            app_config(&["--feature-set-disable=eager_double_linear"]).expect("explicit disable");
+        assert!(!config.feature_set.enabled(Feature::EagerDoubleLinear));
+
+        // Unknown feature names are warned about and ignored, like Charon.
+        app_config(&["--feature-set-enable=not_a_feature"]).expect("unknown feature ignored");
+    }
+
     #[tokio::test]
-    #[should_panic(expected = "not implemented: pluto run")]
-    async fn run_stub_panics_unimplemented() {
-        let config = parse_run(&[]).expect("config should build");
-        let _ = run(config, CancellationToken::new()).await;
+    async fn run_reaches_app_and_fails_on_missing_lock() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let missing_lock = dir.path().join("missing-cluster-lock.json");
+
+        let config = parse_run(&["--lock-file", missing_lock.to_str().expect("lock path")])
+            .expect("config should build");
+        let err = run(config, CancellationToken::new())
+            .await
+            .expect_err("missing lock file should fail the run")
+            .to_string();
+
+        // The bridge reached `App::run`, whose first fallible step (loading
+        // the cluster lock) failed cleanly — no panic, no network access.
+        assert!(err.starts_with("cluster lock:"), "unexpected error: {err}");
     }
 }
diff --git a/crates/cli/src/error.rs b/crates/cli/src/error.rs
index e9cb5350..5e0ee100 100644
--- a/crates/cli/src/error.rs
+++ b/crates/cli/src/error.rs
@@ -126,6 +126,15 @@ pub enum CliError {
     #[error("MEV test error: {0}")]
     MevTest(#[from] MevTestError),
 
+    /// Featureset resolution error.
+    // Verbatim message so `unknown min status: ...` matches Charon's output.
+    #[error("{0}")]
+    Featureset(#[from] pluto_featureset::FeaturesetError),
+
+    /// App (run workflow) error.
+    #[error("{0}")]
+    App(#[from] pluto_app::node::AppError),
+
     /// Generic error with message.
     #[error("{0}")]
     Other(String),
diff --git a/crates/core/src/fetcher/mod.rs b/crates/core/src/fetcher/mod.rs
index 562d0f3c..dbf3b80f 100644
--- a/crates/core/src/fetcher/mod.rs
+++ b/crates/core/src/fetcher/mod.rs
@@ -4,7 +4,7 @@
 
 mod graffiti;
 
-use graffiti::GraffitiBuilder;
+pub use graffiti::{GraffitiBuilder, GraffitiError};
 
 use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc};
 
diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs
index ad6ec3c1..8c4d2ded 100644
--- a/crates/core/src/scheduler.rs
+++ b/crates/core/src/scheduler.rs
@@ -1,7 +1,7 @@
 use std::collections::{HashMap, hash_map::Entry};
 
 use backon::{BackoffBuilder, Retryable};
-use tokio::sync;
+use tokio::{sync, task::JoinHandle};
 use tokio_util::{future::FutureExt, sync::CancellationToken};
 
 use crate::{scheduler::metrics::SCHEDULER_METRICS, types};
@@ -188,12 +188,14 @@ impl SchedulerBuilder {
     /// function.
     ///
     /// The returned [`SchedulerHandle`] can be used to query the scheduler for
-    /// duty definitions.
+    /// duty definitions. The returned [`JoinHandle`] tracks the background
+    /// actor task so callers can supervise its lifecycle (the actor runs until
+    /// `ct` is cancelled).
     pub async fn build(
         self,
         client: pluto_eth2api::BeaconNodeClient,
         ct: CancellationToken,
-    ) -> Result<SchedulerHandle> {
+    ) -> Result<(SchedulerHandle, JoinHandle<()>)> {
         wait_chain_start(&client)
             .with_cancellation_token(&ct)
             .await
@@ -219,9 +221,9 @@ impl SchedulerBuilder {
 
         let (msg_tx, msg_rx) = sync::mpsc::channel(CHANNEL_BUFFER_SIZE);
         let handle = SchedulerHandle { sender: msg_tx };
-        tokio::spawn(actor.run(slot_rx, msg_rx, self.reorg_rx, ct));
+        let task = tokio::spawn(actor.run(slot_rx, msg_rx, self.reorg_rx, ct));
 
-        Ok(handle)
+        Ok((handle, task))
     }
 }
 
diff --git a/crates/dkg/src/node.rs b/crates/dkg/src/node.rs
index 3d6f9777..f6d55b0e 100644
--- a/crates/dkg/src/node.rs
+++ b/crates/dkg/src/node.rs
@@ -9,7 +9,7 @@ use crate::{
     exchanger::{SIG_DEPOSIT_DATA, SigType},
     frostp2p, sync,
 };
-use libp2p::{Multiaddr, multiaddr::Protocol, relay, swarm::NetworkBehaviour};
+use libp2p::{relay, swarm::NetworkBehaviour};
 use pluto_core::{
     types::{Duty, DutyType},
     version,
@@ -60,7 +60,7 @@ pub(crate) async fn setup_p2p(
 
     verify_p2p_key(peers, &key)?;
 
-    let relay_addrs = relay_addrs_for_resolution(&conf.p2p.relays);
+    let relay_addrs = bootnode::relay_addrs_for_resolution(&conf.p2p.relays);
     let relays = bootnode::new_relays(ct, &relay_addrs, &hex::encode(&def_hash)).await?;
 
     let conn_gater = gater::ConnGater::new_conn_gater(peer_ids.clone(), relays.clone());
@@ -161,50 +161,3 @@ pub(crate) async fn setup_p2p(
 
     Ok((node, handlers))
 }
-
-fn relay_addrs_for_resolution(relays: &[Multiaddr]) -> Vec<String> {
-    relays.iter().map(relay_addr_for_resolution).collect()
-}
-
-fn relay_addr_for_resolution(relay: &Multiaddr) -> String {
-    let mut scheme = None;
-    let mut host = None;
-    let mut port = None;
-
-    for protocol in relay.iter() {
-        match protocol {
-            Protocol::Http => scheme = Some("http"),
-            Protocol::Https => scheme = Some("https"),
-            Protocol::Dns(name)
-            | Protocol::Dns4(name)
-            | Protocol::Dns6(name)
-            | Protocol::Dnsaddr(name)
-                if host.is_none() =>
-            {
-                host = Some(name.to_string());
-            }
-            Protocol::Ip4(ip) if host.is_none() => {
-                host = Some(ip.to_string());
-            }
-            Protocol::Ip6(ip) if host.is_none() => {
-                host = Some(format!("[{ip}]"));
-            }
-            Protocol::Tcp(tcp_port) => port = Some(tcp_port),
-            _ => {}
-        }
-    }
-
-    if let (Some(scheme), Some(host)) = (scheme, host) {
-        let default_port = match scheme {
-            "https" => 443,
-            _ => 80,
-        };
-
-        return match port {
-            Some(port) if port != default_port => format!("{scheme}://{host}:{port}"),
-            _ => format!("{scheme}://{host}"),
-        };
-    }
-
-    relay.to_string()
-}
diff --git a/crates/featureset/src/lib.rs b/crates/featureset/src/lib.rs
index dacb6e2a..d455a30a 100644
--- a/crates/featureset/src/lib.rs
+++ b/crates/featureset/src/lib.rs
@@ -44,17 +44,48 @@ pub enum Status {
     Enable = i32::MAX as isize,
 }
 
-impl fmt::Display for Status {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+impl Status {
+    /// Returns the string representation of the status.
+    pub fn as_str(self) -> &'static str {
         match self {
-            Status::Disable => write!(f, "disable"),
-            Status::Alpha => write!(f, "alpha"),
-            Status::Beta => write!(f, "beta"),
-            Status::Stable => write!(f, "stable"),
-            Status::Sentinel => write!(f, "sentinel"),
-            Status::Enable => write!(f, "enable"),
+            Status::Disable => "disable",
+            Status::Alpha => "alpha",
+            Status::Beta => "beta",
+            Status::Stable => "stable",
+            Status::Sentinel => "sentinel",
+            Status::Enable => "enable",
         }
     }
+
+    /// Returns all statuses.
+    pub fn all() -> &'static [Status] {
+        &[
+            Status::Disable,
+            Status::Alpha,
+            Status::Beta,
+            Status::Stable,
+            Status::Sentinel,
+            Status::Enable,
+        ]
+    }
+}
+
+impl fmt::Display for Status {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.as_str())
+    }
+}
+
+impl std::convert::TryFrom<&str> for Status {
+    type Error = String;
+
+    fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
+        Status::all()
+            .iter()
+            .find(|status| value.eq_ignore_ascii_case(status.as_str()))
+            .copied()
+            .ok_or_else(|| format!("unknown status: {}", value))
+    }
 }
 
 /// A feature being rolled out.
@@ -331,6 +362,27 @@ mod tests {
         assert_eq!(Status::Enable.to_string(), "enable");
     }
 
+    #[test]
+    fn status_try_from_round_trips_display() {
+        for status in Status::all() {
+            assert_eq!(Status::try_from(status.to_string().as_str()), Ok(*status));
+        }
+    }
+
+    #[test]
+    fn status_try_from_is_case_insensitive() {
+        assert_eq!(Status::try_from("ALPHA"), Ok(Status::Alpha));
+        assert_eq!(Status::try_from("Stable"), Ok(Status::Stable));
+    }
+
+    #[test]
+    fn status_try_from_rejects_unknown() {
+        assert_eq!(
+            Status::try_from("foo"),
+            Err("unknown status: foo".to_string())
+        );
+    }
+
     #[test]
     fn custom_enabled_all() {
         let featureset = FeatureSet::new();
diff --git a/crates/p2p/src/bootnode.rs b/crates/p2p/src/bootnode.rs
index 2084083a..e07affab 100644
--- a/crates/p2p/src/bootnode.rs
+++ b/crates/p2p/src/bootnode.rs
@@ -3,7 +3,7 @@
 use std::time::Duration;
 
 use backon::Retryable;
-use libp2p::Multiaddr;
+use libp2p::{Multiaddr, multiaddr::Protocol};
 use pluto_eth2util::enr::Record;
 use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
@@ -390,3 +390,89 @@ fn addr_info_from_p2p_addr(addr: &Multiaddr) -> std::result::Result<AddrInfo, Pe
 
     infos.pop().ok_or(PeerError::MissingPeerIdInMultiaddr)
 }
+
+/// Converts configured relay multiaddrs into the string form [`new_relays`]
+/// expects: HTTP(S) relay endpoints become `scheme://host[:port]` URLs (for
+/// background resolution), anything else falls back to the multiaddr string.
+pub fn relay_addrs_for_resolution(relays: &[Multiaddr]) -> Vec<String> {
+    relays.iter().map(relay_addr_for_resolution).collect()
+}
+
+/// Converts one relay multiaddr into the string form [`new_relays`] expects.
+///
+/// The default port for the scheme (80/443) is omitted from the URL.
+pub fn relay_addr_for_resolution(relay: &Multiaddr) -> String {
+    let mut scheme = None;
+    let mut host = None;
+    let mut port = None;
+
+    for protocol in relay.iter() {
+        match protocol {
+            Protocol::Http => scheme = Some("http"),
+            Protocol::Https => scheme = Some("https"),
+            Protocol::Dns(name)
+            | Protocol::Dns4(name)
+            | Protocol::Dns6(name)
+            | Protocol::Dnsaddr(name)
+                if host.is_none() =>
+            {
+                host = Some(name.to_string());
+            }
+            Protocol::Ip4(ip) if host.is_none() => {
+                host = Some(ip.to_string());
+            }
+            Protocol::Ip6(ip) if host.is_none() => {
+                host = Some(format!("[{ip}]"));
+            }
+            Protocol::Tcp(tcp_port) => port = Some(tcp_port),
+            _ => {}
+        }
+    }
+
+    if let (Some(scheme), Some(host)) = (scheme, host) {
+        let default_port = match scheme {
+            "https" => 443,
+            _ => 80,
+        };
+
+        return match port {
+            Some(port) if port != default_port => format!("{scheme}://{host}:{port}"),
+            _ => format!("{scheme}://{host}"),
+        };
+    }
+
+    relay.to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn relay_addr_resolution_forms() {
+        let cases = [
+            (
+                "/dns/relay.example.org/tcp/443/https",
+                "https://relay.example.org",
+            ),
+            (
+                "/dns/relay.example.org/tcp/8443/https",
+                "https://relay.example.org:8443",
+            ),
+            (
+                "/dns4/relay.example.org/tcp/80/http",
+                "http://relay.example.org",
+            ),
+            ("/ip4/10.0.0.1/tcp/3640/http", "http://10.0.0.1:3640"),
+            ("/ip6/::1/tcp/443/https", "https://[::1]"),
+        ];
+        for (addr, expected) in cases {
+            let addr: Multiaddr = addr.parse().expect("valid multiaddr");
+            assert_eq!(relay_addr_for_resolution(&addr), expected);
+        }
+
+        // Non-HTTP multiaddrs fall back to the multiaddr string.
+        let plain: Multiaddr = "/ip4/10.0.0.1/tcp/3610".parse().expect("valid multiaddr");
+        assert_eq!(relay_addr_for_resolution(&plain), plain.to_string());
+    }
+}