diff --git a/aptos-core/consensus/Cargo.toml b/aptos-core/consensus/Cargo.toml
index 1c14df69..f74d4fd0 100644
--- a/aptos-core/consensus/Cargo.toml
+++ b/aptos-core/consensus/Cargo.toml
@@ -88,6 +88,7 @@ fuzzing = [
     "aptos-safety-rules/testing",
 ]
 failpoints = ["fail/failpoints"]
+byzantine-test = []
 randomness_disabled = ["gaptos/randomness_disabled"]
 
 [package.metadata.cargo-machete]
diff --git a/aptos-core/consensus/src/byzantine_test.rs b/aptos-core/consensus/src/byzantine_test.rs
new file mode 100644
index 00000000..0f9d1ba4
--- /dev/null
+++ b/aptos-core/consensus/src/byzantine_test.rs
@@ -0,0 +1,285 @@
+// Copyright © Aptos Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+//! Test-image-only control and evidence for protocol-aware Byzantine faults.
+
+use anyhow::{bail, Context};
+use once_cell::sync::Lazy;
+use serde::{Deserialize, Serialize};
+use std::{
+    env, fs,
+    path::{Path, PathBuf},
+    sync::Mutex,
+};
+
+const CONTROL_PATH_ENV: &str = "BFT_BYZANTINE_CONTROL_PATH";
+const EVIDENCE_PATH_ENV: &str = "BFT_BYZANTINE_EVIDENCE_PATH";
+const FIXTURE_ENABLED_ENV: &str = "BFT_BYZANTINE_FIXTURE_ENABLED";
+const MAX_CONTROL_BYTES: u64 = 64 * 1024;
+
+#[derive(Clone, Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct ControlDocument {
+    schema_version: u64,
+    active: bool,
+    fault_id: String,
+    behavior: String,
+    node_id: String,
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct EquivocationRequest {
+    pub fault_id: String,
+    pub node_id: String,
+}
+
+#[derive(Default)]
+struct RuntimeState {
+    fault_id: Option<String>,
+    event_count: u64,
+    claimed_epoch_round: Option<(u64, u64)>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct EvidenceDocument<'a> {
+    schema_version: u64,
+    fault_id: &'a str,
+    behavior: &'static str,
+    node_id: &'a str,
+    protocol_effect: ProtocolEffect<'a>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ProtocolEffect<'a> {
+    observed: bool,
+    behavior: &'static str,
+    event_count: u64,
+    epoch: u64,
+    round: u64,
+    distinct_message_count: u64,
+    recipient_group_count: u64,
+    first_message_id: &'a str,
+    second_message_id: &'a str,
+    first_recipient_count: usize,
+    second_recipient_count: usize,
+}
+
+static RUNTIME_STATE: Lazy<Mutex<RuntimeState>> = Lazy::new(|| Mutex::new(RuntimeState::default()));
+
+fn valid_token(value: &str) -> bool {
+    !value.is_empty() &&
+        value.len() <= 128 &&
+        value.bytes().enumerate().all(|(index, byte)| match byte {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' => true,
+            b'.' | b'_' | b':' | b'-' => index > 0,
+            _ => false,
+        })
+}
+
+fn configured_path(variable: &str) -> anyhow::Result<Option<PathBuf>> {
+    let Some(value) = env::var_os(variable) else {
+        return Ok(None);
+    };
+    let path = PathBuf::from(value);
+    if !path.is_absolute() {
+        bail!("{variable} must be an absolute path");
+    }
+    Ok(Some(path))
+}
+
+fn read_control() -> anyhow::Result<Option<ControlDocument>> {
+    if env::var(FIXTURE_ENABLED_ENV).as_deref() != Ok("1") {
+        return Ok(None);
+    }
+    let Some(path) = configured_path(CONTROL_PATH_ENV)? else {
+        return Ok(None);
+    };
+    let metadata = match fs::metadata(&path) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(error).context("reading Byzantine control metadata"),
+    };
+    if !metadata.is_file() {
+        bail!("Byzantine control path is not a regular file");
+    }
+    if metadata.len() > MAX_CONTROL_BYTES {
+        bail!("Byzantine control document exceeds 64 KiB");
+    }
+
+    let document: ControlDocument =
+        serde_json::from_slice(&fs::read(&path).context("reading Byzantine control document")?)
+            .context("parsing Byzantine control document")?;
+    if document.schema_version != 1 {
+        bail!("unsupported Byzantine control schema version");
+    }
+    if !valid_token(&document.fault_id) || !valid_token(&document.node_id) {
+        bail!("Byzantine control document contains an invalid identifier");
+    }
+    if document.behavior != "equivocation" {
+        bail!("unsupported Byzantine behavior: {}", document.behavior);
+    }
+    Ok(Some(document))
+}
+
+pub(crate) fn claim_equivocation(
+    epoch: u64,
+    round: u64,
+) -> anyhow::Result<Option<EquivocationRequest>> {
+    let Some(control) = read_control()? else {
+        return Ok(None);
+    };
+    if !control.active {
+        return Ok(None);
+    }
+
+    let mut state = RUNTIME_STATE
+        .lock()
+        .map_err(|_| anyhow::anyhow!("Byzantine runtime state lock is poisoned"))?;
+    if state.fault_id.as_deref() != Some(&control.fault_id) {
+        state.fault_id = Some(control.fault_id.clone());
+        state.event_count = 0;
+        state.claimed_epoch_round = None;
+    }
+    if state.claimed_epoch_round == Some((epoch, round)) {
+        return Ok(None);
+    }
+    state.claimed_epoch_round = Some((epoch, round));
+
+    Ok(Some(EquivocationRequest { fault_id: control.fault_id, node_id: control.node_id }))
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn record_equivocation(
+    request: &EquivocationRequest,
+    epoch: u64,
+    round: u64,
+    first_message_id: &str,
+    second_message_id: &str,
+    first_recipient_count: usize,
+    second_recipient_count: usize,
+) -> anyhow::Result<()> {
+    let Some(path) = configured_path(EVIDENCE_PATH_ENV)? else {
+        bail!("{EVIDENCE_PATH_ENV} is not configured");
+    };
+    if first_message_id == second_message_id {
+        bail!("equivocation messages must have distinct IDs");
+    }
+    if first_recipient_count == 0 || second_recipient_count == 0 {
+        bail!("equivocation recipient groups must be non-empty");
+    }
+
+    let event_count = {
+        let mut state = RUNTIME_STATE
+            .lock()
+            .map_err(|_| anyhow::anyhow!("Byzantine runtime state lock is poisoned"))?;
+        if state.fault_id.as_deref() != Some(&request.fault_id) {
+            bail!("Byzantine fault changed before evidence was recorded");
+        }
+        state.event_count =
+            state.event_count.checked_add(1).context("Byzantine event counter overflow")?;
+        state.event_count
+    };
+
+    let document = EvidenceDocument {
+        schema_version: 1,
+        fault_id: &request.fault_id,
+        behavior: "equivocation",
+        node_id: &request.node_id,
+        protocol_effect: ProtocolEffect {
+            observed: true,
+            behavior: "equivocation",
+            event_count,
+            epoch,
+            round,
+            distinct_message_count: 2,
+            recipient_group_count: 2,
+            first_message_id,
+            second_message_id,
+            first_recipient_count,
+            second_recipient_count,
+        },
+    };
+    write_json_atomically(&path, &document)
+}
+
+fn write_json_atomically(path: &Path, document: &EvidenceDocument<'_>) -> anyhow::Result<()> {
+    let parent = path.parent().context("Byzantine evidence path has no parent directory")?;
+    if !parent.is_dir() {
+        bail!("Byzantine evidence parent directory does not exist");
+    }
+    let file_name = path
+        .file_name()
+        .and_then(|name| name.to_str())
+        .context("Byzantine evidence path has an invalid file name")?;
+    let temporary = parent.join(format!(".{file_name}.{}.tmp", std::process::id()));
+    let bytes = serde_json::to_vec(document).context("serializing Byzantine evidence")?;
+    fs::write(&temporary, bytes).context("writing temporary Byzantine evidence")?;
+    fs::rename(&temporary, path).context("publishing Byzantine evidence")?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex as TestMutex;
+    use tempfile::TempDir;
+
+    static ENV_LOCK: TestMutex<()> = TestMutex::new(());
+
+    fn write_control(directory: &TempDir, active: bool, fault_id: &str) -> PathBuf {
+        let path = directory.path().join("control.json");
+        fs::write(
+            &path,
+            serde_json::json!({
+                "schemaVersion": 1,
+                "active": active,
+                "faultId": fault_id,
+                "behavior": "equivocation",
+                "nodeId": "validator-1"
+            })
+            .to_string(),
+        )
+        .unwrap();
+        path
+    }
+
+    #[test]
+    fn claims_once_per_round_and_records_typed_evidence() {
+        let _guard = ENV_LOCK.lock().unwrap();
+        let directory = TempDir::new().unwrap();
+        let control = write_control(&directory, true, "fault-1");
+        let evidence = directory.path().join("evidence.json");
+        env::set_var(FIXTURE_ENABLED_ENV, "1");
+        env::set_var(CONTROL_PATH_ENV, &control);
+        env::set_var(EVIDENCE_PATH_ENV, &evidence);
+
+        let request = claim_equivocation(7, 11).unwrap().unwrap();
+        assert!(claim_equivocation(7, 11).unwrap().is_none());
+        record_equivocation(&request, 7, 11, "0x01", "0x02", 2, 2).unwrap();
+
+        let document: serde_json::Value =
+            serde_json::from_slice(&fs::read(evidence).unwrap()).unwrap();
+        assert_eq!(document["faultId"], "fault-1");
+        assert_eq!(document["protocolEffect"]["observed"], true);
+        assert_eq!(document["protocolEffect"]["eventCount"], 1);
+        assert_eq!(document["protocolEffect"]["distinctMessageCount"], 2);
+
+        env::remove_var(CONTROL_PATH_ENV);
+        env::remove_var(EVIDENCE_PATH_ENV);
+        env::remove_var(FIXTURE_ENABLED_ENV);
+    }
+
+    #[test]
+    fn inactive_control_does_not_claim() {
+        let _guard = ENV_LOCK.lock().unwrap();
+        let directory = TempDir::new().unwrap();
+        let control = write_control(&directory, false, "fault-2");
+        env::set_var(FIXTURE_ENABLED_ENV, "1");
+        env::set_var(CONTROL_PATH_ENV, control);
+        assert!(claim_equivocation(9, 13).unwrap().is_none());
+        env::remove_var(CONTROL_PATH_ENV);
+        env::remove_var(FIXTURE_ENABLED_ENV);
+    }
+}
diff --git a/aptos-core/consensus/src/lib.rs b/aptos-core/consensus/src/lib.rs
index d1197f0f..c1babe56 100644
--- a/aptos-core/consensus/src/lib.rs
+++ b/aptos-core/consensus/src/lib.rs
@@ -22,6 +22,8 @@ extern crate scopeguard;
 extern crate core;
 
 mod block_storage;
+#[cfg(feature = "byzantine-test")]
+mod byzantine_test;
 pub mod consensusdb;
 mod dag;
 mod epoch_manager;
diff --git a/aptos-core/consensus/src/network.rs b/aptos-core/consensus/src/network.rs
index 7998d78b..b9b29154 100644
--- a/aptos-core/consensus/src/network.rs
+++ b/aptos-core/consensus/src/network.rs
@@ -454,7 +454,7 @@ impl NetworkSender {
         self.send(msg, recipients).await
     }
 
-    #[cfg(feature = "failpoints")]
+    #[cfg(any(feature = "failpoints", feature = "byzantine-test"))]
     pub async fn send_proposal(&self, proposal_msg: ProposalMsg, recipients: Vec<Author>) {
         fail_point!("consensus::send::proposal", |_| ());
         let msg = ConsensusMsg::ProposalMsg(Box::new(proposal_msg));
diff --git a/aptos-core/consensus/src/round_manager.rs b/aptos-core/consensus/src/round_manager.rs
index d9d01fa1..27e9318e 100644
--- a/aptos-core/consensus/src/round_manager.rs
+++ b/aptos-core/consensus/src/round_manager.rs
@@ -27,6 +27,8 @@ use crate::{
     util::is_vtxn_expected,
 };
 use anyhow::{bail, ensure, Context};
+#[cfg(feature = "byzantine-test")]
+use aptos_consensus_types::block_data::BlockData;
 use aptos_consensus_types::{
     block::Block,
     block_data::BlockType,
@@ -460,10 +462,33 @@ impl RoundManager {
             sync_info,
             network.clone(),
             proposal_generator,
-            safety_rules,
+            safety_rules.clone(),
             proposer_election,
         )
         .await?;
+        #[cfg(feature = "byzantine-test")]
+        match Self::attempt_to_equivocate(
+            epoch_state.clone(),
+            network.clone(),
+            safety_rules,
+            &proposal_msg,
+        )
+        .await
+        {
+            Ok(true) => {
+                counters::PROPOSALS_COUNT.inc();
+                return Ok(());
+            }
+            Ok(false) => {}
+            Err(error) => {
+                warn!(
+                    error = ?error,
+                    epoch = epoch,
+                    round = proposal_msg.proposal().round(),
+                    "Byzantine test hook failed; broadcasting the original proposal"
+                );
+            }
+        }
         #[cfg(feature = "failpoints")]
         {
             if Self::check_whether_to_inject_reconfiguration_error() {
@@ -596,6 +621,102 @@ impl RoundManager {
         Ok(ProposalMsg::new(signed_proposal, sync_info))
     }
 
+    #[cfg(feature = "byzantine-test")]
+    fn conflicting_proposal_data(original: &BlockData) -> anyhow::Result<BlockData> {
+        let author = original.author().context("proposal has no author")?;
+        let payload = original.payload().cloned().context("proposal has no payload")?;
+        let failed_authors = original.failed_authors().cloned().unwrap_or_default();
+        let timestamp =
+            original.timestamp_usecs().checked_add(1).context("proposal timestamp overflow")?;
+        let round = original.round();
+        let quorum_cert = original.quorum_cert().clone();
+
+        match original.block_type() {
+            BlockType::Proposal { .. } => Ok(BlockData::new_proposal(
+                payload,
+                author,
+                failed_authors,
+                round,
+                timestamp,
+                quorum_cert,
+            )),
+            BlockType::ProposalExt(_) => Ok(BlockData::new_proposal_ext(
+                original.validator_txns().cloned().unwrap_or_default(),
+                payload,
+                author,
+                failed_authors,
+                round,
+                timestamp,
+                quorum_cert,
+            )),
+            block_type => bail!("cannot equivocate proposal block type {block_type:?}"),
+        }
+    }
+
+    #[cfg(feature = "byzantine-test")]
+    async fn attempt_to_equivocate(
+        epoch_state: Arc<EpochState>,
+        network: Arc<NetworkSender>,
+        safety_rules: Arc<Mutex<MetricsSafetyRules>>,
+        proposal_msg: &ProposalMsg,
+    ) -> anyhow::Result<bool> {
+        let epoch = proposal_msg.epoch();
+        let round = proposal_msg.proposal().round();
+        let Some(request) = crate::byzantine_test::claim_equivocation(epoch, round)? else {
+            return Ok(false);
+        };
+
+        let conflicting_data =
+            Self::conflicting_proposal_data(proposal_msg.proposal().block_data())?;
+        let signature = safety_rules.lock().sign_proposal(&conflicting_data)?;
+        let conflicting = ProposalMsg::new(
+            Block::new_proposal_from_block_data_and_signature(conflicting_data, signature),
+            proposal_msg.sync_info().clone(),
+        );
+        ensure!(
+            proposal_msg.proposal().id() != conflicting.proposal().id(),
+            "equivocation proposals must have distinct IDs"
+        );
+
+        let mut recipients: Vec<_> =
+            epoch_state.verifier.get_ordered_account_addresses_iter().collect();
+        ensure!(recipients.len() >= 2, "equivocation requires at least two validators");
+        recipients.sort_unstable();
+        let second_group = recipients.split_off(recipients.len() / 2);
+        let first_group = recipients;
+        ensure!(
+            !first_group.is_empty() && !second_group.is_empty(),
+            "equivocation recipient groups must be non-empty"
+        );
+
+        network.send_proposal(proposal_msg.clone(), first_group.clone()).await;
+        network.send_proposal(conflicting.clone(), second_group.clone()).await;
+
+        let first_id = proposal_msg.proposal().id().to_string();
+        let second_id = conflicting.proposal().id().to_string();
+        crate::byzantine_test::record_equivocation(
+            &request,
+            epoch,
+            round,
+            &first_id,
+            &second_id,
+            first_group.len(),
+            second_group.len(),
+        )?;
+        warn!(
+            fault_id = request.fault_id,
+            node_id = request.node_id,
+            epoch = epoch,
+            round = round,
+            first_message_id = first_id,
+            second_message_id = second_id,
+            first_recipient_count = first_group.len(),
+            second_recipient_count = second_group.len(),
+            "Byzantine test hook emitted conflicting signed proposals"
+        );
+        Ok(true)
+    }
+
     /// Process the proposal message:
     /// 1. ensure after processing sync info, we're at the same round as the proposal
     /// 2. execute and decide whether to vote for the proposal
diff --git a/aptos-core/consensus/src/round_manager_test.rs b/aptos-core/consensus/src/round_manager_test.rs
index 3de1bede..36edb951 100644
--- a/aptos-core/consensus/src/round_manager_test.rs
+++ b/aptos-core/consensus/src/round_manager_test.rs
@@ -123,6 +123,30 @@ pub struct NodeSetup {
     onchain_jwk_consensus_config: OnChainJWKConsensusConfig,
 }
 
+#[cfg(feature = "byzantine-test")]
+#[test]
+fn byzantine_conflicting_proposal_preserves_round_and_changes_block_id() {
+    let signer = ValidatorSigner::from_int(1);
+    let original = Block::new_proposal(
+        Payload::empty(false, true),
+        1,
+        10,
+        certificate_for_genesis(),
+        &signer,
+        Vec::new(),
+    )
+    .unwrap();
+
+    let conflicting_data = RoundManager::conflicting_proposal_data(original.block_data()).unwrap();
+    let conflicting = Block::new_proposal_from_block_data(conflicting_data, &signer).unwrap();
+
+    assert_eq!(original.epoch(), conflicting.epoch());
+    assert_eq!(original.round(), conflicting.round());
+    assert_eq!(original.parent_id(), conflicting.parent_id());
+    assert_eq!(original.timestamp_usecs() + 1, conflicting.timestamp_usecs());
+    assert_ne!(original.id(), conflicting.id());
+}
+
 impl NodeSetup {
     fn create_round_state(time_service: Arc<dyn TimeService>) -> RoundState {
         let base_timeout = Duration::new(60, 0);
diff --git a/bin/bench/src/main.rs b/bin/bench/src/main.rs
index 849a95be..8feadf28 100644
--- a/bin/bench/src/main.rs
+++ b/bin/bench/src/main.rs
@@ -39,7 +39,8 @@ impl TestConsensusLayer {
                 },
                 EmptyTxPool::boxed(),
             )
-            .await,
+            .await
+            .expect("failed to initialize consensus engine"),
         }
     }
 
diff --git a/bin/gravity_node/Cargo.toml b/bin/gravity_node/Cargo.toml
index 670b3588..a4010426 100644
--- a/bin/gravity_node/Cargo.toml
+++ b/bin/gravity_node/Cargo.toml
@@ -13,6 +13,7 @@ rust-version.workspace = true
 [features]
 # Forward feature to gaptos so `-p gravity_node --features randomness_disabled` works
 randomness_disabled = ["gaptos/randomness_disabled"]
+byzantine-test = ["api/byzantine-test"]
 default = []
 
 [dependencies]
diff --git a/bin/gravity_node/src/main.rs b/bin/gravity_node/src/main.rs
index fa5e5b5e..376f3bed 100644
--- a/bin/gravity_node/src/main.rs
+++ b/bin/gravity_node/src/main.rs
@@ -345,7 +345,7 @@ fn main() {
     // `_engine` owns tokio Runtimes; it must be returned out of `block_on` so it
     // drops in this sync context — dropping a Runtime inside an async context
     // panics in tokio's blocking-pool shutdown.
-    let (coordinator_result, _engine) = rt.block_on(async move {
+    let (coordinator_result, _engine, failed_init_coordinator) = rt.block_on(async move {
         let datadir = datadir_rx.await.expect("datadir should be sent");
         let client = Arc::new(RethCli::new(consensus_args, txn_cache, shutdown_rx_cli).await);
         let chain_id = client.chain_id();
@@ -372,8 +372,7 @@ fn main() {
                     panic!("failed to set global relayer");
                 }
             }
-            _engine = Some(
-                ConsensusEngine::init(
+            match ConsensusEngine::init(
                     ConsensusEngineArgs {
                         node_config: gcei_config,
                         chain_id,
@@ -384,8 +383,19 @@ fn main() {
                     },
                     pool,
                 )
-                .await,
-            );
+                .await
+            {
+                Ok(engine) => _engine = Some(engine),
+                Err(error) => {
+                    tracing::error!("Failed to initialize consensus engine: {error:#}");
+                    let _ = shutdown_tx.send(());
+                    return (
+                        Err(format!("failed to initialize consensus engine: {error:#}")),
+                        _engine,
+                        Some(coordinator),
+                    );
+                }
+            }
         }
         coordinator.send_execution_args().await;
         let result = coordinator.run().await;
@@ -395,12 +405,17 @@ fn main() {
         }
 
         info!("Main shutdown complete");
-        (result, _engine)
+        (result, _engine, None)
     });
     drop(rt);
     drop(_engine);
 
-    if let Err(err) = reth_thread.join() {
+    // Keep the execution-args sender alive until Reth has shut down. Its receiver
+    // currently unwraps channel closure, so dropping the failed coordinator first
+    // would turn an expected initialization error into a background-task panic.
+    let reth_result = reth_thread.join();
+    drop(failed_init_coordinator);
+    if let Err(err) = reth_result {
         eprintln!("Reth thread panicked: {err:?}");
         std::process::exit(1);
     }
diff --git a/crates/api/Cargo.toml b/crates/api/Cargo.toml
index 35b2564c..4a9fa708 100644
--- a/crates/api/Cargo.toml
+++ b/crates/api/Cargo.toml
@@ -51,4 +51,5 @@ bytes = { workspace = true }
 [features]
 default = []
 failpoints = ["fail/failpoints", "aptos-consensus/failpoints", "aptos-mempool/failpoints"]
+byzantine-test = ["aptos-consensus/byzantine-test"]
 jemalloc-profiling = ["tikv-jemallocator/profiling", "tikv-jemalloc-sys/profiling"]
diff --git a/crates/api/src/consensus_api.rs b/crates/api/src/consensus_api.rs
index 81c38e73..0f518c7e 100644
--- a/crates/api/src/consensus_api.rs
+++ b/crates/api/src/consensus_api.rs
@@ -114,7 +114,10 @@ pub struct ConsensusEngineArgs {
 }
 
 impl ConsensusEngine {
-    pub async fn init(args: ConsensusEngineArgs, pool: Box<dyn TxPool>) -> Arc<Self> {
+    pub async fn init(
+        args: ConsensusEngineArgs,
+        pool: Box<dyn TxPool>,
+    ) -> anyhow::Result<Arc<Self>> {
         let ConsensusEngineArgs { node_config, chain_id, latest_block_number, config_storage } =
             args;
         // Setup panic handler
@@ -316,9 +319,12 @@ impl ConsensusEngine {
             );
             runtimes.push(jwk_consensus_runtime);
         }
-        init_block_buffer_manager(&consensus_db, latest_block_number)
-            .await
-            .expect("failed to initialize BlockBufferManager");
+        if let Err(error) = init_block_buffer_manager(&consensus_db, latest_block_number).await {
+            for runtime in runtimes.drain(..) {
+                runtime.shutdown_background();
+            }
+            return Err(error);
+        }
         let mut args = ConsensusAdapterArgs::new(consensus_db.clone());
         let (consensus_runtime, _, _) = start_consensus(
             &node_config,
@@ -371,6 +377,6 @@ impl ConsensusEngine {
         // process new round should be after init retƒh hash
         info!("pass latest_block_number: {:?} to event_subscription_service", latest_block_number);
         let _ = event_subscription_service.lock().await.notify_initial_configs(latest_block_number);
-        arc_consensus_engine
+        Ok(arc_consensus_engine)
     }
 }
diff --git a/docker/gravity_node/Dockerfile b/docker/gravity_node/Dockerfile
index 0462318f..15cba8a3 100644
--- a/docker/gravity_node/Dockerfile
+++ b/docker/gravity_node/Dockerfile
@@ -1,6 +1,6 @@
 # syntax=docker/dockerfile:1.7
 #
-# Two build targets:
+# Runtime build targets:
 #
 #   1. `runtime` (default) — compile gravity_node and gravity_cli from source
 #      inside the container's `builder` stage:
@@ -17,8 +17,18 @@
 #                     -t gravity_node:rel \
 #                     -f docker/gravity_node/Dockerfile .
 #
-# Stage pruning means `--target runtime-host-binary` does not enter the
-# `builder` stage — the cargo build is genuinely skipped, not just cached.
+# The opt-in `runtime-storage-test` and
+# `runtime-host-binary-storage-test` targets add the destructive BFT storage
+# fixture and its child-process supervisor. They are never selected by the
+# default or release build.
+#
+# The opt-in `runtime-byzantine-test` and
+# `runtime-host-binary-byzantine-test` targets add the protocol-aware
+# equivocation hook. The source target compiles gravity_node with the
+# `byzantine-test` feature; neither target is used by default or release builds.
+#
+# Stage pruning means host-binary targets do not enter the `builder` stage;
+# the cargo build is genuinely skipped, not just cached.
 
 # ─── Stage 1: builder (used only by the `runtime` target) ────────────
 FROM rust:1.93-slim-bookworm AS builder
@@ -52,6 +62,19 @@ RUN --mount=type=cache,target=/build/target,id=gravity-target \
     && cp "target/${CARGO_PROFILE}/gravity_node" /out/gravity_node \
     && cp "target/${CARGO_PROFILE}/gravity_cli" /out/gravity_cli
 
+# Reuse the normal build cache, then replace only gravity_node with a binary
+# that contains the test-only Byzantine control path.
+FROM builder AS builder-byzantine-test
+
+ARG CARGO_PROFILE=release
+
+RUN --mount=type=cache,target=/build/target,id=gravity-target \
+    --mount=type=cache,target=/usr/local/cargo/registry,id=gravity-cargo-registry \
+    --mount=type=cache,target=/usr/local/cargo/git,id=gravity-cargo-git \
+    cargo build --bin gravity_node --profile "${CARGO_PROFILE}" \
+        --features byzantine-test \
+    && cp "target/${CARGO_PROFILE}/gravity_node" /out/gravity_node
+
 # ─── Stage 2: runtime-base (shared apt + user setup + entrypoint) ────
 # Common to both leaf targets. The binaries are added in the leaves
 # so we can pick the source (builder vs build context) per target.
@@ -94,14 +117,81 @@ USER gravity
 ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/entrypoint.sh"]
 CMD ["node"]
 
-# ─── Stage 3b: runtime (default target) — copy from in-container build ─
-# Put last so `docker build .` (no --target) picks this and the
-# from-source build is the default behaviour, matching prior versions.
-FROM runtime-base AS runtime
+# Opt-in destructive storage fixture. This target is never selected by a
+# default or release build and must be paired with an explicitly disposable
+# data volume at runtime.
+FROM runtime-host-binary AS runtime-host-binary-storage-test
+
+USER root
+COPY docker/gravity_node/bft-node-supervisor.sh /usr/local/bin/bft-node-supervisor
+COPY docker/gravity_node/bft-storage /usr/local/bin/bft-storage
+RUN chmod 755 /usr/local/bin/bft-node-supervisor /usr/local/bin/bft-storage \
+    && mkdir -p /run/bft-node \
+    && chown gravity:gravity /run/bft-node
+
+USER gravity
+ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/bft-node-supervisor"]
+CMD ["node"]
+
+# Opt-in protocol-aware Byzantine fixture for a host-built binary. The caller
+# must supply a gravity_node compiled with `--features byzantine-test`.
+FROM runtime-host-binary AS runtime-host-binary-byzantine-test
+
+USER root
+COPY docker/gravity_node/bft-byzantine /usr/local/bin/bft-byzantine
+RUN chmod 755 /usr/local/bin/bft-byzantine \
+    && mkdir -p /run/bft-node \
+    && chown gravity:gravity /run/bft-node
+
+ENV BFT_BYZANTINE_CONTROL_PATH=/run/bft-node/byzantine-control.json \
+    BFT_BYZANTINE_EVIDENCE_PATH=/run/bft-node/byzantine-evidence.json
+
+USER gravity
+ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/entrypoint.sh"]
+CMD ["node"]
+
+# ─── Stage 3b: source-runtime-base — copy from in-container build ───
+FROM runtime-base AS source-runtime-base
 
 COPY --from=builder --chmod=755 /out/gravity_node /usr/local/bin/gravity_node
 COPY --from=builder --chmod=755 /out/gravity_cli /usr/local/bin/gravity_cli
 
+# Source-built equivalent of the host-binary storage-test target.
+FROM source-runtime-base AS runtime-storage-test
+
+COPY docker/gravity_node/bft-node-supervisor.sh /usr/local/bin/bft-node-supervisor
+COPY docker/gravity_node/bft-storage /usr/local/bin/bft-storage
+RUN chmod 755 /usr/local/bin/bft-node-supervisor /usr/local/bin/bft-storage \
+    && mkdir -p /run/bft-node \
+    && chown gravity:gravity /run/bft-node
+
+USER gravity
+ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/bft-node-supervisor"]
+CMD ["node"]
+
+# Source-built protocol-aware Byzantine fixture. The binary comes from the
+# dedicated feature-enabled builder stage, not from the normal runtime stage.
+FROM runtime-base AS runtime-byzantine-test
+
+COPY --from=builder-byzantine-test --chmod=755 /out/gravity_node /usr/local/bin/gravity_node
+COPY --from=builder-byzantine-test --chmod=755 /out/gravity_cli /usr/local/bin/gravity_cli
+COPY docker/gravity_node/bft-byzantine /usr/local/bin/bft-byzantine
+RUN chmod 755 /usr/local/bin/bft-byzantine \
+    && mkdir -p /run/bft-node \
+    && chown gravity:gravity /run/bft-node
+
+ENV BFT_BYZANTINE_CONTROL_PATH=/run/bft-node/byzantine-control.json \
+    BFT_BYZANTINE_EVIDENCE_PATH=/run/bft-node/byzantine-evidence.json
+
+USER gravity
+ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/entrypoint.sh"]
+CMD ["node"]
+
+# ─── Stage 3c: runtime (default target) ─────────────────────────────
+# Put last so `docker build .` (no --target) picks this and the
+# from-source build remains the default behaviour.
+FROM source-runtime-base AS runtime
+
 USER gravity
 ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/entrypoint.sh"]
 CMD ["node"]
diff --git a/docker/gravity_node/README.md b/docker/gravity_node/README.md
index 632ee8b5..8b1feb52 100644
--- a/docker/gravity_node/README.md
+++ b/docker/gravity_node/README.md
@@ -11,8 +11,13 @@ image tag — configuration and chain state persist across restarts.
 
 | File | Purpose |
 |---|---|
-| `Dockerfile` | Multi-stage build (`rust:1.93-slim` → `ubuntu:24.04`). Includes `gravity_node`, `gravity_cli`, and `curl`. Non-root (uid `10001`). `tini` as PID 1. |
+| `Dockerfile` | Multi-stage build (`rust:1.93-slim` → `ubuntu:24.04`). Includes normal runtime targets plus explicit storage and Byzantine test targets. Non-root (uid `10001`). `tini` as PID 1. |
 | `entrypoint.sh` | Reads `reth_config.json` (same schema as `cluster/templates/reth_config.json.tpl`) and `exec`s `gravity_node node` in the foreground. |
+| `bft-node-supervisor.sh` | Test-target-only child-process supervisor that keeps the container available while a corrupted node process is stopped. |
+| `bft-storage` | Test-target-only WAL/database backup, truncation, evidence, and restoration hook for bft-jepsen. |
+| `bft-byzantine` | Test-target-only protocol-aware equivocation control and evidence hook for bft-jepsen. |
+| `test-storage-fixture.sh` | Isolated image-level acceptance for storage corruption, container restart, exact restoration, and recovery. |
+| `test-byzantine-fixture.sh` | Isolated image-level acceptance for Byzantine hook authorization, capabilities, evidence, and recovery. |
 | `docker-compose.yaml` | Single-node deployment. Intended for one host running one validator. |
 | `docker-compose.cluster.yaml` | 4 validators + 1 VFN on one host. For end-to-end image verification against `cluster/` artifacts. |
 | `render-cluster-config.sh` | Renders the 5-node config set from `cluster/output/` + `cluster/templates/`. |
@@ -116,6 +121,123 @@ Default port range used by this topology: `6180–6183`, `6190–6195`,
 Ensure nothing else on the host — including `cluster`'s host-mode deployment
 — is holding these ports before starting.
 
+## Disposable BFT storage-fault image
+
+The normal `runtime` and `runtime-host-binary` images do not contain the
+destructive storage hook. Build one of the explicit test targets for a cluster
+whose data volumes can be discarded:
+
+```bash
+# Build gravity_node from source.
+docker build --target runtime-storage-test \
+  -t gravity_node:storage-test \
+  -f docker/gravity_node/Dockerfile .
+
+# Or package binaries already built on the host.
+docker build --target runtime-host-binary-storage-test \
+  --build-arg HOST_BINARY=target/quick-release/gravity_node \
+  --build-arg HOST_CLI_BINARY=target/quick-release/gravity_cli \
+  -t gravity_node:storage-test \
+  -f docker/gravity_node/Dockerfile .
+```
+
+Each target container must use a fresh disposable data volume and receive all
+of these environment variables:
+
+```yaml
+environment:
+  BFT_STORAGE_FIXTURE_ENABLED: "1"
+  BFT_STORAGE_DISPOSABLE_DATA: I_UNDERSTAND_THIS_DATA_WILL_BE_DESTROYED
+  BFT_STORAGE_DATA_ROOT: /gravity/data
+  BFT_STORAGE_WAL_PATH: /gravity/data/data/consensus_db
+  BFT_STORAGE_DATABASE_PATH: /gravity/data/data/reth/db
+  BFT_STORAGE_DATABASE_MUTATION_FILE: state/CURRENT
+```
+
+For `wal`, the hook selects the newest non-empty `*.log` below the configured
+component path. For `database`, the mutation file is relative to the component
+path. Before truncation, the stopped component is archived in full and its
+SHA-256 is recorded. Healing verifies the archive, replaces the mutated
+component, checks the original file size and hash, and requires the node child
+process to remain stable. A successful heal removes the large backup archive
+but retains the small JSON evidence under `/gravity/data/.bft-storage/states`.
+An active fault also leaves `/gravity/data/.bft-storage-active`, so replacing
+or restarting the container cannot turn a corrupted child into a restart
+storm; the new supervisor waits for `heal` while keeping `docker exec` usable.
+
+The backup is stored on the same disposable volume. The hook refuses injection
+unless free space is at least the component size plus 256 MiB; adjust only with
+`BFT_STORAGE_BACKUP_RESERVE_MIB`. Never enable this target against an existing
+operator or long-running test volume.
+
+The bft-jepsen controller invokes the hook as root through `docker exec`:
+
+```text
+/usr/local/bin/bft-storage inject <wal|database> <fault-id> <node-id>
+/usr/local/bin/bft-storage heal <fault-id>
+/usr/local/bin/bft-storage read <fault-id> <wal|database> <node-id>
+```
+
+Run the isolated image-level acceptance test with fake node binaries and a
+temporary named volume:
+
+```bash
+bash docker/gravity_node/test-storage-fixture.sh
+```
+
+## Protocol-aware Byzantine test image
+
+The normal runtime images do not contain the Byzantine hook, and the normal
+binary does not compile the conflicting-message path. Build the explicit
+source target for a disposable BFT test cluster:
+
+```bash
+docker build --target runtime-byzantine-test \
+  -t gravity_node:byzantine-test \
+  -f docker/gravity_node/Dockerfile .
+```
+
+To package host binaries, compile `gravity_node` with the test feature first,
+then select the matching host-binary target:
+
+```bash
+RUSTFLAGS="--cfg tokio_unstable" \
+  cargo build --bin gravity_node --profile quick-release \
+    --features byzantine-test
+
+docker build --target runtime-host-binary-byzantine-test \
+  --build-arg HOST_BINARY=target/quick-release/gravity_node \
+  --build-arg HOST_CLI_BINARY=target/quick-release/gravity_cli \
+  -t gravity_node:byzantine-test \
+  -f docker/gravity_node/Dockerfile .
+```
+
+Each instrumented validator must opt in at runtime:
+
+```yaml
+environment:
+  BFT_BYZANTINE_FIXTURE_ENABLED: "1"
+```
+
+The current Gravity target advertises only `equivocation`. While the selected
+validator is proposer, it signs two different proposals for the same
+epoch/round and sends them to two non-overlapping validator groups. The node
+writes typed protocol evidence under `/run/bft-node`; the hook exposes only
+normalized counters to bft-jepsen:
+
+```text
+/usr/local/bin/bft-byzantine inject equivocation <fault-id> <node-id>
+/usr/local/bin/bft-byzantine read <fault-id>
+/usr/local/bin/bft-byzantine heal <fault-id>
+```
+
+`double-sign` and `twin` are rejected until they have independent, real
+protocol implementations. Run the isolated image/hook contract test with:
+
+```bash
+bash docker/gravity_node/test-byzantine-fixture.sh
+```
+
 ## Single-node deployment
 
 ```bash
diff --git a/docker/gravity_node/bft-byzantine b/docker/gravity_node/bft-byzantine
new file mode 100644
index 00000000..37725bf5
--- /dev/null
+++ b/docker/gravity_node/bft-byzantine
@@ -0,0 +1,228 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+die() {
+    echo "bft-byzantine: $*" >&2
+    exit 1
+}
+
+validate_token() {
+    local label="$1"
+    local value="$2"
+
+    [[ "${value}" =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ ]] \
+        || die "${label} contains unsupported characters"
+}
+
+canonical_state_file() {
+    local raw_path="$1"
+    local parent
+    local name
+
+    [[ "${raw_path}" == /* ]] || die "state file path must be absolute"
+    parent="$(readlink -f -- "$(dirname -- "${raw_path}")")" \
+        || die "cannot resolve state file parent"
+    name="$(basename -- "${raw_path}")"
+    [[ -n "${name}" && "${name}" != "." && "${name}" != ".." ]] \
+        || die "invalid state file name"
+    printf '%s/%s\n' "${parent}" "${name}"
+}
+
+initialize() {
+    [[ "${BFT_BYZANTINE_FIXTURE_ENABLED:-}" == "1" ]] \
+        || die "fixture is disabled; set BFT_BYZANTINE_FIXTURE_ENABLED=1"
+    [[ "$(id -u)" == "0" ]] \
+        || die "the hook must run as root (docker exec --user 0)"
+    command -v jq >/dev/null 2>&1 || die "required command not found: jq"
+
+    STATE_ROOT="${BFT_BYZANTINE_STATE_ROOT:-/run/bft-node}"
+    [[ "${STATE_ROOT}" == /* ]] || die "state root must be an absolute path"
+    mkdir -p "${STATE_ROOT}"
+    STATE_ROOT="$(readlink -f -- "${STATE_ROOT}")"
+    CONTROL_PATH="${BFT_BYZANTINE_CONTROL_PATH:-${STATE_ROOT}/byzantine-control.json}"
+    EVIDENCE_PATH="${BFT_BYZANTINE_EVIDENCE_PATH:-${STATE_ROOT}/byzantine-evidence.json}"
+    LOCK_DIR="${STATE_ROOT}/byzantine.lock"
+
+    CONTROL_PATH="$(canonical_state_file "${CONTROL_PATH}")"
+    EVIDENCE_PATH="$(canonical_state_file "${EVIDENCE_PATH}")"
+    for path in "${CONTROL_PATH}" "${EVIDENCE_PATH}"; do
+        [[ "${path}" == "${STATE_ROOT}"/* ]] \
+            || die "control and evidence paths must be below the state root"
+    done
+}
+
+acquire_lock() {
+    local owner=""
+
+    if mkdir "${LOCK_DIR}" 2>/dev/null; then
+        printf '%s\n' "$$" > "${LOCK_DIR}/pid"
+        trap release_lock EXIT
+        return
+    fi
+    [[ -f "${LOCK_DIR}/pid" ]] && owner="$(<"${LOCK_DIR}/pid")"
+    if [[ "${owner}" =~ ^[0-9]+$ ]] && kill -0 "${owner}" 2>/dev/null; then
+        die "another Byzantine hook operation is active (PID ${owner})"
+    fi
+    rm -rf -- "${LOCK_DIR}"
+    mkdir "${LOCK_DIR}" 2>/dev/null \
+        || die "another Byzantine hook operation acquired the lock"
+    printf '%s\n' "$$" > "${LOCK_DIR}/pid"
+    trap release_lock EXIT
+}
+
+release_lock() {
+    rm -rf -- "${LOCK_DIR:-/run/bft-node/byzantine.lock}"
+}
+
+validate_control() {
+    local path="$1"
+
+    jq -e '
+      type == "object" and
+      .schemaVersion == 1 and
+      (.active | type == "boolean") and
+      (.faultId | type == "string") and
+      .behavior == "equivocation" and
+      (.nodeId | type == "string")
+    ' "${path}" >/dev/null || die "invalid Byzantine control state"
+}
+
+write_control() {
+    local active="$1"
+    local fault_id="$2"
+    local node_id="$3"
+    local temporary="${CONTROL_PATH}.$$"
+
+    jq -cn \
+        --argjson active "${active}" \
+        --arg faultId "${fault_id}" \
+        --arg nodeId "${node_id}" \
+        '{schemaVersion: 1,
+          active: $active,
+          faultId: $faultId,
+          behavior: "equivocation",
+          nodeId: $nodeId}' > "${temporary}"
+    chmod 644 "${temporary}"
+    mv -f -- "${temporary}" "${CONTROL_PATH}"
+}
+
+inject_fault() {
+    local behavior="$1"
+    local fault_id="$2"
+    local node_id="$3"
+
+    [[ "${behavior}" == "equivocation" ]] \
+        || die "unsupported behavior: ${behavior}; capabilities: equivocation"
+    validate_token fault-id "${fault_id}"
+    validate_token node-id "${node_id}"
+    acquire_lock
+
+    if [[ -f "${CONTROL_PATH}" ]]; then
+        validate_control "${CONTROL_PATH}"
+        if [[ "$(jq -r '.active' "${CONTROL_PATH}")" == "true" ]]; then
+            die "another Byzantine fault is active: $(jq -r '.faultId' "${CONTROL_PATH}")"
+        fi
+    fi
+    rm -f -- "${EVIDENCE_PATH}"
+    write_control true "${fault_id}" "${node_id}"
+}
+
+heal_fault() {
+    local fault_id="$1"
+    local node_id="unknown"
+
+    validate_token fault-id "${fault_id}"
+    acquire_lock
+    if [[ -f "${CONTROL_PATH}" ]]; then
+        validate_control "${CONTROL_PATH}"
+        local current_fault
+        current_fault="$(jq -r '.faultId' "${CONTROL_PATH}")"
+        [[ "${current_fault}" == "${fault_id}" ]] \
+            || die "control state belongs to fault ${current_fault}"
+        node_id="$(jq -r '.nodeId' "${CONTROL_PATH}")"
+    fi
+    write_control false "${fault_id}" "${node_id}"
+}
+
+read_state() {
+    local fault_id="$1"
+    local active=false
+    local behavior="equivocation"
+    local node_id=""
+    local effect
+
+    validate_token fault-id "${fault_id}"
+    if [[ -f "${CONTROL_PATH}" ]]; then
+        validate_control "${CONTROL_PATH}"
+        local current_fault
+        current_fault="$(jq -r '.faultId' "${CONTROL_PATH}")"
+        if [[ "$(jq -r '.active' "${CONTROL_PATH}")" == "true" ]]; then
+            [[ "${current_fault}" == "${fault_id}" ]] \
+                || die "active control state belongs to fault ${current_fault}"
+        fi
+        active="$(jq -r '.active' "${CONTROL_PATH}")"
+        behavior="$(jq -r '.behavior' "${CONTROL_PATH}")"
+        node_id="$(jq -r '.nodeId' "${CONTROL_PATH}")"
+    fi
+
+    effect="$(jq -cn \
+        --arg behavior "${behavior}" \
+        '{observed: false,
+          behavior: $behavior,
+          eventCount: 0,
+          distinctMessageCount: 0,
+          recipientGroupCount: 0}')"
+    if [[ -f "${EVIDENCE_PATH}" ]] \
+        && jq -e --arg faultId "${fault_id}" '
+             type == "object" and
+             .schemaVersion == 1 and
+             .faultId == $faultId and
+             .behavior == "equivocation" and
+             (.protocolEffect | type == "object")
+           ' "${EVIDENCE_PATH}" >/dev/null; then
+        effect="$(jq -c '.protocolEffect' "${EVIDENCE_PATH}")"
+    fi
+
+    jq -cn \
+        --argjson active "${active}" \
+        --arg faultId "${fault_id}" \
+        --arg behavior "${behavior}" \
+        --arg nodeId "${node_id}" \
+        --argjson protocolEffect "${effect}" \
+        '{schemaVersion: 1,
+          active: $active,
+          faultId: $faultId,
+          behavior: $behavior,
+          nodeId: $nodeId,
+          capabilities: ["equivocation"],
+          protocolEffect: $protocolEffect}'
+}
+
+usage() {
+    cat >&2 <<'EOF'
+usage:
+  bft-byzantine inject equivocation <fault-id> <node-id>
+  bft-byzantine heal <fault-id>
+  bft-byzantine read <fault-id>
+EOF
+    exit 2
+}
+
+initialize
+case "${1:-}" in
+    inject)
+        [[ "$#" == 4 ]] || usage
+        inject_fault "$2" "$3" "$4"
+        ;;
+    heal)
+        [[ "$#" == 2 ]] || usage
+        heal_fault "$2"
+        ;;
+    read)
+        [[ "$#" == 2 ]] || usage
+        read_state "$2"
+        ;;
+    *)
+        usage
+        ;;
+esac
diff --git a/docker/gravity_node/bft-node-supervisor.sh b/docker/gravity_node/bft-node-supervisor.sh
new file mode 100755
index 00000000..bef7e728
--- /dev/null
+++ b/docker/gravity_node/bft-node-supervisor.sh
@@ -0,0 +1,102 @@
+#!/usr/bin/env bash
+set -uo pipefail
+
+ENTRYPOINT="${BFT_NODE_ENTRYPOINT:-/usr/local/bin/entrypoint.sh}"
+RUN_DIR="${BFT_NODE_SUPERVISOR_DIR:-/run/bft-node}"
+NODE_PID_FILE="${RUN_DIR}/node.pid"
+SUPERVISOR_PID_FILE="${RUN_DIR}/supervisor.pid"
+MAINTENANCE_FILE="${RUN_DIR}/maintenance"
+FAULT_FILE="${RUN_DIR}/storage-fault"
+DATA_ROOT="${BFT_STORAGE_DATA_ROOT:-/gravity/data}"
+PERSISTENT_FAULT_FILE="${DATA_ROOT}/.bft-storage-active"
+RESTART_DELAY="${BFT_NODE_RESTART_DELAY_SECONDS:-1}"
+
+if [[ "${1:-}" != "node" ]]; then
+    exec "${ENTRYPOINT}" "$@"
+fi
+
+if [[ ! "${RESTART_DELAY}" =~ ^[0-9]+([.][0-9]+)?$ ]]; then
+    echo "bft-node-supervisor: invalid restart delay: ${RESTART_DELAY}" >&2
+    exit 2
+fi
+
+mkdir -p "${RUN_DIR}"
+umask 0027
+printf '%s\n' "$$" > "${SUPERVISOR_PID_FILE}"
+rm -f "${NODE_PID_FILE}"
+
+child_pid=""
+shutting_down=0
+
+remove_pid_file() {
+    local recorded=""
+
+    if [[ -f "${NODE_PID_FILE}" ]]; then
+        recorded="$(<"${NODE_PID_FILE}")"
+    fi
+    if [[ -z "${child_pid}" || "${recorded}" == "${child_pid}" ]]; then
+        rm -f "${NODE_PID_FILE}"
+    fi
+}
+
+request_shutdown() {
+    shutting_down=1
+    if [[ "${child_pid}" =~ ^[0-9]+$ ]] && kill -0 "${child_pid}" 2>/dev/null; then
+        kill -TERM "${child_pid}" 2>/dev/null || true
+    fi
+}
+
+cleanup() {
+    remove_pid_file
+    rm -f "${SUPERVISOR_PID_FILE}"
+}
+
+trap request_shutdown TERM INT
+trap cleanup EXIT
+
+if [[ -e "${PERSISTENT_FAULT_FILE}" ]]; then
+    echo "bft-node-supervisor: persistent storage fault found; waiting for heal" >&2
+    while [[ -e "${PERSISTENT_FAULT_FILE}" ]] && (( shutting_down == 0 )); do
+        sleep 0.2
+    done
+fi
+
+while (( shutting_down == 0 )); do
+    while [[ -e "${MAINTENANCE_FILE}" ]] && (( shutting_down == 0 )); do
+        sleep 0.2
+    done
+    (( shutting_down == 0 )) || break
+
+    "${ENTRYPOINT}" "$@" &
+    child_pid=$!
+    pid_tmp="${NODE_PID_FILE}.$$"
+    printf '%s\n' "${child_pid}" > "${pid_tmp}"
+    mv -f "${pid_tmp}" "${NODE_PID_FILE}"
+
+    wait "${child_pid}"
+    child_status=$?
+
+    if (( shutting_down != 0 )) && kill -0 "${child_pid}" 2>/dev/null; then
+        wait "${child_pid}" 2>/dev/null || true
+    fi
+    remove_pid_file
+    child_pid=""
+
+    (( shutting_down == 0 )) || break
+
+    if [[ -e "${FAULT_FILE}" || -e "${PERSISTENT_FAULT_FILE}" ]]; then
+        echo "bft-node-supervisor: node exited during an active storage fault; waiting for heal" >&2
+        while [[ -e "${FAULT_FILE}" || -e "${PERSISTENT_FAULT_FILE}" ]] \
+              && (( shutting_down == 0 )); do
+            sleep 0.2
+        done
+        continue
+    fi
+
+    if [[ ! -e "${MAINTENANCE_FILE}" ]]; then
+        echo "bft-node-supervisor: node exited with status ${child_status}; restarting" >&2
+        sleep "${RESTART_DELAY}"
+    fi
+done
+
+exit 0
diff --git a/docker/gravity_node/bft-storage b/docker/gravity_node/bft-storage
new file mode 100755
index 00000000..8faff362
--- /dev/null
+++ b/docker/gravity_node/bft-storage
@@ -0,0 +1,688 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+readonly DISPOSABLE_CONFIRMATION="I_UNDERSTAND_THIS_DATA_WILL_BE_DESTROYED"
+
+die() {
+    echo "bft-storage: $*" >&2
+    exit 1
+}
+
+require_command() {
+    command -v "$1" >/dev/null 2>&1 || die "required command not found: $1"
+}
+
+validate_token() {
+    local label="$1"
+    local value="$2"
+
+    [[ "${value}" =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ ]] \
+        || die "${label} contains unsupported characters"
+}
+
+validate_non_negative_integer() {
+    local label="$1"
+    local value="$2"
+
+    [[ "${value}" =~ ^[0-9]+$ ]] || die "${label} must be a non-negative integer"
+}
+
+require_enabled() {
+    [[ "${BFT_STORAGE_FIXTURE_ENABLED:-}" == "1" ]] \
+        || die "fixture is disabled; set BFT_STORAGE_FIXTURE_ENABLED=1"
+    [[ "${BFT_STORAGE_DISPOSABLE_DATA:-}" == "${DISPOSABLE_CONFIRMATION}" ]] \
+        || die "disposable-data confirmation is missing"
+    [[ "$(id -u)" == "0" ]] \
+        || die "the hook must run as root (docker exec --user 0)"
+}
+
+path_is_within() {
+    local path="$1"
+    local root="$2"
+
+    [[ "${path}" == "${root}"/* ]]
+}
+
+canonical_existing_dir() {
+    local label="$1"
+    local raw_path="$2"
+    local path
+
+    [[ -n "${raw_path}" ]] || die "${label} is not configured"
+    [[ -d "${raw_path}" ]] || die "${label} is not a directory: ${raw_path}"
+    path="$(readlink -f -- "${raw_path}")"
+    [[ -n "${path}" && -d "${path}" ]] || die "cannot resolve ${label}: ${raw_path}"
+    printf '%s\n' "${path}"
+}
+
+initialize() {
+    require_enabled
+    require_command awk
+    require_command df
+    require_command du
+    require_command find
+    require_command jq
+    require_command readlink
+    require_command sha256sum
+    require_command stat
+    require_command tar
+    require_command truncate
+
+    DATA_ROOT="$(canonical_existing_dir BFT_STORAGE_DATA_ROOT \
+        "${BFT_STORAGE_DATA_ROOT:-/gravity/data}")"
+    RUN_DIR="${BFT_NODE_SUPERVISOR_DIR:-/run/bft-node}"
+    [[ -d "${RUN_DIR}" ]] || die "node supervisor directory is missing: ${RUN_DIR}"
+    RUN_DIR="$(readlink -f -- "${RUN_DIR}")"
+
+    STATE_ROOT_RAW="${BFT_STORAGE_STATE_ROOT:-${DATA_ROOT}/.bft-storage}"
+    mkdir -p "${STATE_ROOT_RAW}"
+    STATE_ROOT="$(readlink -f -- "${STATE_ROOT_RAW}")"
+    path_is_within "${STATE_ROOT}" "${DATA_ROOT}" \
+        || die "BFT_STORAGE_STATE_ROOT must be below BFT_STORAGE_DATA_ROOT"
+
+    STATE_DIR="${STATE_ROOT}/states"
+    BACKUP_ROOT="${STATE_ROOT}/backups"
+    LOCK_DIR="${RUN_DIR}/bft-storage.lock"
+    NODE_PID_FILE="${RUN_DIR}/node.pid"
+    SUPERVISOR_PID_FILE="${RUN_DIR}/supervisor.pid"
+    MAINTENANCE_FILE="${RUN_DIR}/maintenance"
+    FAULT_FILE="${RUN_DIR}/storage-fault"
+    PERSISTENT_FAULT_FILE="${DATA_ROOT}/.bft-storage-active"
+    mkdir -p "${STATE_DIR}" "${BACKUP_ROOT}"
+    chmod 700 "${STATE_ROOT}" "${STATE_DIR}" "${BACKUP_ROOT}"
+}
+
+read_numeric_pid() {
+    local path="$1"
+    local pid=""
+
+    [[ -f "${path}" ]] || return 1
+    pid="$(<"${path}")"
+    [[ "${pid}" =~ ^[0-9]+$ ]] || return 1
+    printf '%s\n' "${pid}"
+}
+
+ensure_supervisor() {
+    local pid
+    local command_line
+
+    pid="$(read_numeric_pid "${SUPERVISOR_PID_FILE}")" \
+        || die "node supervisor PID is unavailable"
+    kill -0 "${pid}" 2>/dev/null || die "node supervisor is not running"
+    command_line="$(tr '\0' ' ' < "/proc/${pid}/cmdline")"
+    [[ "${command_line}" == *"bft-node-supervisor"* ]] \
+        || die "PID ${pid} is not the storage-test node supervisor"
+    SUPERVISOR_PID="${pid}"
+}
+
+acquire_lock() {
+    local owner=""
+
+    if mkdir "${LOCK_DIR}" 2>/dev/null; then
+        printf '%s\n' "$$" > "${LOCK_DIR}/pid"
+        trap release_lock EXIT
+        return
+    fi
+
+    if [[ -f "${LOCK_DIR}/pid" ]]; then
+        owner="$(<"${LOCK_DIR}/pid")"
+    fi
+    if [[ ! "${owner}" =~ ^[0-9]+$ ]]; then
+        die "storage lock exists without a valid owner; inspect ${LOCK_DIR}"
+    fi
+    if kill -0 "${owner}" 2>/dev/null; then
+        die "another storage operation is active (PID ${owner})"
+    fi
+
+    rm -rf -- "${LOCK_DIR}"
+    mkdir "${LOCK_DIR}" 2>/dev/null \
+        || die "another storage operation acquired the lock"
+    printf '%s\n' "$$" > "${LOCK_DIR}/pid"
+    trap release_lock EXIT
+}
+
+release_lock() {
+    rm -rf -- "${LOCK_DIR:-/run/bft-node/bft-storage.lock}"
+}
+
+state_file_for() {
+    local fault_id="$1"
+    printf '%s/%s.json\n' "${STATE_DIR}" "${fault_id}"
+}
+
+write_state() {
+    local state_file="$1"
+    local active="$2"
+    local fault_id="$3"
+    local component="$4"
+    local node_id="$5"
+    local backup_ready="$6"
+    local mutation_applied="$7"
+    local restored="$8"
+    local target_path="$9"
+    local mutation_path="${10}"
+    local backup_path="${11}"
+    local backup_sha256="${12}"
+    local original_size="${13}"
+    local original_sha256="${14}"
+    local node_running="${15}"
+    local tmp="${state_file}.$$"
+
+    jq -cn \
+        --argjson active "${active}" \
+        --arg faultId "${fault_id}" \
+        --arg component "${component}" \
+        --arg nodeId "${node_id}" \
+        --argjson backupReady "${backup_ready}" \
+        --argjson mutationApplied "${mutation_applied}" \
+        --argjson restored "${restored}" \
+        --arg targetPath "${target_path}" \
+        --arg mutationPath "${mutation_path}" \
+        --arg backupPath "${backup_path}" \
+        --arg backupSha256 "${backup_sha256}" \
+        --argjson originalSizeBytes "${original_size}" \
+        --arg originalSha256 "${original_sha256}" \
+        --argjson nodeRunning "${node_running}" \
+        '{schemaVersion: 1,
+          active: $active,
+          faultId: $faultId,
+          component: $component,
+          nodeId: $nodeId,
+          backupReady: $backupReady,
+          mutationApplied: $mutationApplied,
+          restored: $restored,
+          targetPath: $targetPath,
+          mutationPath: $mutationPath,
+          backupPath: $backupPath,
+          backupSha256: $backupSha256,
+          originalSizeBytes: $originalSizeBytes,
+          originalSha256: $originalSha256,
+          nodeRunning: $nodeRunning}' > "${tmp}"
+    chmod 600 "${tmp}"
+    mv -f "${tmp}" "${state_file}"
+}
+
+validate_state_document() {
+    local state_file="$1"
+
+    jq -e '
+      type == "object" and
+      .schemaVersion == 1 and
+      (.active | type == "boolean") and
+      (.backupReady | type == "boolean") and
+      (.mutationApplied | type == "boolean") and
+      (.restored | type == "boolean")
+    ' "${state_file}" >/dev/null \
+        || die "invalid storage state: ${state_file}"
+}
+
+latest_non_empty_log() {
+    local target="$1"
+    local candidate
+    local candidate_mtime
+    local selected=""
+    local selected_mtime=-1
+
+    shopt -s nullglob
+    for candidate in "${target}"/*.log; do
+        [[ -f "${candidate}" && -s "${candidate}" ]] || continue
+        candidate_mtime="$(stat -c '%Y' -- "${candidate}")"
+        if (( candidate_mtime > selected_mtime )) \
+            || { (( candidate_mtime == selected_mtime )) \
+                 && [[ "${candidate}" > "${selected}" ]]; }; then
+            selected="${candidate}"
+            selected_mtime="${candidate_mtime}"
+        fi
+    done
+    shopt -u nullglob
+
+    [[ -n "${selected}" ]] || die "no non-empty *.log WAL file found below ${target}"
+    printf '%s\n' "${selected}"
+}
+
+resolve_mutation_path() {
+    local component="$1"
+    local target="$2"
+    local specification
+    local candidate
+
+    case "${component}" in
+        wal)
+            specification="${BFT_STORAGE_WAL_MUTATION_FILE:-@latest-log}"
+            ;;
+        database)
+            specification="${BFT_STORAGE_DATABASE_MUTATION_FILE:-state/CURRENT}"
+            ;;
+        *)
+            die "unsupported storage component: ${component}"
+            ;;
+    esac
+
+    if [[ "${specification}" == "@latest-log" ]]; then
+        [[ "${component}" == "wal" ]] \
+            || die "@latest-log is supported only for the wal component"
+        candidate="$(latest_non_empty_log "${target}")"
+    else
+        [[ "${specification}" != /* ]] \
+            || die "mutation file must be relative to its component path"
+        candidate="$(readlink -f -- "${target}/${specification}")" \
+            || die "cannot resolve mutation file: ${specification}"
+    fi
+
+    [[ -n "${candidate}" && -f "${candidate}" ]] \
+        || die "mutation file does not exist: ${specification}"
+    path_is_within "${candidate}" "${target}" \
+        || die "mutation file escapes its component path"
+    [[ -s "${candidate}" ]] || die "mutation file is already empty: ${candidate}"
+    printf '%s\n' "${candidate}"
+}
+
+resolve_component_target() {
+    local component="$1"
+    local raw_target
+
+    case "${component}" in
+        wal)
+            raw_target="${BFT_STORAGE_WAL_PATH:-}"
+            ;;
+        database)
+            raw_target="${BFT_STORAGE_DATABASE_PATH:-}"
+            ;;
+        *)
+            die "component must be wal or database"
+            ;;
+    esac
+
+    COMPONENT_TARGET="$(canonical_existing_dir "${component} component path" "${raw_target}")"
+    path_is_within "${COMPONENT_TARGET}" "${DATA_ROOT}" \
+        || die "${component} component path must be below BFT_STORAGE_DATA_ROOT"
+    [[ "${COMPONENT_TARGET}" != "${DATA_ROOT}" ]] \
+        || die "a component path may not equal BFT_STORAGE_DATA_ROOT"
+    if path_is_within "${STATE_ROOT}" "${COMPONENT_TARGET}"; then
+        die "storage state directory may not be inside a component path"
+    fi
+}
+
+resolve_component() {
+    local component="$1"
+
+    resolve_component_target "${component}"
+    MUTATION_PATH="$(resolve_mutation_path "${component}" "${COMPONENT_TARGET}")"
+}
+
+ensure_no_active_fault() {
+    local state_file
+
+    shopt -s nullglob
+    for state_file in "${STATE_DIR}"/*.json; do
+        validate_state_document "${state_file}"
+        if [[ "$(jq -r '.active' "${state_file}")" == "true" ]]; then
+            die "another storage fault is active: $(jq -r '.faultId' "${state_file}")"
+        fi
+    done
+    shopt -u nullglob
+}
+
+write_owned_marker() {
+    local path="$1"
+    local fault_id="$2"
+    local current=""
+    local tmp
+
+    if [[ -f "${path}" ]]; then
+        current="$(<"${path}")"
+        [[ "${current}" == "${fault_id}" ]] \
+            || die "control marker ${path} belongs to ${current}"
+        return
+    fi
+
+    tmp="${path}.$$"
+    printf '%s\n' "${fault_id}" > "${tmp}"
+    chmod 640 "${tmp}"
+    mv -f "${tmp}" "${path}"
+}
+
+remove_owned_marker() {
+    local path="$1"
+    local fault_id="$2"
+    local current=""
+
+    [[ -e "${path}" ]] || return
+    [[ -f "${path}" ]] || die "control marker is not a regular file: ${path}"
+    current="$(<"${path}")"
+    [[ "${current}" == "${fault_id}" ]] \
+        || die "control marker ${path} belongs to ${current}"
+    rm -f -- "${path}"
+}
+
+owned_node_pid() {
+    local pid
+    local parent_pid
+
+    pid="$(read_numeric_pid "${NODE_PID_FILE}")" || return 1
+    kill -0 "${pid}" 2>/dev/null || return 1
+    [[ -r "/proc/${pid}/stat" ]] || return 1
+    parent_pid="$(awk '{print $4}' "/proc/${pid}/stat")"
+    [[ "${parent_pid}" == "${SUPERVISOR_PID}" ]] || return 1
+    printf '%s\n' "${pid}"
+}
+
+stop_node() {
+    local fault_id="$1"
+    local timeout="${BFT_STORAGE_STOP_TIMEOUT_SECONDS:-30}"
+    local deadline
+    local pid=""
+
+    validate_non_negative_integer BFT_STORAGE_STOP_TIMEOUT_SECONDS "${timeout}"
+    write_owned_marker "${MAINTENANCE_FILE}" "${fault_id}"
+
+    if pid="$(owned_node_pid)"; then
+        kill -TERM "${pid}"
+        deadline=$((SECONDS + timeout))
+        while kill -0 "${pid}" 2>/dev/null && (( SECONDS < deadline )); do
+            sleep 0.2
+        done
+        if kill -0 "${pid}" 2>/dev/null; then
+            kill -KILL "${pid}"
+        fi
+    fi
+
+    deadline=$((SECONDS + 5))
+    while [[ -f "${NODE_PID_FILE}" ]] && (( SECONDS < deadline )); do
+        sleep 0.2
+    done
+    [[ ! -f "${NODE_PID_FILE}" ]] \
+        || die "node supervisor did not acknowledge the stopped process"
+}
+
+wait_for_stable_node() {
+    local timeout="$1"
+    local stable_seconds="$2"
+    local deadline=$((SECONDS + timeout))
+    local stable_since=-1
+
+    while (( SECONDS < deadline )); do
+        if owned_node_pid >/dev/null; then
+            if (( stable_since < 0 )); then
+                stable_since=${SECONDS}
+            fi
+            if (( SECONDS - stable_since >= stable_seconds )); then
+                return 0
+            fi
+        else
+            stable_since=-1
+        fi
+        sleep 0.2
+    done
+    return 1
+}
+
+start_node() {
+    local fault_id="$1"
+    local timeout="$2"
+    local stable_seconds="$3"
+
+    validate_non_negative_integer start-timeout "${timeout}"
+    validate_non_negative_integer stable-seconds "${stable_seconds}"
+    remove_owned_marker "${MAINTENANCE_FILE}" "${fault_id}"
+    wait_for_stable_node "${timeout}" "${stable_seconds}"
+}
+
+verify_backup_capacity() {
+    local target="$1"
+    local reserve_mib="${BFT_STORAGE_BACKUP_RESERVE_MIB:-256}"
+    local target_bytes
+    local available_kib
+    local available_bytes
+    local required_bytes
+
+    validate_non_negative_integer BFT_STORAGE_BACKUP_RESERVE_MIB "${reserve_mib}"
+    target_bytes="$(du -sb -- "${target}" | awk '{print $1}')"
+    available_kib="$(df -Pk -- "${BACKUP_ROOT}" | awk 'NR == 2 {print $4}')"
+    validate_non_negative_integer target-size "${target_bytes}"
+    validate_non_negative_integer available-space "${available_kib}"
+    available_bytes=$((available_kib * 1024))
+    required_bytes=$((target_bytes + reserve_mib * 1024 * 1024))
+    (( available_bytes >= required_bytes )) \
+        || die "insufficient backup space: need ${required_bytes} bytes, have ${available_bytes}"
+}
+
+inject_fault() {
+    local component="$1"
+    local fault_id="$2"
+    local node_id="$3"
+    local state_file
+    local backup_dir
+    local backup_path
+    local backup_tmp
+    local backup_sha256=""
+    local original_size
+    local original_sha256
+    local node_running=false
+    local inject_timeout="${BFT_STORAGE_INJECT_START_TIMEOUT_SECONDS:-10}"
+    local inject_stable="${BFT_STORAGE_INJECT_STABLE_SECONDS:-2}"
+
+    validate_token fault-id "${fault_id}"
+    validate_token node-id "${node_id}"
+    resolve_component "${component}"
+    ensure_supervisor
+    acquire_lock
+    ensure_no_active_fault
+
+    state_file="$(state_file_for "${fault_id}")"
+    [[ ! -e "${state_file}" ]] || die "fault ID has already been used: ${fault_id}"
+    backup_dir="${BACKUP_ROOT}/${fault_id}"
+    backup_path="${backup_dir}/component.tar"
+    write_state "${state_file}" true "${fault_id}" "${component}" "${node_id}" \
+        false false false "${COMPONENT_TARGET}" "${MUTATION_PATH}" \
+        "${backup_path}" "" 0 "" false
+
+    stop_node "${fault_id}"
+    sync
+    [[ -f "${MUTATION_PATH}" && -s "${MUTATION_PATH}" ]] \
+        || die "mutation file changed before the stopped snapshot: ${MUTATION_PATH}"
+    original_size="$(stat -c '%s' -- "${MUTATION_PATH}")"
+    original_sha256="$(sha256sum -- "${MUTATION_PATH}" | awk '{print $1}')"
+    write_state "${state_file}" true "${fault_id}" "${component}" "${node_id}" \
+        false false false "${COMPONENT_TARGET}" "${MUTATION_PATH}" \
+        "${backup_path}" "" "${original_size}" "${original_sha256}" false
+
+    verify_backup_capacity "${COMPONENT_TARGET}"
+    mkdir -m 700 "${backup_dir}"
+    backup_tmp="${backup_path}.tmp"
+    tar -C "$(dirname "${COMPONENT_TARGET}")" \
+        -cf "${backup_tmp}" -- "$(basename "${COMPONENT_TARGET}")"
+    tar -tf "${backup_tmp}" >/dev/null
+    mv -f "${backup_tmp}" "${backup_path}"
+    backup_sha256="$(sha256sum -- "${backup_path}" | awk '{print $1}')"
+
+    write_state "${state_file}" true "${fault_id}" "${component}" "${node_id}" \
+        true false false "${COMPONENT_TARGET}" "${MUTATION_PATH}" \
+        "${backup_path}" "${backup_sha256}" "${original_size}" \
+        "${original_sha256}" false
+
+    write_owned_marker "${PERSISTENT_FAULT_FILE}" "${fault_id}"
+    write_owned_marker "${FAULT_FILE}" "${fault_id}"
+    truncate -s 0 -- "${MUTATION_PATH}"
+    sync
+    [[ "$(stat -c '%s' -- "${MUTATION_PATH}")" == "0" ]] \
+        || die "truncation verification failed: ${MUTATION_PATH}"
+    [[ "$(sha256sum -- "${MUTATION_PATH}" | awk '{print $1}')" != "${original_sha256}" ]] \
+        || die "mutation did not change the selected file"
+
+    write_state "${state_file}" true "${fault_id}" "${component}" "${node_id}" \
+        true true false "${COMPONENT_TARGET}" "${MUTATION_PATH}" \
+        "${backup_path}" "${backup_sha256}" "${original_size}" \
+        "${original_sha256}" false
+
+    if start_node "${fault_id}" "${inject_timeout}" "${inject_stable}"; then
+        node_running=true
+    fi
+
+    write_state "${state_file}" true "${fault_id}" "${component}" "${node_id}" \
+        true true false "${COMPONENT_TARGET}" "${MUTATION_PATH}" \
+        "${backup_path}" "${backup_sha256}" "${original_size}" \
+        "${original_sha256}" "${node_running}"
+    jq -c . "${state_file}"
+}
+
+heal_fault() {
+    local fault_id="$1"
+    local state_file
+    local component
+    local node_id
+    local backup_ready
+    local mutation_applied
+    local target_path
+    local mutation_path
+    local backup_path
+    local backup_sha256
+    local original_size
+    local original_sha256
+    local actual_backup_sha256
+    local heal_timeout="${BFT_STORAGE_HEAL_START_TIMEOUT_SECONDS:-60}"
+    local heal_stable="${BFT_STORAGE_HEAL_STABLE_SECONDS:-5}"
+
+    validate_token fault-id "${fault_id}"
+    ensure_supervisor
+    acquire_lock
+    state_file="$(state_file_for "${fault_id}")"
+    [[ -f "${state_file}" ]] || die "storage state not found for fault ${fault_id}"
+    validate_state_document "${state_file}"
+
+    [[ "$(jq -r '.faultId' "${state_file}")" == "${fault_id}" ]] \
+        || die "state fault ID does not match ${fault_id}"
+    if [[ "$(jq -r '.active' "${state_file}")" == "false" \
+          && "$(jq -r '.restored' "${state_file}")" == "true" ]]; then
+        rm -rf -- "${BACKUP_ROOT:?}/${fault_id}"
+        jq -c . "${state_file}"
+        return
+    fi
+
+    component="$(jq -r '.component' "${state_file}")"
+    node_id="$(jq -r '.nodeId' "${state_file}")"
+    backup_ready="$(jq -r '.backupReady' "${state_file}")"
+    mutation_applied="$(jq -r '.mutationApplied' "${state_file}")"
+    target_path="$(jq -r '.targetPath' "${state_file}")"
+    mutation_path="$(jq -r '.mutationPath' "${state_file}")"
+    backup_path="$(jq -r '.backupPath' "${state_file}")"
+    backup_sha256="$(jq -r '.backupSha256' "${state_file}")"
+    original_size="$(jq -r '.originalSizeBytes' "${state_file}")"
+    original_sha256="$(jq -r '.originalSha256' "${state_file}")"
+
+    resolve_component_target "${component}"
+    [[ "${COMPONENT_TARGET}" == "${target_path}" ]] \
+        || die "configured component path no longer matches the recovery state"
+    path_is_within "${mutation_path}" "${target_path}" \
+        || die "recovery mutation path escapes its component path"
+    [[ "${backup_path}" == "${BACKUP_ROOT}/${fault_id}/component.tar" ]] \
+        || die "backup path does not belong to this fault"
+
+    stop_node "${fault_id}"
+    if [[ "${backup_ready}" == "true" ]]; then
+        [[ -f "${backup_path}" ]] || die "recovery backup is missing: ${backup_path}"
+        actual_backup_sha256="$(sha256sum -- "${backup_path}" | awk '{print $1}')"
+        [[ "${actual_backup_sha256}" == "${backup_sha256}" ]] \
+            || die "recovery backup hash mismatch"
+
+        path_is_within "${target_path}" "${DATA_ROOT}" \
+            || die "refusing to restore a target outside the data root"
+        [[ "${target_path}" != "${DATA_ROOT}" ]] \
+            || die "refusing to replace the data root"
+        rm -rf -- "${target_path}"
+        tar -C "$(dirname "${target_path}")" -xf "${backup_path}"
+        [[ -f "${mutation_path}" ]] || die "restored mutation file is missing"
+        [[ "$(stat -c '%s' -- "${mutation_path}")" == "${original_size}" ]] \
+            || die "restored mutation file has the wrong size"
+        [[ "$(sha256sum -- "${mutation_path}" | awk '{print $1}')" == "${original_sha256}" ]] \
+            || die "restored mutation file hash mismatch"
+    elif [[ "${mutation_applied}" == "true" ]]; then
+        die "state reports a mutation without a verified backup"
+    fi
+
+    sync
+    remove_owned_marker "${FAULT_FILE}" "${fault_id}"
+    remove_owned_marker "${PERSISTENT_FAULT_FILE}" "${fault_id}"
+    start_node "${fault_id}" "${heal_timeout}" "${heal_stable}" \
+        || die "node did not become stable after storage restoration"
+
+    write_state "${state_file}" false "${fault_id}" "${component}" "${node_id}" \
+        "${backup_ready}" "${mutation_applied}" true "${target_path}" \
+        "${mutation_path}" "${backup_path}" "${backup_sha256}" \
+        "${original_size}" "${original_sha256}" true
+    rm -rf -- "${BACKUP_ROOT:?}/${fault_id}"
+    jq -c . "${state_file}"
+}
+
+read_state() {
+    local fault_id="$1"
+    local expected_component="${2:-}"
+    local expected_node="${3:-}"
+    local state_file
+
+    validate_token fault-id "${fault_id}"
+    if [[ -n "${expected_node}" ]]; then
+        validate_token node-id "${expected_node}"
+    fi
+    ensure_supervisor
+
+    state_file="$(state_file_for "${fault_id}")"
+    if [[ ! -f "${state_file}" ]]; then
+        if [[ -n "${expected_component}" ]]; then
+            resolve_component "${expected_component}"
+        fi
+        jq -cn '{active: false, backupReady: false,
+                 mutationApplied: false, restored: false}'
+        return
+    fi
+
+    validate_state_document "${state_file}"
+    [[ "$(jq -r '.faultId' "${state_file}")" == "${fault_id}" ]] \
+        || die "state fault ID does not match ${fault_id}"
+    if [[ -n "${expected_component}" ]]; then
+        resolve_component_target "${expected_component}"
+        [[ "$(jq -r '.component' "${state_file}")" == "${expected_component}" ]] \
+            || die "state component does not match ${expected_component}"
+        [[ "$(jq -r '.targetPath' "${state_file}")" == "${COMPONENT_TARGET}" ]] \
+            || die "state component path does not match current configuration"
+    fi
+    if [[ -n "${expected_node}" ]]; then
+        [[ "$(jq -r '.nodeId' "${state_file}")" == "${expected_node}" ]] \
+            || die "state node ID does not match ${expected_node}"
+    fi
+    jq -c . "${state_file}"
+}
+
+usage() {
+    cat >&2 <<'EOF'
+Usage:
+  bft-storage inject <wal|database> <fault-id> <node-id>
+  bft-storage heal <fault-id>
+  bft-storage read <fault-id> [wal|database] [node-id]
+EOF
+    exit 2
+}
+
+[[ $# -ge 1 ]] || usage
+command_name="$1"
+shift
+
+initialize
+
+case "${command_name}" in
+    inject)
+        [[ $# -eq 3 ]] || usage
+        inject_fault "$1" "$2" "$3"
+        ;;
+    heal)
+        [[ $# -eq 1 ]] || usage
+        heal_fault "$1"
+        ;;
+    read)
+        [[ $# -ge 1 && $# -le 3 ]] || usage
+        read_state "$@"
+        ;;
+    *)
+        usage
+        ;;
+esac
diff --git a/docker/gravity_node/test-byzantine-fixture.sh b/docker/gravity_node/test-byzantine-fixture.sh
new file mode 100644
index 00000000..e0c4ba20
--- /dev/null
+++ b/docker/gravity_node/test-byzantine-fixture.sh
@@ -0,0 +1,98 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
+SUFFIX="${BFT_BYZANTINE_TEST_SUFFIX:-$$}"
+IMAGE="gravity-node-byzantine-fixture:${SUFFIX}"
+NORMAL_IMAGE="gravity-node-byzantine-normal:${SUFFIX}"
+CONTAINER="gravity-byzantine-fixture-${SUFFIX}"
+
+cleanup() {
+    docker rm -f "${CONTAINER}" >/dev/null 2>&1 || true
+    if [[ "${BFT_BYZANTINE_TEST_KEEP_IMAGES:-0}" != "1" ]]; then
+        docker image rm -f "${IMAGE}" "${NORMAL_IMAGE}" >/dev/null 2>&1 || true
+    fi
+}
+trap cleanup EXIT
+
+fail() {
+    echo "Byzantine fixture test: $*" >&2
+    exit 1
+}
+
+assert_json() {
+    local document="$1"
+    local expression="$2"
+
+    jq -e "${expression}" <<<"${document}" >/dev/null \
+        || fail "JSON assertion failed: ${expression}; document=${document}"
+}
+
+build_args=(
+    --build-arg HOST_BINARY=docker/gravity_node/tests/byzantine-fixture/fixture-gravity-node.sh
+    --build-arg HOST_CLI_BINARY=docker/gravity_node/tests/storage-fixture/fixture-gravity-cli.sh
+    -f docker/gravity_node/Dockerfile
+)
+
+docker build --load "${build_args[@]}" \
+    --target runtime-host-binary-byzantine-test \
+    -t "${IMAGE}" "${REPO_ROOT}"
+docker build --load "${build_args[@]}" \
+    --target runtime-host-binary \
+    -t "${NORMAL_IMAGE}" "${REPO_ROOT}"
+
+docker run --rm --entrypoint /bin/sh "${NORMAL_IMAGE}" \
+    -c 'test ! -e /usr/local/bin/bft-byzantine' \
+    || fail "normal runtime unexpectedly contains the Byzantine hook"
+
+if docker run --rm --user 0 --entrypoint /usr/local/bin/bft-byzantine \
+    "${IMAGE}" read disabled-fixture >/dev/null 2>&1; then
+    fail "Byzantine hook ran without explicit runtime authorization"
+fi
+
+docker run -d \
+    --name "${CONTAINER}" \
+    --entrypoint /usr/local/bin/gravity_node \
+    -e BFT_BYZANTINE_FIXTURE_ENABLED=1 \
+    "${IMAGE}" node >/dev/null
+
+baseline="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-byzantine \
+    read equivocation-1)"
+assert_json "${baseline}" \
+    '(.active | not) and .capabilities == ["equivocation"] and (.protocolEffect.observed | not)'
+
+if docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-byzantine \
+    inject double-sign unsupported-1 validator-1 >/dev/null 2>&1; then
+    fail "hook accepted an unsupported double-sign behavior"
+fi
+
+docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-byzantine \
+    inject equivocation equivocation-1 validator-1
+
+attempts=100
+while (( attempts > 0 )); do
+    active="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-byzantine \
+        read equivocation-1)"
+    if jq -e '
+        .active and
+        .protocolEffect.observed and
+        .protocolEffect.eventCount >= 1 and
+        .protocolEffect.distinctMessageCount >= 2 and
+        .protocolEffect.recipientGroupCount >= 2
+      ' <<<"${active}" >/dev/null; then
+        break
+    fi
+    attempts=$((attempts - 1))
+    sleep 0.1
+done
+(( attempts > 0 )) || fail "fixture node did not publish protocol-effect evidence"
+
+docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-byzantine \
+    heal equivocation-1
+healed="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-byzantine \
+    read equivocation-1)"
+assert_json "${healed}" \
+    '(.active | not) and .protocolEffect.observed and .protocolEffect.distinctMessageCount == 2'
+
+echo "Byzantine fixture test: injection, evidence, and recovery passed"
diff --git a/docker/gravity_node/test-storage-fixture.sh b/docker/gravity_node/test-storage-fixture.sh
new file mode 100755
index 00000000..37bace98
--- /dev/null
+++ b/docker/gravity_node/test-storage-fixture.sh
@@ -0,0 +1,187 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
+SUFFIX="${BFT_STORAGE_TEST_SUFFIX:-$$}"
+IMAGE="gravity-node-storage-fixture:${SUFFIX}"
+NORMAL_IMAGE="gravity-node-storage-normal:${SUFFIX}"
+CONTAINER="gravity-storage-fixture-${SUFFIX}"
+VOLUME="gravity-storage-fixture-${SUFFIX}"
+CONFIG_DIR=""
+
+cleanup() {
+    docker rm -f "${CONTAINER}" >/dev/null 2>&1 || true
+    docker volume rm "${VOLUME}" >/dev/null 2>&1 || true
+    if [[ -n "${CONFIG_DIR}" ]]; then
+        rm -rf -- "${CONFIG_DIR}"
+    fi
+    if [[ "${BFT_STORAGE_TEST_KEEP_IMAGES:-0}" != "1" ]]; then
+        docker image rm -f "${IMAGE}" "${NORMAL_IMAGE}" >/dev/null 2>&1 || true
+    fi
+}
+trap cleanup EXIT
+
+fail() {
+    echo "storage fixture test: $*" >&2
+    exit 1
+}
+
+wait_for_node() {
+    local attempts=100
+
+    while (( attempts > 0 )); do
+        if docker exec -u 0 "${CONTAINER}" sh -c \
+            'test -s /run/bft-node/node.pid && test -s /gravity/data/fixture-node.heartbeat' \
+            >/dev/null 2>&1; then
+            return
+        fi
+        attempts=$((attempts - 1))
+        sleep 0.2
+    done
+    fail "fixture node did not become ready"
+}
+
+wait_for_fault_exit() {
+    local attempts=50
+
+    while (( attempts > 0 )); do
+        if docker exec -u 0 "${CONTAINER}" test ! -e /run/bft-node/node.pid \
+            >/dev/null 2>&1; then
+            return
+        fi
+        attempts=$((attempts - 1))
+        sleep 0.2
+    done
+    fail "fixture node did not exit after storage truncation"
+}
+
+assert_json() {
+    local document="$1"
+    local expression="$2"
+
+    jq -e "${expression}" <<<"${document}" >/dev/null \
+        || fail "JSON assertion failed: ${expression}; document=${document}"
+}
+
+build_args=(
+    --build-arg HOST_BINARY=docker/gravity_node/tests/storage-fixture/fixture-gravity-node.sh
+    --build-arg HOST_CLI_BINARY=docker/gravity_node/tests/storage-fixture/fixture-gravity-cli.sh
+    -f docker/gravity_node/Dockerfile
+)
+
+CONFIG_DIR="$(mktemp -d "${SCRIPT_DIR}/tests/storage-fixture/config.XXXXXX")"
+cat > "${CONFIG_DIR}/reth_config.json" <<'EOF'
+{
+  "reth_args": {},
+  "env_vars": {}
+}
+EOF
+
+docker build --load "${build_args[@]}" \
+    --target runtime-host-binary-storage-test \
+    -t "${IMAGE}" "${REPO_ROOT}"
+docker build --load "${build_args[@]}" \
+    --target runtime-host-binary \
+    -t "${NORMAL_IMAGE}" "${REPO_ROOT}"
+
+docker run --rm --entrypoint /bin/sh "${NORMAL_IMAGE}" \
+    -c 'test ! -e /usr/local/bin/bft-storage' \
+    || fail "normal runtime unexpectedly contains the destructive hook"
+
+if docker run --rm --entrypoint /usr/local/bin/bft-storage "${IMAGE}" \
+    read disabled-fixture >/dev/null 2>&1; then
+    fail "storage hook ran without explicit disposable-data authorization"
+fi
+
+docker volume create "${VOLUME}" >/dev/null
+docker run -d \
+    --name "${CONTAINER}" \
+    --mount "type=volume,src=${VOLUME},dst=/gravity/data" \
+    --mount "type=bind,src=${CONFIG_DIR},dst=/gravity/config,readonly" \
+    -e BFT_STORAGE_FIXTURE_ENABLED=1 \
+    -e BFT_STORAGE_DISPOSABLE_DATA=I_UNDERSTAND_THIS_DATA_WILL_BE_DESTROYED \
+    -e BFT_STORAGE_WAL_PATH=/gravity/data/data/consensus_db \
+    -e BFT_STORAGE_DATABASE_PATH=/gravity/data/data/reth/db \
+    -e BFT_STORAGE_DATABASE_MUTATION_FILE=state/CURRENT \
+    -e BFT_STORAGE_BACKUP_RESERVE_MIB=1 \
+    -e BFT_STORAGE_INJECT_START_TIMEOUT_SECONDS=2 \
+    -e BFT_STORAGE_INJECT_STABLE_SECONDS=1 \
+    -e BFT_STORAGE_HEAL_START_TIMEOUT_SECONDS=10 \
+    -e BFT_STORAGE_HEAL_STABLE_SECONDS=1 \
+    "${IMAGE}" >/dev/null
+
+wait_for_node
+
+baseline="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    read wal-1 wal validator-1)"
+assert_json "${baseline}" '.active == false and .restored == false'
+
+wal_before="$(docker exec -u 0 "${CONTAINER}" sha256sum \
+    /gravity/data/data/consensus_db/000001.log | awk '{print $1}')"
+wal_active="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    inject wal wal-1 validator-1)"
+assert_json "${wal_active}" \
+    '.active and .backupReady and .mutationApplied and (.restored | not)'
+wal_read="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    read wal-1 wal validator-1)"
+assert_json "${wal_read}" \
+    '.active and .backupReady and .mutationApplied and (.restored | not)'
+docker exec -u 0 "${CONTAINER}" test ! -s \
+    /gravity/data/data/consensus_db/000001.log
+wait_for_fault_exit
+docker inspect --format '{{.State.Running}}' "${CONTAINER}" | grep -qx true \
+    || fail "container exited with the corrupted fixture node"
+
+# A whole-container restart loses /run but must retain the volume-backed fault
+# marker. The replacement supervisor stays available for read/heal without
+# entering a restart storm against the corrupted node.
+docker restart "${CONTAINER}" >/dev/null
+wait_for_fault_exit
+docker exec -u 0 "${CONTAINER}" test -e /gravity/data/.bft-storage-active
+wal_after_restart="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    read wal-1 wal validator-1)"
+assert_json "${wal_after_restart}" \
+    '.active and .backupReady and .mutationApplied and (.restored | not)'
+
+wal_healed="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    heal wal-1)"
+assert_json "${wal_healed}" \
+    '(.active | not) and .backupReady and .mutationApplied and .restored and .nodeRunning'
+wait_for_node
+wal_after="$(docker exec -u 0 "${CONTAINER}" sha256sum \
+    /gravity/data/data/consensus_db/000001.log | awk '{print $1}')"
+[[ "${wal_after}" == "${wal_before}" ]] || fail "WAL bytes were not restored"
+
+wal_healed_again="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    heal wal-1)"
+assert_json "${wal_healed_again}" '(.active | not) and .restored'
+
+database_before="$(docker exec -u 0 "${CONTAINER}" sha256sum \
+    /gravity/data/data/reth/db/state/CURRENT | awk '{print $1}')"
+database_active="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    inject database database-1 pfn-1)"
+assert_json "${database_active}" \
+    '.active and .backupReady and .mutationApplied and (.restored | not)'
+database_read="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    read database-1 database pfn-1)"
+assert_json "${database_read}" \
+    '.active and .backupReady and .mutationApplied and (.restored | not)'
+docker exec -u 0 "${CONTAINER}" test ! -s \
+    /gravity/data/data/reth/db/state/CURRENT
+wait_for_fault_exit
+
+database_healed="$(docker exec -u 0 "${CONTAINER}" /usr/local/bin/bft-storage \
+    heal database-1)"
+assert_json "${database_healed}" \
+    '(.active | not) and .backupReady and .mutationApplied and .restored and .nodeRunning'
+wait_for_node
+database_after="$(docker exec -u 0 "${CONTAINER}" sha256sum \
+    /gravity/data/data/reth/db/state/CURRENT | awk '{print $1}')"
+[[ "${database_after}" == "${database_before}" ]] \
+    || fail "database bytes were not restored"
+
+docker exec -u 0 "${CONTAINER}" test ! -d \
+    /gravity/data/.bft-storage/backups/database-1
+
+echo "storage fixture test: WAL and database injection/recovery passed"
diff --git a/docker/gravity_node/tests/byzantine-fixture/fixture-gravity-node.sh b/docker/gravity_node/tests/byzantine-fixture/fixture-gravity-node.sh
new file mode 100644
index 00000000..2ec16ba7
--- /dev/null
+++ b/docker/gravity_node/tests/byzantine-fixture/fixture-gravity-node.sh
@@ -0,0 +1,43 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+CONTROL_PATH="${BFT_BYZANTINE_CONTROL_PATH:-/run/bft-node/byzantine-control.json}"
+EVIDENCE_PATH="${BFT_BYZANTINE_EVIDENCE_PATH:-/run/bft-node/byzantine-evidence.json}"
+shutdown=0
+last_fault=""
+
+trap 'shutdown=1' TERM INT
+
+while (( shutdown == 0 )); do
+    if [[ -f "${CONTROL_PATH}" ]] \
+        && [[ "$(jq -r '.active // false' "${CONTROL_PATH}")" == "true" ]]; then
+        fault_id="$(jq -r '.faultId' "${CONTROL_PATH}")"
+        if [[ "${fault_id}" != "${last_fault}" ]]; then
+            node_id="$(jq -r '.nodeId' "${CONTROL_PATH}")"
+            temporary="${EVIDENCE_PATH}.$$"
+            jq -cn \
+                --arg faultId "${fault_id}" \
+                --arg nodeId "${node_id}" \
+                '{schemaVersion: 1,
+                  faultId: $faultId,
+                  behavior: "equivocation",
+                  nodeId: $nodeId,
+                  protocolEffect: {
+                    observed: true,
+                    behavior: "equivocation",
+                    eventCount: 1,
+                    epoch: 7,
+                    round: 11,
+                    distinctMessageCount: 2,
+                    recipientGroupCount: 2,
+                    firstMessageId: "0x01",
+                    secondMessageId: "0x02",
+                    firstRecipientCount: 2,
+                    secondRecipientCount: 2
+                  }}' > "${temporary}"
+            mv -f -- "${temporary}" "${EVIDENCE_PATH}"
+            last_fault="${fault_id}"
+        fi
+    fi
+    sleep 0.1
+done
diff --git a/docker/gravity_node/tests/storage-fixture/fixture-gravity-cli.sh b/docker/gravity_node/tests/storage-fixture/fixture-gravity-cli.sh
new file mode 100755
index 00000000..d12efe2d
--- /dev/null
+++ b/docker/gravity_node/tests/storage-fixture/fixture-gravity-cli.sh
@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+echo "fixture-gravity-cli $*"
diff --git a/docker/gravity_node/tests/storage-fixture/fixture-gravity-node.sh b/docker/gravity_node/tests/storage-fixture/fixture-gravity-node.sh
new file mode 100755
index 00000000..6a95c1a3
--- /dev/null
+++ b/docker/gravity_node/tests/storage-fixture/fixture-gravity-node.sh
@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+WAL_DIR="${BFT_STORAGE_WAL_PATH:-/gravity/data/data/consensus_db}"
+DATABASE_DIR="${BFT_STORAGE_DATABASE_PATH:-/gravity/data/data/reth/db}"
+WAL_FILE="${WAL_DIR}/000001.log"
+DATABASE_FILE="${DATABASE_DIR}/state/CURRENT"
+HEARTBEAT="${BFT_FIXTURE_HEARTBEAT:-/gravity/data/fixture-node.heartbeat}"
+
+mkdir -p "${WAL_DIR}" "$(dirname "${DATABASE_FILE}")"
+if [[ ! -e "${WAL_FILE}" ]]; then
+    printf 'fixture-wal-original-bytes\n' > "${WAL_FILE}"
+fi
+if [[ ! -e "${DATABASE_FILE}" ]]; then
+    printf 'fixture-database-original-bytes\n' > "${DATABASE_FILE}"
+fi
+
+if [[ ! -s "${WAL_FILE}" || ! -s "${DATABASE_FILE}" ]]; then
+    echo "fixture-gravity-node: refusing to start with truncated storage" >&2
+    exit 42
+fi
+
+shutdown=0
+trap 'shutdown=1' TERM INT
+
+while (( shutdown == 0 )); do
+    heartbeat_tmp="${HEARTBEAT}.$$"
+    printf '%s\n' "$$" > "${heartbeat_tmp}"
+    mv -f "${heartbeat_tmp}" "${HEARTBEAT}"
+    sleep 0.2
+done
+
+rm -f "${HEARTBEAT}"