diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml
index e83b3c54..04da6cf0 100644
--- a/.github/workflows/rust-test.yml
+++ b/.github/workflows/rust-test.yml
@@ -76,6 +76,10 @@ jobs:
         run: |
           cargo clippy -p lance-context-wal --features raft-log --all-targets -- -D warnings
           cargo test -p lance-context-wal --features raft-log --lib --test raft_log --test raft_log_cache_identity
+      - name: Replicated WAL quorum and process recovery
+        run: |
+          cargo clippy -p lance-context-wal --features replication --all-targets -- -D warnings
+          cargo test -p lance-context-wal --features replication --lib --test replication_process
       - name: WAL resource observation in a container cgroup namespace
         run: |
           cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl"
diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml
index 61a9c4fe..21333b55 100644
--- a/crates/lance-context-wal/Cargo.toml
+++ b/crates/lance-context-wal/Cargo.toml
@@ -7,6 +7,7 @@ publish = false
 description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
+replication = ["raft-log", "record", "dep:bincode", "dep:slog"]
 raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"]
 vectored-io = ["dep:bytes", "dep:tokio"]
 runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"]
@@ -19,6 +20,8 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de
 lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
+bincode = { version = "1.3", optional = true }
+slog = { version = "2", optional = true }
 tokio = { version = "1", features = ["io-util"], optional = true }
 anyhow = { version = "1", optional = true }
 arrow-array = { version = "58", optional = true }
diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs
index 39ffe258..fe9b8a0d 100644
--- a/crates/lance-context-wal/src/lib.rs
+++ b/crates/lance-context-wal/src/lib.rs
@@ -11,6 +11,9 @@ pub mod placement;
 #[cfg(feature = "raft-log")]
 pub mod raft_log;
 
+#[cfg(feature = "replication")]
+pub mod replication;
+
 #[cfg(feature = "admission")]
 pub mod admission;
 
diff --git a/crates/lance-context-wal/src/raft_log.rs b/crates/lance-context-wal/src/raft_log.rs
index 6011c0fb..e8145871 100644
--- a/crates/lance-context-wal/src/raft_log.rs
+++ b/crates/lance-context-wal/src/raft_log.rs
@@ -367,6 +367,37 @@ pub struct GroupLog {
 }
 
 impl GroupLog {
+    #[cfg(all(test, feature = "replication"))]
+    pub(crate) fn fail_next_sync_for_test(&self) {
+        self.node.fault.fail_sync_for_test();
+    }
+
+    /// Bind an immutable application format/policy before its first entry.
+    /// Reopen must supply identical bytes; an unidentified existing log cannot
+    /// be silently interpreted as a different state machine.
+    pub fn bind_application(&self, identity: &[u8]) -> Result<()> {
+        ensure!(
+            !identity.is_empty() && identity.len() <= 4096,
+            "invalid application identity"
+        );
+        let state = self
+            .state
+            .write()
+            .map_err(|_| anyhow::anyhow!("group state lock poisoned"))?;
+        self.node.check()?;
+        if let Some(saved) = self.node.engine.get(self.group, b"application") {
+            ensure!(saved == identity, "application format or policy changed");
+            return self.node.check();
+        }
+        ensure!(
+            state.last() == 0,
+            "cannot identify an existing application log"
+        );
+        let mut batch = LogBatch::default();
+        batch.put(self.group, b"application".to_vec(), identity.to_vec())?;
+        self.node.write(&mut batch)
+    }
+
     fn read(&self) -> Result<RwLockReadGuard<'_, GroupState>> {
         self.node.check()?;
         let state = self
diff --git a/crates/lance-context-wal/src/raft_log/fault_tests.rs b/crates/lance-context-wal/src/raft_log/fault_tests.rs
index bd1d81b0..3032de0a 100644
--- a/crates/lance-context-wal/src/raft_log/fault_tests.rs
+++ b/crates/lance-context-wal/src/raft_log/fault_tests.rs
@@ -15,11 +15,19 @@ type FileHandle = <DefaultFileSystem as FileSystem>::Handle;
 #[derive(Default)]
 pub(super) struct FaultControl {
     fail_sync: AtomicBool,
+    fail_once: AtomicBool,
     sync_calls: AtomicU64,
     entered: Mutex<Option<mpsc::SyncSender<()>>>,
     release: Mutex<Option<mpsc::Receiver<()>>>,
 }
 
+impl FaultControl {
+    #[cfg(feature = "replication")]
+    pub(super) fn fail_sync_for_test(&self) {
+        self.fail_once.store(true, Ordering::Release);
+    }
+}
+
 pub(super) struct FaultHandle {
     inner: Arc<FileHandle>,
     control: Arc<FaultControl>,
@@ -36,7 +44,9 @@ impl Handle for FaultHandle {
 
     fn sync(&self) -> io::Result<()> {
         self.control.sync_calls.fetch_add(1, Ordering::Relaxed);
-        if self.control.fail_sync.load(Ordering::Acquire) {
+        if self.control.fail_once.swap(false, Ordering::AcqRel)
+            || self.control.fail_sync.load(Ordering::Acquire)
+        {
             if let Some(entered) = self.control.entered.lock().unwrap().take() {
                 entered.send(()).unwrap();
             }
diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs
new file mode 100644
index 00000000..918a18ba
--- /dev/null
+++ b/crates/lance-context-wal/src/replication.rs
@@ -0,0 +1,532 @@
+//! Synchronous, transport-independent Raft ingestion for static RF3 groups.
+//!
+//! `propose` queues input; only `poll` can return a committed result. All outgoing
+//! messages and completions follow durable Ready/LightReady persistence. Exact
+//! retries enter consensus again and return the original accepted index after
+//! commit. An isolated former leader cannot answer retries from local state.
+//!
+//! The owner schedules ticks, bounded network I/O and calls on one thread. This
+//! increment retains the full log and bounded metadata: it is not the production
+//! service, a snapshot/retention implementation, or a bounded disk allocator.
+//! Message scope checks are not authentication. Do not expose it to untrusted
+//! transport or enable ingestion without the remaining admission/lifecycle work.
+
+mod state;
+#[cfg(test)]
+mod tests;
+
+use std::{collections::BTreeMap, path::Path, sync::Arc};
+
+use anyhow::{ensure, Context, Result};
+use protobuf::Message as _;
+use raft::{RawNode, StateRole, Storage};
+use raft_proto::eraftpb::{Entry, Message, MessageType};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+use crate::{
+    placement::Layout,
+    raft_log::{GroupLog, PhysicalLog},
+    record::{Batch, BatchMeta},
+};
+use state::State;
+
+/// Immutable, replicated-state bounds. All peers and reopens must use the same
+/// policy. These bounds do not account for queues in the caller's transport.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Limits {
+    pub max_record_bytes: u64,
+    pub max_records: u64,
+    pub max_producers: u64,
+    pub max_pending: u64,
+    pub max_uncommitted_bytes: u64,
+}
+impl Default for Limits {
+    fn default() -> Self {
+        Self {
+            max_record_bytes: 8 * 1024 * 1024,
+            max_records: 100_000,
+            max_producers: 16_384,
+            max_pending: 128,
+            max_uncommitted_bytes: 64 * 1024 * 1024,
+        }
+    }
+}
+impl Limits {
+    fn identity(&self) -> Result<Vec<u8>> {
+        ensure!(
+            (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
+                && (1..=500_000).contains(&self.max_records)
+                && (1..=100_000).contains(&self.max_producers)
+                && (1..=4096).contains(&self.max_pending)
+                && self.max_uncommitted_bytes >= self.max_record_bytes.max(1024 * 1024)
+                && self.max_uncommitted_bytes <= 256 * 1024 * 1024,
+            "invalid replication limits"
+        );
+        Ok(bincode::serialize(&(
+            "lance-context/replicated-append/v1",
+            self,
+        ))?)
+    }
+}
+
+/// Transport supplies a bounded protobuf decoder. Scope checks prevent mixing
+/// groups/layouts/policies; authenticated peer identity is still the caller's job.
+#[derive(Clone, Debug)]
+pub struct Packet {
+    pub cluster: Uuid,
+    pub layout: [u8; 32],
+    pub policy: [u8; 32],
+    pub group: u64,
+    pub message: Message,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum Applied {
+    /// A new append or exact latest retry. Index identifies the original append.
+    Accepted {
+        index: u64,
+    },
+    /// Gap, stale sequence or changed content at an already-used sequence.
+    Rejected {
+        expected_sequence: u64,
+    },
+    Capacity,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum Outcome {
+    Committed(Applied),
+    /// Leadership changed before this owner observed the result. The append may
+    /// still commit; retry its exact identity and body, not a new sequence.
+    Unknown,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Completion {
+    pub group: u64,
+    /// Caller correlation within this process; never persisted as a producer ID.
+    pub request: u64,
+    pub outcome: Outcome,
+}
+
+#[derive(Default)]
+pub struct Events {
+    pub messages: Vec<Packet>,
+    pub completions: Vec<Completion>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Status {
+    pub leader: u64,
+    pub term: u64,
+    pub committed: u64,
+    pub applied: u64,
+    pub accepted_records: usize,
+    pub producers: usize,
+    pub pending: usize,
+}
+
+/// A bounded local view captured together with its next cursor. Preserve the
+/// group/layout scope; this is neither a quorum read nor proof of Lance visibility.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Page {
+    pub entries: Vec<(u64, BatchMeta)>,
+    /// All accepted records after the requested cursor through this position
+    /// are included. May advance over rejected, duplicate or blank entries.
+    pub through: u64,
+}
+
+struct Group {
+    raw: RawNode<GroupLog>,
+    state: State,
+    // Index AND term prevents suffix replacement from completing another input.
+    pending: BTreeMap<(u64, u64), (u64, BatchMeta)>,
+}
+
+/// Owns all Raft groups in one shared physical engine. Do not call concurrently
+/// or retain storage handles outside it. An internal/disk error fences this node;
+/// drop it, reconcile outstanding requests and reopen the original directories.
+pub struct Node {
+    layout: Arc<Layout>,
+    layout_hash: [u8; 32],
+    policy: [u8; 32],
+    limits: Limits,
+    id: u64,
+    groups: BTreeMap<u64, Group>,
+    failed: bool,
+}
+
+impl Node {
+    /// Give each group to an independent worker while retaining the shared
+    /// physical engine and its sync/failure fence. Consumes the aggregate owner;
+    /// it never clones a RawNode or permits two owners of one group.
+    pub fn into_groups(self) -> Vec<Self> {
+        self.groups
+            .into_iter()
+            .map(|(id, group)| Self {
+                layout: self.layout.clone(),
+                layout_hash: self.layout_hash,
+                policy: self.policy,
+                limits: self.limits.clone(),
+                id: self.id,
+                groups: BTreeMap::from([(id, group)]),
+                failed: self.failed,
+            })
+            .collect()
+    }
+
+    pub fn open(directory: &Path, layout: Arc<Layout>, id: u64, limits: Limits) -> Result<Self> {
+        let layout_hash = layout.fingerprint()?;
+        let identity = limits.identity()?;
+        let policy = Sha256::digest(&identity).into();
+        let log = PhysicalLog::open(directory, &layout, id)?;
+        let mut groups = BTreeMap::new();
+        let logger = slog::Logger::root(slog::Discard, slog::o!());
+        for (group, voters) in &layout.groups {
+            if !voters.contains(&id) {
+                continue;
+            }
+            let store = log.group(*group)?;
+            store.bind_application(&identity)?;
+            let committed = store.initial_state()?.hard_state.commit;
+            let mut state = State::default();
+            while state.applied < committed {
+                let entries = store.entries(
+                    state.applied + 1,
+                    (state.applied + 257).min(committed + 1),
+                    Some(1024 * 1024),
+                    raft::GetEntriesContext::empty(false),
+                )?;
+                ensure!(!entries.is_empty(), "missing committed recovery entries");
+                for entry in entries {
+                    state::validate_entry(&layout, *group, &limits, &entry, true)?;
+                    state.apply(&entry, &limits)?;
+                }
+            }
+            let config = raft::Config {
+                id,
+                applied: committed,
+                election_tick: 10,
+                heartbeat_tick: 2,
+                check_quorum: true,
+                pre_vote: true,
+                max_size_per_msg: 1024 * 1024,
+                max_inflight_msgs: 16,
+                max_uncommitted_size: limits.max_uncommitted_bytes,
+                max_committed_size_per_ready: 1024 * 1024,
+                ..Default::default()
+            };
+            let raw = RawNode::new(&config, store, &logger)?;
+            groups.insert(
+                *group,
+                Group {
+                    raw,
+                    state,
+                    pending: BTreeMap::new(),
+                },
+            );
+        }
+        Ok(Self {
+            layout,
+            layout_hash,
+            policy,
+            limits,
+            id,
+            groups,
+            failed: false,
+        })
+    }
+
+    fn healthy(&self) -> Result<()> {
+        ensure!(!self.failed, "replication node fenced; drop and reopen it");
+        // Another group shares this engine's uncertain-write fence.
+        for group in self.groups.values() {
+            group.raw.store().initial_state()?;
+        }
+        Ok(())
+    }
+
+    pub fn tick(&mut self) -> Result<()> {
+        self.healthy()?;
+        for group in self.groups.values_mut() {
+            group.raw.tick();
+        }
+        Ok(())
+    }
+
+    pub fn campaign(&mut self, group: u64) -> Result<()> {
+        self.healthy()?;
+        self.groups
+            .get_mut(&group)
+            .context("unhosted group")?
+            .raw
+            .campaign()?;
+        Ok(())
+    }
+
+    pub fn status(&self, group: u64) -> Result<Status> {
+        self.healthy()?;
+        let g = self.groups.get(&group).context("unhosted group")?;
+        Ok(Status {
+            leader: g.raw.raft.leader_id,
+            term: g.raw.raft.term,
+            committed: g.raw.store().initial_state()?.hard_state.commit,
+            applied: g.state.applied,
+            accepted_records: g.state.accepted.len(),
+            producers: g.state.producers.len(),
+            pending: g.pending.len(),
+        })
+    }
+
+    /// Ok means queued locally, NOT durable or committed. A token may have only
+    /// one unresolved request on this node. No retry result bypasses consensus.
+    pub fn propose(&mut self, group: u64, request: u64, batch: Batch) -> Result<()> {
+        self.healthy()?;
+        ensure!(
+            !self
+                .groups
+                .values()
+                .any(|g| g.pending.values().any(|p| p.0 == request)),
+            "request token already pending"
+        );
+        state::validate_meta(
+            &self.layout,
+            group,
+            &self.limits,
+            &batch.meta,
+            &batch.body,
+            true,
+        )?;
+        let g = self.groups.get_mut(&group).context("unhosted group")?;
+        ensure!(g.raw.raft.state == StateRole::Leader, "not leader");
+        ensure!(
+            g.pending.len() < self.limits.max_pending as usize,
+            "pending proposal capacity"
+        );
+        let index = g.raw.raft.raft_log.last_index() + 1;
+        let term = g.raw.raft.term;
+        // RawNode::propose accepts Vec and would copy the opaque Bytes. This is
+        // the equivalent raft-rs proposal path with shared entry.data ownership.
+        let entry = Entry {
+            context: state::encode_meta(&batch.meta)?.into(),
+            data: batch.body,
+            ..Default::default()
+        };
+        let mut message = Message::default();
+        message.set_msg_type(MessageType::MsgPropose);
+        message.from = self.id;
+        message.set_entries(vec![entry].into());
+        g.raw.raft.step(message)?;
+        ensure!(
+            g.raw.raft.raft_log.last_index() == index,
+            "proposal was not queued"
+        );
+        g.pending.insert((index, term), (request, batch.meta));
+        Ok(())
+    }
+
+    pub fn step(&mut self, packet: Packet) -> Result<()> {
+        self.healthy()?;
+        ensure!(
+            packet.cluster == self.layout.cluster
+                && packet.layout == self.layout_hash
+                && packet.policy == self.policy,
+            "foreign replication scope or policy"
+        );
+        let voters = self
+            .layout
+            .groups
+            .get(&packet.group)
+            .context("unknown group")?;
+        let message = packet.message;
+        ensure!(
+            message.to == self.id && message.from != self.id && voters.contains(&message.from),
+            "invalid Raft peer route"
+        );
+        ensure!(
+            matches!(
+                message.get_msg_type(),
+                MessageType::MsgAppend
+                    | MessageType::MsgAppendResponse
+                    | MessageType::MsgHeartbeat
+                    | MessageType::MsgHeartbeatResponse
+                    | MessageType::MsgRequestVote
+                    | MessageType::MsgRequestVoteResponse
+                    | MessageType::MsgRequestPreVote
+                    | MessageType::MsgRequestPreVoteResponse
+            ),
+            "unsupported peer message"
+        );
+        ensure!(
+            message.compute_size() <= 16 * 1024 * 1024 && message.snapshot.is_none(),
+            "oversized message or unsupported snapshot"
+        );
+        for entry in &message.entries {
+            state::validate_entry(&self.layout, packet.group, &self.limits, entry, true)?;
+        }
+        self.groups
+            .get_mut(&packet.group)
+            .context("unhosted group")?
+            .raw
+            .step(message)?;
+        Ok(())
+    }
+
+    /// Performs blocking storage I/O. The transport may drop output messages;
+    /// Raft will retransmit. On error NONE of this call's events are released.
+    pub fn poll(&mut self) -> Result<Events> {
+        self.healthy()?;
+        let result = self.drive();
+        if result.is_err() {
+            self.failed = true;
+        }
+        result
+    }
+
+    fn drive(&mut self) -> Result<Events> {
+        let mut events = Events::default();
+        for (id, group) in &mut self.groups {
+            let mut outbound = Vec::new();
+            while group.raw.has_ready() {
+                let mut ready = group.raw.ready();
+                ensure!(ready.snapshot().is_empty(), "snapshot integration required");
+                group.raw.store().persist(ready.entries(), ready.hs())?;
+                outbound.extend(ready.take_messages());
+                outbound.extend(ready.take_persisted_messages());
+                apply(
+                    group,
+                    *id,
+                    ready.take_committed_entries(),
+                    &self.limits,
+                    &mut events,
+                )?;
+                let mut light = group.raw.advance(ready);
+                if light.commit_index().is_some() {
+                    group
+                        .raw
+                        .store()
+                        .persist(&[], Some(&group.raw.raft.hard_state()))?;
+                }
+                outbound.extend(light.take_messages());
+                apply(
+                    group,
+                    *id,
+                    light.take_committed_entries(),
+                    &self.limits,
+                    &mut events,
+                )?;
+                group.raw.advance_apply();
+            }
+            // Old-term pending entries may still commit after leader loss. They
+            // must not be reported as a definite rejection or leak forever.
+            let uncertain: Vec<_> = group
+                .pending
+                .keys()
+                .copied()
+                .filter(|(_, term)| {
+                    group.raw.raft.state != StateRole::Leader || *term != group.raw.raft.term
+                })
+                .collect();
+            for key in uncertain {
+                let (request, _) = group.pending.remove(&key).unwrap();
+                events.completions.push(Completion {
+                    group: *id,
+                    request,
+                    outcome: Outcome::Unknown,
+                });
+            }
+            events
+                .messages
+                .extend(outbound.into_iter().map(|message| Packet {
+                    cluster: self.layout.cluster,
+                    layout: self.layout_hash,
+                    policy: self.policy,
+                    group: *id,
+                    message,
+                }));
+        }
+        self.healthy()?;
+        Ok(events)
+    }
+
+    /// Local accepted rows at this durable applied cut. This is not a quorum
+    /// read or a producer ACK and cannot be used to shortcut retry completion.
+    pub fn accepted(&self, group: u64, after: u64, limit: usize) -> Result<Page> {
+        self.healthy()?;
+        ensure!((1..=128).contains(&limit), "invalid accepted page limit");
+        let state = &self.groups.get(&group).context("unhosted group")?.state;
+        ensure!(
+            after <= state.applied,
+            "cursor exceeds local applied position"
+        );
+        let mut remaining = state
+            .accepted
+            .range((std::ops::Bound::Excluded(after), std::ops::Bound::Unbounded));
+        let entries: Vec<_> = remaining
+            .by_ref()
+            .take(limit)
+            .map(|(index, meta)| (*index, meta.clone()))
+            .collect();
+        let through = if remaining.next().is_some() {
+            entries.last().expect("positive page limit").0
+        } else {
+            state.applied
+        };
+        Ok(Page { entries, through })
+    }
+
+    pub fn read_accepted(&self, group: u64, index: u64) -> Result<Option<Batch>> {
+        self.healthy()?;
+        let group = self.groups.get(&group).context("unhosted group")?;
+        let Some(meta) = group.state.accepted.get(&index) else {
+            return Ok(None);
+        };
+        let mut entries = group.raw.store().entries(
+            index,
+            index + 1,
+            None,
+            raft::GetEntriesContext::empty(false),
+        )?;
+        let entry = entries.pop().context("missing accepted payload")?;
+        ensure!(
+            &state::decode_meta(&entry.context)? == meta,
+            "accepted metadata mismatch"
+        );
+        meta.validate_payload(&entry.data, self.limits.max_record_bytes as usize)?;
+        Ok(Some(Batch {
+            meta: meta.clone(),
+            body: entry.data,
+        }))
+    }
+}
+
+fn apply(
+    group: &mut Group,
+    id: u64,
+    entries: Vec<Entry>,
+    limits: &Limits,
+    events: &mut Events,
+) -> Result<()> {
+    let durable = group.raw.store().initial_state()?.hard_state.commit;
+    for entry in entries {
+        ensure!(
+            entry.index <= durable,
+            "application precedes durable commit"
+        );
+        let result = group.state.apply(&entry, limits)?;
+        if let Some((request, meta)) = group.pending.remove(&(entry.index, entry.term)) {
+            ensure!(
+                state::decode_meta(&entry.context)? == meta,
+                "pending proposal identity changed"
+            );
+            events.completions.push(Completion {
+                group: id,
+                request,
+                outcome: Outcome::Committed(result.context("pending proposal applied as blank")?),
+            });
+        }
+    }
+    Ok(())
+}
diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs
new file mode 100644
index 00000000..451fbf74
--- /dev/null
+++ b/crates/lance-context-wal/src/replication/state.rs
@@ -0,0 +1,108 @@
+use super::*;
+use crate::record::RecordKind;
+use bincode::Options;
+use raft_proto::eraftpb::EntryType;
+
+fn codec() -> impl Options {
+    bincode::DefaultOptions::new()
+        .with_fixint_encoding()
+        .with_little_endian()
+        .with_limit(256)
+        .reject_trailing_bytes()
+}
+pub(super) fn encode_meta(meta: &BatchMeta) -> Result<Vec<u8>> {
+    Ok(codec().serialize(meta)?)
+}
+pub(super) fn decode_meta(bytes: &[u8]) -> Result<BatchMeta> {
+    ensure!(bytes.len() <= 256, "oversized record metadata");
+    Ok(codec().deserialize(bytes)?)
+}
+pub(super) fn validate_meta(
+    layout: &Layout,
+    group: u64,
+    limits: &Limits,
+    meta: &BatchMeta,
+    body: &[u8],
+    digest: bool,
+) -> Result<()> {
+    ensure!(
+        meta.kind == RecordKind::Data
+            && !meta.stream.is_nil()
+            && !meta.producer.is_nil()
+            && meta.sequence < u64::MAX
+            && layout.group_for_bucket(meta.bucket)? == group,
+        "invalid data identity, record kind or group placement"
+    );
+    if digest {
+        meta.validate_payload(body, limits.max_record_bytes as usize)?;
+    } else {
+        meta.check_body_length(body.len(), limits.max_record_bytes as usize)?;
+    }
+    Ok(())
+}
+pub(super) fn validate_entry(
+    layout: &Layout,
+    group: u64,
+    limits: &Limits,
+    entry: &Entry,
+    digest: bool,
+) -> Result<()> {
+    ensure!(
+        entry.get_entry_type() == EntryType::EntryNormal,
+        "membership changes not integrated"
+    );
+    if entry.data.is_empty() {
+        ensure!(
+            entry.context.is_empty(),
+            "empty data with nonempty metadata"
+        );
+        return Ok(());
+    }
+    validate_meta(
+        layout,
+        group,
+        limits,
+        &decode_meta(&entry.context)?,
+        &entry.data,
+        digest,
+    )
+}
+
+#[derive(Default)]
+pub(super) struct State {
+    pub applied: u64,
+    pub accepted: BTreeMap<u64, BatchMeta>,
+    pub producers: BTreeMap<(Uuid, u32, Uuid), (BatchMeta, u64)>,
+}
+impl State {
+    pub fn apply(&mut self, entry: &Entry, limits: &Limits) -> Result<Option<Applied>> {
+        ensure!(
+            entry.index == self.applied + 1,
+            "noncontiguous committed apply"
+        );
+        self.applied = entry.index;
+        if entry.data.is_empty() {
+            return Ok(None);
+        }
+        let meta = decode_meta(&entry.context)?;
+        let key = (meta.stream, meta.bucket, meta.producer);
+        let previous = self.producers.get(&key);
+        if let Some((old, index)) = previous {
+            if *old == meta {
+                return Ok(Some(Applied::Accepted { index: *index }));
+            }
+        }
+        let expected_sequence = previous.map_or(1, |(old, _)| old.sequence + 1);
+        if meta.sequence != expected_sequence {
+            return Ok(Some(Applied::Rejected { expected_sequence }));
+        }
+        if self.accepted.len() >= limits.max_records as usize
+            || (previous.is_none() && self.producers.len() >= limits.max_producers as usize)
+        {
+            return Ok(Some(Applied::Capacity));
+        }
+        self.producers.insert(key, (meta.clone(), entry.index));
+        self.accepted.insert(entry.index, meta);
+        Ok(Some(Applied::Accepted { index: entry.index }))
+    }
+}
diff --git a/crates/lance-context-wal/src/replication/tests.rs b/crates/lance-context-wal/src/replication/tests.rs
new file mode 100644
index 00000000..ac877f6e
--- /dev/null
+++ b/crates/lance-context-wal/src/replication/tests.rs
@@ -0,0 +1,272 @@
+use super::*;
+use bytes::Bytes;
+
+fn layout() -> Arc<Layout> {
+    Arc::new(Layout {
+        cluster: Uuid::from_u128(1),
+        nodes: (1..=3)
+            .map(|id| (id, format!("wal-{id}.example.invalid:7000")))
+            .collect(),
+        groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]),
+        buckets: vec![10, 20],
+    })
+}
+
+fn batch(sequence: u64) -> Batch {
+    Batch::new(
+        Uuid::from_u128(2),
+        0,
+        Uuid::from_u128(3),
+        sequence,
+        Bytes::from(vec![sequence as u8; 4096]),
+    )
+}
+
+fn pump(nodes: &mut [Node]) -> Result<Vec<Completion>> {
+    let mut completions = Vec::new();
+    for _ in 0..100 {
+        let mut messages = Vec::new();
+        for node in &mut *nodes {
+            let events = node.poll()?;
+            completions.extend(events.completions);
+            messages.extend(events.messages);
+        }
+        if messages.is_empty() {
+            return Ok(completions);
+        }
+        for packet in messages {
+            nodes[packet.message.to as usize - 1].step(packet)?;
+        }
+    }
+    anyhow::bail!("message pump did not settle")
+}
+
+#[test]
+fn no_quorum_ack_before_peer_persistence_and_proposal_keeps_body_allocation() -> Result<()> {
+    let directory = tempfile::tempdir()?;
+    let mut nodes = (1..=3)
+        .map(|id| {
+            Node::open(
+                &directory.path().join(id.to_string()),
+                layout(),
+                id,
+                Limits::default(),
+            )
+        })
+        .collect::<Result<Vec<_>>>()?;
+    nodes[0].campaign(10)?;
+    pump(&mut nodes)?;
+    let body = batch(1);
+    let address = body.body.as_ptr();
+    nodes[0].propose(10, 7, body)?;
+    let unstable = &nodes[0].groups[&10].raw.raft.raft_log.unstable.entries;
+    assert_eq!(unstable.last().unwrap().data.as_ptr(), address);
+    let events = nodes[0].poll()?;
+    assert!(events.completions.is_empty());
+    assert_eq!(nodes[0].status(10)?.accepted_records, 0);
+    for packet in events.messages.into_iter().filter(|p| p.message.to == 2) {
+        nodes[1].step(packet)?;
+    }
+    // Receiving the append in memory is not a durable follower response.
+    assert!(nodes[0].poll()?.completions.is_empty());
+    nodes[1].groups[&10].raw.store().fail_next_sync_for_test();
+    assert!(nodes[1].poll().is_err());
+    assert!(
+        nodes[1].status(20).is_err(),
+        "one engine failure fences every group"
+    );
+    assert!(nodes[1].poll().is_err());
+    assert!(nodes[0].poll()?.completions.is_empty());
+    Ok(())
+}
+
+#[test]
+fn committed_retries_preserve_original_index_and_scope_with_bounded_state() -> Result<()> {
+    let directory = tempfile::tempdir()?;
+    let limits = Limits {
+        max_producers: 1,
+        max_records: 2,
+        ..Default::default()
+    };
+    let mut nodes = (1..=3)
+        .map(|id| {
+            Node::open(
+                &directory.path().join(id.to_string()),
+                layout(),
+                id,
+                limits.clone(),
+            )
+        })
+        .collect::<Result<Vec<_>>>()?;
+    nodes[0].campaign(10)?;
+    pump(&mut nodes)?;
+    nodes[0].propose(10, 1, batch(1))?;
+    let first = pump(&mut nodes)?.remove(0).outcome;
+    assert!(matches!(
+        first,
+        Outcome::Committed(Applied::Accepted { .. })
+    ));
+    nodes[0].propose(10, 2, batch(1))?;
+    assert_eq!(pump(&mut nodes)?.remove(0).outcome, first);
+    let mut changed = batch(1);
+    changed.body = Bytes::from(vec![4; 4096]);
+    changed.meta.digest = Sha256::digest(&changed.body).into();
+    nodes[0].propose(10, 3, changed)?;
+    assert_eq!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Rejected {
+            expected_sequence: 2
+        })
+    );
+    nodes[0].propose(10, 4, batch(3))?;
+    assert_eq!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Rejected {
+            expected_sequence: 2
+        })
+    );
+    let mut other = batch(1);
+    other.meta.stream = Uuid::from_u128(9);
+    nodes[0].propose(10, 5, other)?;
+    assert_eq!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Capacity)
+    );
+    nodes[0].propose(10, 6, batch(2))?;
+    assert!(matches!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Accepted { .. })
+    ));
+    nodes[0].propose(10, 7, batch(3))?;
+    assert_eq!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Capacity)
+    );
+    // Latest retry still works at capacity, and old sequences do not.
+    nodes[0].propose(10, 8, batch(2))?;
+    assert!(matches!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Accepted { .. })
+    ));
+    for node in &nodes {
+        assert_eq!(node.status(10)?.accepted_records, 2);
+    }
+    Ok(())
+}
+
+#[test]
+fn foreign_policy_invalid_input_and_unidentified_logs_are_rejected() -> Result<()> {
+    let directory = tempfile::tempdir()?;
+    let path = directory.path().join("log");
+    let mut node = Node::open(&path, layout(), 1, Limits::default())?;
+    let mut invalid = batch(1);
+    invalid.meta.bucket = 1;
+    assert!(node.propose(10, 1, invalid).is_err());
+    node.campaign(10)?;
+    let mut packet = node.poll()?.messages.remove(0);
+    packet.message.to = 1;
+    packet.message.from = 2;
+    packet.policy[0] ^= 1;
+    assert!(node.step(packet).is_err());
+    assert_eq!(node.status(10)?.accepted_records, 0);
+    drop(node);
+    let limits = Limits {
+        max_records: 1,
+        ..Default::default()
+    };
+    assert!(Node::open(&path, layout(), 1, limits).is_err());
+    Node::open(&path, layout(), 1, Limits::default())?;
+    let path = directory.path().join("unidentified");
+    let log = PhysicalLog::open(&path, &layout(), 1)?;
+    log.group(10)?.persist(
+        &[Entry {
+            index: 1,
+            term: 1,
+            data: vec![1].into(),
+            ..Default::default()
+        }],
+        Some(&raft_proto::eraftpb::HardState {
+            term: 1,
+            vote: 1,
+            commit: 1,
+            ..Default::default()
+        }),
+    )?;
+    drop(log);
+    assert!(Node::open(&path, layout(), 1, Limits::default()).is_err());
+    Ok(())
+}
+
+#[test]
+fn accepted_pages_capture_progress_without_skipping_later_appends() -> Result<()> {
+    let directory = tempfile::tempdir()?;
+    let mut nodes = (1..=3)
+        .map(|id| {
+            Node::open(
+                &directory.path().join(id.to_string()),
+                layout(),
+                id,
+                Limits::default(),
+            )
+        })
+        .collect::<Result<Vec<_>>>()?;
+    nodes[0].campaign(10)?;
+    pump(&mut nodes)?;
+    // The election's blank entry advances even an empty page's cursor.
+    let blank = nodes[0].accepted(10, 0, 1)?;
+    assert!(blank.entries.is_empty());
+    assert!(blank.through > 0);
+    for sequence in 1..=2 {
+        nodes[0].propose(10, sequence, batch(sequence))?;
+        assert!(matches!(
+            pump(&mut nodes)?.remove(0).outcome,
+            Outcome::Committed(Applied::Accepted { .. })
+        ));
+    }
+    let first = nodes[0].accepted(10, blank.through, 1)?;
+    assert_eq!(first.entries.len(), 1);
+    assert_eq!(first.entries[0].1, batch(1).meta);
+    assert_eq!(first.through, first.entries[0].0);
+    let second = nodes[0].accepted(10, first.through, 1)?;
+    assert_eq!(second.entries.len(), 1);
+    assert_eq!(second.entries[0].1, batch(2).meta);
+    let captured = second.through;
+    // A concurrent append after page capture must be read on the next call.
+    nodes[0].propose(10, 3, batch(3))?;
+    assert!(matches!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Accepted { .. })
+    ));
+    assert!(nodes[0].status(10)?.applied > captured);
+    let third = nodes[0].accepted(10, captured, 1)?;
+    assert_eq!(third.entries.len(), 1);
+    assert_eq!(third.entries[0].1, batch(3).meta);
+    // Rejected and duplicate entries make progress without new accepted rows.
+    nodes[0].propose(10, 4, batch(5))?;
+    assert_eq!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Rejected {
+            expected_sequence: 4
+        })
+    );
+    nodes[0].propose(10, 5, batch(3))?;
+    assert_eq!(
+        pump(&mut nodes)?.remove(0).outcome,
+        Outcome::Committed(Applied::Accepted {
+            index: third.entries[0].0
+        })
+    );
+    let tail = nodes[0].accepted(10, third.through, 1)?;
+    assert!(tail.entries.is_empty());
+    assert_eq!(tail.through, nodes[0].status(10)?.applied);
+    assert!(tail.through > third.through);
+    assert_eq!(nodes[0].accepted(10, tail.through, 1)?, tail);
+    // A full page with no more accepted rows can still cover the non-accepted tail.
+    let last = nodes[0].accepted(10, captured, 1)?;
+    assert_eq!(last.entries, third.entries);
+    assert_eq!(last.through, tail.through);
+    assert!(nodes[0].accepted(10, tail.through + 1, 1).is_err());
+    assert!(nodes[0].accepted(10, 0, 0).is_err());
+    assert!(nodes[0].accepted(10, 0, 129).is_err());
+    Ok(())
+}
diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs
new file mode 100644
index 00000000..3b91948f
--- /dev/null
+++ b/crates/lance-context-wal/tests/replication_process.rs
@@ -0,0 +1,441 @@
+#![cfg(feature = "replication")]
+//! Real child processes and files; the parent relays Raft packets over TCP and
+//! deliberately drops selected links. No simulated consensus or storage engine.
+use anyhow::{ensure, Context, Result};
+use bytes::Bytes;
+use lance_context_wal::{
+    placement::Layout,
+    record::Batch,
+    replication::{Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, Status},
+};
+use protobuf::Message as _;
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use std::{
+    collections::{BTreeMap, VecDeque},
+    fs::File,
+    io::{Read, Write},
+    net::{TcpListener, TcpStream},
+    path::{Path, PathBuf},
+    process::{Child, Command, Stdio},
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use uuid::Uuid;
+
+#[derive(Serialize, Deserialize)]
+struct WirePacket {
+    cluster: Uuid,
+    layout: [u8; 32],
+    policy: [u8; 32],
+    group: u64,
+    message: Vec<u8>,
+}
+impl WirePacket {
+    fn encode(packet: Packet) -> Result<Self> {
+        Ok(Self {
+            cluster: packet.cluster,
+            layout: packet.layout,
+            policy: packet.policy,
+            group: packet.group,
+            message: packet.message.write_to_bytes()?,
+        })
+    }
+    fn decode(self) -> Result<Packet> {
+        Ok(Packet {
+            cluster: self.cluster,
+            layout: self.layout,
+            policy: self.policy,
+            group: self.group,
+            message: raft_proto::eraftpb::Message::parse_from_bytes(&self.message)?,
+        })
+    }
+}
+#[derive(Serialize, Deserialize)]
+enum Request {
+    Campaign(u64),
+    Propose(u64, u64, Batch),
+    Step(WirePacket),
+    Tick,
+    Status(u64),
+    Read(u64, u64),
+    Page(u64),
+}
+#[derive(Serialize, Deserialize)]
+enum Response {
+    Events(Vec<WirePacket>, Vec<Completion>),
+    Status(Status),
+    Record(Option<Batch>),
+    Page(Page),
+    Error(String),
+}
+fn send<T: Serialize>(stream: &mut TcpStream, value: &T) -> Result<()> {
+    let bytes = bincode::serialize(value)?;
+    ensure!(bytes.len() <= 32 * 1024 * 1024, "oversized test frame");
+    stream.write_all(&(bytes.len() as u32).to_le_bytes())?;
+    stream.write_all(&bytes)?;
+    Ok(())
+}
+fn receive<T: DeserializeOwned>(stream: &mut TcpStream) -> Result<T> {
+    let mut len = [0u8; 4];
+    stream.read_exact(&mut len)?;
+    let len = u32::from_le_bytes(len) as usize;
+    ensure!(len <= 32 * 1024 * 1024, "oversized test frame");
+    let mut bytes = vec![0; len];
+    stream.read_exact(&mut bytes)?;
+    Ok(bincode::deserialize(&bytes)?)
+}
+fn run(node: &mut Node, request: Request) -> Result<Response> {
+    match request {
+        Request::Campaign(group) => node.campaign(group)?,
+        Request::Propose(group, token, batch) => node.propose(group, token, batch)?,
+        Request::Step(packet) => node.step(packet.decode()?)?,
+        Request::Tick => node.tick()?,
+        Request::Status(group) => return Ok(Response::Status(node.status(group)?)),
+        Request::Read(group, index) => {
+            return Ok(Response::Record(node.read_accepted(group, index)?))
+        }
+        Request::Page(group) => return Ok(Response::Page(node.accepted(group, 0, 128)?)),
+    }
+    let Events {
+        messages,
+        completions,
+    } = node.poll()?;
+    Ok(Response::Events(
+        messages
+            .into_iter()
+            .map(WirePacket::encode)
+            .collect::<Result<_>>()?,
+        completions,
+    ))
+}
+
+#[test]
+#[ignore = "child fixture, invoked by the process qualification"]
+fn replication_peer_fixture() -> Result<()> {
+    let Ok(root) = std::env::var("CONTEXT_REPLICATION_FIXTURE") else {
+        return Ok(());
+    };
+    let root = PathBuf::from(root);
+    let id = std::env::var("CONTEXT_REPLICATION_NODE")?.parse()?;
+    let listener = TcpListener::bind("127.0.0.1:0")?;
+    let mut node = Node::open(
+        &root.join(format!("data-{id}")),
+        Arc::new(serde_json::from_slice(&std::fs::read(
+            root.join("layout.json"),
+        )?)?),
+        id,
+        Limits::default(),
+    )?;
+    let address = PathBuf::from(std::env::var("CONTEXT_REPLICATION_ADDRESS_FILE")?);
+    let temporary = address.with_extension("tmp");
+    std::fs::write(&temporary, listener.local_addr()?.to_string())?;
+    std::fs::rename(temporary, address)?;
+    let (mut stream, _) = listener.accept()?;
+    stream.set_nodelay(true)?;
+    stream.set_read_timeout(Some(Duration::from_secs(30)))?;
+    stream.set_write_timeout(Some(Duration::from_secs(10)))?;
+    loop {
+        let request = receive(&mut stream)?;
+        let response =
+            run(&mut node, request).unwrap_or_else(|error| Response::Error(format!("{error:#}")));
+        send(&mut stream, &response)?;
+    }
+}
+
+struct Peer {
+    child: Child,
+    stream: TcpStream,
+}
+impl Peer {
+    fn spawn(root: &Path, id: u64, incarnation: u64) -> Result<Self> {
+        let address = root.join(format!("address-{id}-{incarnation}"));
+        let mut child = Command::new(std::env::current_exe()?)
+            .args([
+                "--exact",
+                "replication_peer_fixture",
+                "--ignored",
+                "--nocapture",
+            ])
+            .env("CONTEXT_REPLICATION_FIXTURE", root)
+            .env("CONTEXT_REPLICATION_NODE", id.to_string())
+            .env("CONTEXT_REPLICATION_ADDRESS_FILE", &address)
+            .stdout(Stdio::from(File::create(
+                root.join(format!("stdout-{id}-{incarnation}")),
+            )?))
+            .stderr(Stdio::from(File::create(
+                root.join(format!("stderr-{id}-{incarnation}")),
+            )?))
+            .spawn()?;
+        let result = (|| -> Result<TcpStream> {
+            let deadline = Instant::now() + Duration::from_secs(10);
+            while !address.exists() {
+                ensure!(child.try_wait()?.is_none(), "peer exited during startup");
+                ensure!(Instant::now() < deadline, "peer startup outcome unknown");
+                std::thread::sleep(Duration::from_millis(5));
+            }
+            let stream = TcpStream::connect(std::fs::read_to_string(address)?)?;
+            stream.set_nodelay(true)?;
+            stream.set_read_timeout(Some(Duration::from_secs(10)))?;
+            stream.set_write_timeout(Some(Duration::from_secs(10)))?;
+            Ok(stream)
+        })();
+        match result {
+            Ok(stream) => Ok(Self { child, stream }),
+            Err(error) => {
+                let _ = child.kill();
+                let _ = child.wait();
+                Err(error)
+            }
+        }
+    }
+    fn call(&mut self, request: Request) -> Result<Response> {
+        send(&mut self.stream, &request)?;
+        match receive(&mut self.stream)? {
+            Response::Error(error) => anyhow::bail!("peer rejected request: {error}"),
+            response => Ok(response),
+        }
+    }
+}
+impl Drop for Peer {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+struct Cluster {
+    // Peers must die before TempDir removes their exclusively owned files.
+    peers: BTreeMap<u64, Peer>,
+    root: tempfile::TempDir,
+    incarnations: BTreeMap<u64, u64>,
+    queue: VecDeque<Packet>,
+    completions: Vec<Completion>,
+}
+impl Cluster {
+    fn new() -> Result<Self> {
+        let root = tempfile::tempdir()?;
+        let layout = Layout {
+            cluster: Uuid::from_u128(50),
+            nodes: (1..=3)
+                .map(|id| (id, format!("wal-{id}.example.invalid:7800")))
+                .collect(),
+            groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]),
+            buckets: vec![10, 20],
+        };
+        std::fs::write(
+            root.path().join("layout.json"),
+            serde_json::to_vec(&layout)?,
+        )?;
+        let mut this = Self {
+            peers: BTreeMap::new(),
+            root,
+            incarnations: BTreeMap::new(),
+            queue: VecDeque::new(),
+            completions: Vec::new(),
+        };
+        for id in 1..=3 {
+            this.start(id)?;
+        }
+        Ok(this)
+    }
+    fn start(&mut self, id: u64) -> Result<()> {
+        ensure!(
+            !self.peers.contains_key(&id),
+            "cannot launch an existing peer"
+        );
+        let incarnation = self.incarnations.entry(id).or_default();
+        *incarnation += 1;
+        self.peers
+            .insert(id, Peer::spawn(self.root.path(), id, *incarnation)?);
+        Ok(())
+    }
+    fn command(&mut self, id: u64, request: Request) -> Result<()> {
+        let Response::Events(packets, completions) = self
+            .peers
+            .get_mut(&id)
+            .context("missing peer")?
+            .call(request)?
+        else {
+            anyhow::bail!("wrong reply");
+        };
+        for packet in packets {
+            self.queue.push_back(packet.decode()?);
+        }
+        self.completions.extend(completions);
+        Ok(())
+    }
+    fn pump(&mut self, allowed: &[u64]) -> Result<()> {
+        for _ in 0..5000 {
+            let Some(packet) = self.queue.pop_front() else {
+                return Ok(());
+            };
+            if allowed.contains(&packet.message.from)
+                && allowed.contains(&packet.message.to)
+                && self.peers.contains_key(&packet.message.to)
+            {
+                self.command(
+                    packet.message.to,
+                    Request::Step(WirePacket::encode(packet)?),
+                )?;
+            }
+        }
+        anyhow::bail!("test message queue did not settle")
+    }
+    fn outcome(&mut self, token: u64) -> Option<Outcome> {
+        self.completions
+            .iter()
+            .position(|c| c.request == token)
+            .map(|position| self.completions.remove(position).outcome)
+    }
+    fn elect(&mut self, id: u64, group: u64, allowed: &[u64]) -> Result<u64> {
+        self.command(id, Request::Campaign(group))?;
+        self.pump(allowed)?;
+        for _ in 0..100 {
+            let mut views = Vec::new();
+            for node in allowed {
+                let Response::Status(status) = self
+                    .peers
+                    .get_mut(node)
+                    .unwrap()
+                    .call(Request::Status(group))?
+                else {
+                    anyhow::bail!("wrong status");
+                };
+                views.push(status);
+            }
+            let leader = views[0].leader;
+            if allowed.contains(&leader)
+                && views
+                    .iter()
+                    .all(|s| s.leader == leader && s.term == views[0].term)
+            {
+                return Ok(leader);
+            }
+            // Expire the old election lease; do not assume which randomized
+            // timer wins when both eligible replicas have identical logs.
+            for node in allowed {
+                self.command(*node, Request::Tick)?;
+            }
+            self.pump(allowed)?;
+        }
+        anyhow::bail!("majority did not elect a leader")
+    }
+}
+
+fn input(bucket: u32, sequence: u64, fill: u8) -> Batch {
+    Batch::new(
+        Uuid::from_u128(80),
+        bucket,
+        Uuid::from_u128(90),
+        sequence,
+        Bytes::from(vec![fill; 64 * 1024]),
+    )
+}
+
+#[test]
+fn durable_quorum_retry_leader_loss_suffix_replacement_and_full_restart() -> Result<()> {
+    let mut c = Cluster::new()?;
+    c.command(1, Request::Campaign(10))?;
+    c.pump(&[1, 2, 3])?;
+    let first = input(0, 1, 7);
+    c.command(1, Request::Propose(10, 1, first.clone()))?;
+    assert!(
+        c.outcome(1).is_none(),
+        "local append was mistaken for quorum ACK"
+    );
+    c.pump(&[1, 2])?;
+    let first_ack = c.outcome(1).context("no majority ACK")?;
+    assert!(matches!(
+        first_ack,
+        Outcome::Committed(Applied::Accepted { .. })
+    ));
+    // Only one follower held the acknowledged input when its leader was killed.
+    drop(c.peers.remove(&1));
+    assert_eq!(
+        c.elect(2, 10, &[2, 3])?,
+        2,
+        "only replica 2 has the committed suffix"
+    );
+    c.command(2, Request::Propose(10, 2, first.clone()))?;
+    c.pump(&[2, 3])?;
+    assert_eq!(
+        c.outcome(2),
+        Some(first_ack.clone()),
+        "ambiguous client retry lost original receipt"
+    );
+    c.start(1)?;
+    for _ in 0..4 {
+        c.command(2, Request::Tick)?;
+        c.pump(&[1, 2, 3])?;
+    }
+    // A still-self-declared but isolated leader cannot acknowledge a local retry.
+    c.command(2, Request::Propose(10, 3, first.clone()))?;
+    c.pump(&[2])?;
+    assert!(c.outcome(3).is_none());
+    let abandoned = input(0, 2, 99);
+    c.command(2, Request::Propose(10, 4, abandoned))?;
+    c.pump(&[2])?;
+    assert!(c.outcome(4).is_none());
+    let replacement = c.elect(1, 10, &[1, 3])?;
+    let second = input(0, 2, 8);
+    c.command(replacement, Request::Propose(10, 5, second.clone()))?;
+    c.pump(&[1, 3])?;
+    assert!(matches!(
+        c.outcome(5),
+        Some(Outcome::Committed(Applied::Accepted { .. }))
+    ));
+    for _ in 0..4 {
+        c.command(replacement, Request::Tick)?;
+        c.pump(&[1, 2, 3])?;
+    }
+    assert_eq!(c.outcome(3), Some(Outcome::Unknown));
+    assert_eq!(
+        c.outcome(4),
+        Some(Outcome::Unknown),
+        "replaced suffix completed the wrong input"
+    );
+    let other_leader = c.elect(replacement, 20, &[1, 2, 3])?;
+    let separate = input(1, 1, 9);
+    c.command(other_leader, Request::Propose(20, 6, separate.clone()))?;
+    c.pump(&[1, 2, 3])?;
+    let other_outcome = c.outcome(6);
+    assert!(
+        matches!(
+            other_outcome,
+            Some(Outcome::Committed(Applied::Accepted { .. }))
+        ),
+        "other group outcome: {other_outcome:?}"
+    );
+    // Kill every process, retain its original files, then reconstruct metadata
+    // and exact payloads solely from its durable committed prefix.
+    c.queue.clear();
+    c.peers.clear();
+    for id in 1..=3 {
+        c.start(id)?;
+    }
+    for peer in c.peers.values_mut() {
+        for (group, expected) in [(10, vec![&first, &second]), (20, vec![&separate])] {
+            let Response::Page(page) = peer.call(Request::Page(group))? else {
+                anyhow::bail!("wrong page");
+            };
+            assert_eq!(page.entries.len(), expected.len());
+            for ((index, meta), batch) in page.entries.into_iter().zip(expected) {
+                assert_eq!(&meta, &batch.meta);
+                let Response::Record(Some(actual)) = peer.call(Request::Read(group, index))? else {
+                    anyhow::bail!("missing record");
+                };
+                assert_eq!(&actual, batch);
+            }
+        }
+    }
+    c.command(1, Request::Campaign(10))?;
+    c.pump(&[1, 2, 3])?;
+    c.command(1, Request::Propose(10, 7, second))?;
+    c.pump(&[1, 2, 3])?;
+    assert!(matches!(
+        c.outcome(7),
+        Some(Outcome::Committed(Applied::Accepted { .. }))
+    ));
+    println!("verified real RF3 processes: durable majority ACK, latest retry, isolated old leader, replaced uncommitted suffix, two groups and full process restart with exact payloads");
+    Ok(())
+}
diff --git a/docs/design/replicated-wal-append.md b/docs/design/replicated-wal-append.md
new file mode 100644
index 00000000..c94c51c5
--- /dev/null
+++ b/docs/design/replicated-wal-append.md
@@ -0,0 +1,108 @@
+# Replicated WAL append component
+
+Tracked by [#378](https://github.com/lance-format/lance-context/issues/378).
+The optional `lance-context-wal/replication` feature connects raft-rs to the shared
+durable log from #399. It is an embeddable Rust component for fixed three-voter
+groups. The existing production ingestion path is not changed.
+
+## Scheduling and ownership
+
+Open a `replication::Node` with a fresh directory, immutable `placement::Layout`,
+node ID and `Limits`. The component binds its application format and limits in
+the durable group log. Reopen requires identical layout, identity and policy;
+unidentified logs containing entries cannot be imported implicitly. Directories
+from the earlier experimental service have a different format.
+
+The owner supplies ticks, bounded network queues and a worker for blocking log
+I/O. `into_groups` consumes the aggregate owner and transfers each RawNode to an
+independent worker while keeping the physical engine and sync/failure fence
+shared. It never clones a group's state machine. Transport routes `Packet`s by
+group and recipient, preserving the cluster/layout/policy scope. Dropping a Raft
+packet is safe; retry/retransmission is Raft's responsibility. Packet scope checks
+are not authenticated peer identity, nor a wire-level allocation limit.
+
+## Append and ACK boundary
+
+`propose(group, request, batch)` returning `Ok` means that the local leader queued
+the input. It is not an ACK. `poll` performs this sequence for each Ready:
+
+1. Persist and sync entries and HardState through the shared engine boundary.
+2. Collect Ready messages and apply its committed entries only through the
+   durable commit index.
+3. Advance Ready, persist any LightReady commit-index update, then apply that
+   committed suffix and advance the applied position.
+4. Return transport messages and request completions only after all storage
+   operations in this poll have succeeded.
+
+A committed completion is consequently backed by Raft's majority replication
+and the local durable commit boundary. An engine error fences every group sharing
+that engine. An internal poll error also fences its driver; the owner must stop,
+reconcile unknown requests and reopen. A failed poll returns no partially built
+message/ACK output. Losing leadership returns `Outcome::Unknown` for unresolved
+requests; it does not claim their entries were rejected or cannot later commit.
+
+Raft stores the original shared `Bytes` in `Entry.data`; it does not copy the
+body into the `Vec` accepted by `RawNode::propose`. Only bounded metadata is
+encoded separately. Ingress/peer admission verifies the payload digest without
+decoding Arrow. Replay validates persisted records again. This is not a claim
+that protobuf networking, raft-engine or kernel I/O perform no copies.
+
+## Retry and recovery
+
+The replicated state machine tracks the latest `(metadata, accepted index)` for
+each `(table, bucket, producer)`. New producers start at sequence one; advancing
+requires the next sequence. An exact latest retry returns the original index.
+A changed payload at the same sequence, a stale sequence, or a gap is rejected.
+The table and bucket are part of identity, so one stream cannot consume another
+stream's retry state.
+
+Every retry in this initial component is proposed through consensus again. This
+costs an extra log entry/body write, but prevents an isolated former leader from
+answering directly from local state. A later quorum-read optimization must
+preserve that boundary. Pending local requests bind both log index and term;
+replacement of an uncommitted suffix cannot complete a different request.
+
+Reopen reconstructs accepted metadata and latest producer state by replaying
+only the durable committed prefix. The uncommitted suffix remains available to
+Raft for reconciliation. Accepted pages and payload reads expose a **local
+durable applied view**, not a quorum read, authenticated receipt or proof of Lance
+visibility. Consumers must preserve the group/layout scope. Their use by a fenced
+Lance publisher is subsequent integration.
+
+`accepted(group, after, limit)` captures `Page { entries, through }` together.
+After consuming the page, use its `through` as the next cursor; do not substitute
+a separately read status position, which may include appends absent from the page.
+A truncated page advances only through its last returned record. Otherwise the
+cursor advances through the captured applied position, including rejected,
+duplicate and blank entries. A cursor beyond this replica's applied position is
+rejected, so a lagging replica cannot make the consumer move backward.
+
+## Resource limits and unfinished integration
+
+The immutable policy bounds record size, accepted metadata, tracked producers,
+pending local requests and Raft's uncommitted-data budget. Capacity rejection is
+itself deterministic after commit; an exact latest retry can still succeed when
+metadata capacity is full. Historical accepted metadata and all raw log entries
+are retained in this increment, including rejected or duplicate proposals.
+These are not bounded physical disk usage or an unlimited-duration retention
+policy. Transport must also bound frame decoding and queued messages.
+
+Producer slot reuse, authenticated ownership/revoke, snapshot transfer, replica
+replacement, disk credits, log reclamation and fenced Lance publication remain
+subsequent increments. This feature is not a deployment-ready WAL service and
+does not establish the complete ingestion goal or a performance result.
+
+## Verification
+
+Run `cargo test -p lance-context-wal --features replication --lib --test replication_process`.
+The process test starts actual Rust peers with independent raft-engine files and
+relays their Raft packets over TCP. It withholds links, kills a leader after an
+ACK backed by only one follower, retries after failover, isolates a leader during
+retry, replaces its uncommitted suffix, and restarts every process. Both groups'
+accepted metadata and complete opaque bodies are compared after recovery.
+
+The unit tests inject a real file-sync failure after a follower has received an
+append, verify that no follower response or leader ACK is released, and check
+the shared engine fence. They also check original body allocation ownership,
+exact retries, changed/gapped sequences, metadata capacity and immutable policy
+binding. These exercise real raft-rs and raft-engine, not a consensus model.