diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml
index b8916625..d6eab420 100644
--- a/.github/workflows/rust-test.yml
+++ b/.github/workflows/rust-test.yml
@@ -101,6 +101,10 @@ jobs:
         run: |
           cargo clippy -p lance-context-wal --features peer-transport --all-targets -- -D warnings
           cargo test -p lance-context-wal --features peer-transport --lib --test replication_process
+      - name: Standalone WAL service and automatic maintenance
+        run: |
+          cargo clippy -p lance-context-wal --features service --all-targets -- -D warnings
+          cargo test -p lance-context-wal --features service --test node_service
       - 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/Cargo.lock b/Cargo.lock
index 83f3d314..979fd11c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4317,6 +4317,9 @@ dependencies = [
  "clap",
  "futures",
  "hdrhistogram",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
  "lance",
  "lance-io",
  "lance-table",
diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml
index 18e29e4b..34d5b40b 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]
+service = ["peer-transport", "dep:hyper", "dep:hyper-util", "dep:http-body-util", "dep:clap", "tokio/signal", "tokio/rt-multi-thread"]
 peer-transport = ["runtime", "vectored-io", "dep:tokio-rustls", "tokio/net", "tokio/time", "tokio/macros"]
 runtime = ["replication", "dep:tokio", "tokio/sync", "tokio/rt"]
 # Real Lance materialization/recovery qualification; normal WAL replicas do not link Lance.
@@ -25,6 +26,9 @@ 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]
+hyper = { version = "1", features = ["server", "http1"], optional = true }
+hyper-util = { version = "0.1", features = ["tokio"], optional = true }
+http-body-util = { version = "0.1", optional = true }
 tokio-rustls = { version = "0.26.4", default-features = false, features = ["ring", "tls12"], optional = true }
 lance = { version = "13.0.0", optional = true }
 lance-table = { version = "13.0.0", optional = true }
@@ -64,6 +68,11 @@ bincode = "1.3"
 arrow-ipc = { version = "58", features = ["zstd"] }
 tempfile = "3"
 
+[[bin]]
+name = "context-wal-node"
+path = "src/bin/node.rs"
+required-features = ["service"]
+
 [[bin]]
 name = "context-wal-runner"
 path = "src/bin/runner.rs"
@@ -78,3 +87,7 @@ required-features = ["runner"]
 name = "context-wal-lab"
 path = "src/bin/lab.rs"
 required-features = ["lab"]
+
+[[test]]
+name = "node_service"
+required-features = ["service"]
diff --git a/crates/lance-context-wal/src/bin/node.rs b/crates/lance-context-wal/src/bin/node.rs
new file mode 100644
index 00000000..c68a5128
--- /dev/null
+++ b/crates/lance-context-wal/src/bin/node.rs
@@ -0,0 +1,49 @@
+use anyhow::Result;
+use clap::Parser;
+use lance_context_wal::service::{Config, Service};
+use std::path::PathBuf;
+
+#[derive(Parser)]
+#[command(
+    about = "Independent durable WAL node; configuration and signing identities persist across restarts"
+)]
+struct Args {
+    #[arg(long)]
+    config: PathBuf,
+    #[arg(long,default_value_t=2,value_parser=clap::value_parser!(u16).range(1..=64))]
+    io_threads: u16,
+    #[arg(long,default_value_t=4,value_parser=clap::value_parser!(u16).range(1..=64))]
+    blocking_threads: u16,
+}
+async fn signal() -> Result<()> {
+    #[cfg(unix)]
+    {
+        let mut terminate =
+            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
+        tokio::select! { result=tokio::signal::ctrl_c()=>result?, _=terminate.recv()=>{} }
+    }
+    #[cfg(not(unix))]
+    tokio::signal::ctrl_c().await?;
+    Ok(())
+}
+fn main() -> Result<()> {
+    let args = Args::parse();
+    let config = Config::read(&args.config)?;
+    tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .worker_threads(usize::from(args.io_threads))
+        .max_blocking_threads(usize::from(args.blocking_threads))
+        .build()?
+        .block_on(async {
+            let mut service = Service::start(config).await?;
+            tokio::select! {
+                result=signal()=>{
+                    // Even signal-handler setup failure must join the running
+                    // service before dropping this Tokio runtime.
+                    let stopped=service.shutdown().await;
+                    result.and(stopped)
+                },
+                result=service.wait()=>result,
+            }
+        })
+}
diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs
index e5c7b15a..cc602d49 100644
--- a/crates/lance-context-wal/src/lib.rs
+++ b/crates/lance-context-wal/src/lib.rs
@@ -14,6 +14,9 @@ pub mod raft_log;
 #[cfg(feature = "replication")]
 pub mod replication;
 
+#[cfg(feature = "service")]
+pub mod service;
+
 #[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 d1ac224b..6f0b035a 100644
--- a/crates/lance-context-wal/src/raft_log.rs
+++ b/crates/lance-context-wal/src/raft_log.rs
@@ -348,6 +348,8 @@ pub struct PhysicalLog {
 /// Local physical-file tuning, independent of the replicated layout and policy.
 /// Thresholds trigger explicit maintenance; they are not disk-admission quotas.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct LogOptions {
     pub target_file_bytes: u64,
     /// Native file-count size estimate; oversized batches can exceed it.
diff --git a/crates/lance-context-wal/src/raft_log/budget.rs b/crates/lance-context-wal/src/raft_log/budget.rs
index 1cd3e600..561d1262 100644
--- a/crates/lance-context-wal/src/raft_log/budget.rs
+++ b/crates/lance-context-wal/src/raft_log/budget.rs
@@ -9,6 +9,8 @@ use crate::admission::Snapshot;
 /// accepted Raft/control traffic and physical rewrites. Peer writes must still
 /// persist and can consume this window. This is not a hard filesystem quota.
 #[derive(Clone, Copy, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(deny_unknown_fields))]
 pub struct DiskOptions {
     pub max_bytes: usize,
     pub data_bytes: usize,
diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs
index a31914cb..30d95cab 100644
--- a/crates/lance-context-wal/src/replication.rs
+++ b/crates/lance-context-wal/src/replication.rs
@@ -92,7 +92,7 @@ impl Default for Limits {
     }
 }
 impl Limits {
-    fn identity(&self) -> Result<Vec<u8>> {
+    pub(crate) fn identity(&self) -> Result<Vec<u8>> {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
                 && self.catalog_key.is_none_or(|key| key != [0; 32])
@@ -525,6 +525,18 @@ impl Node {
         })
     }
 
+    pub(crate) fn checkpoint_index(&self, group: u64) -> Result<u64> {
+        self.healthy()?;
+        Ok(self
+            .groups
+            .get(&group)
+            .context("unhosted group")?
+            .raw
+            .store()
+            .first_index()?
+            .saturating_sub(1))
+    }
+
     fn request_pending(&self, request: u64) -> bool {
         self.groups
             .values()
diff --git a/crates/lance-context-wal/src/replication/runtime.rs b/crates/lance-context-wal/src/replication/runtime.rs
index 76d0a3d9..2f95b42e 100644
--- a/crates/lance-context-wal/src/replication/runtime.rs
+++ b/crates/lance-context-wal/src/replication/runtime.rs
@@ -38,6 +38,8 @@ pub mod peer;
 /// Counts do not include caller allocations or an oversized allocation pinned
 /// by a small Bytes slice. The transport must account for those owners too.
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     pub data_requests: usize,
     pub data_bytes: usize,
@@ -45,6 +47,7 @@ pub struct Options {
     pub control_bytes: usize,
     pub peer_requests: usize,
     pub peer_bytes: usize,
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub tick_interval: Duration,
     /// Maximum commands from each lane before polling durable Raft progress.
     pub batch_requests: usize,
@@ -64,7 +67,7 @@ impl Default for Options {
     }
 }
 impl Options {
-    fn validate(&self) -> Result<()> {
+    pub(crate) fn validate(&self) -> Result<()> {
         for (count, bytes) in [
             (self.data_requests, self.data_bytes),
             (self.control_requests, self.control_bytes),
@@ -245,6 +248,8 @@ struct Shared {
 #[derive(Clone, Debug)]
 pub struct Observation {
     pub status: Status,
+    /// Persisted image boundary; local scheduling data, not a quorum result.
+    pub checkpoint_index: u64,
     pub observed_at: Instant,
     pub polling: bool,
     pub stopped: bool,
@@ -477,7 +482,14 @@ impl Runtime {
             layout: node.layout_hash,
             policy: node.policy,
             id: node.id,
-            voters: Arc::new(node.layout.groups.clone()),
+            voters: Arc::new(
+                node.layout
+                    .groups
+                    .iter()
+                    .filter(|(_, voters)| voters.contains(&node.id))
+                    .map(|(&group, voters)| (group, voters.clone()))
+                    .collect(),
+            ),
             max_record: node.limits.max_record_bytes as usize,
         };
         // Validate every initial view before starting any thread. A failure
@@ -488,14 +500,16 @@ impl Runtime {
             .map(|node| {
                 let group = *node.groups.keys().next().context("empty worker group")?;
                 let status = node.status(group)?;
-                Ok((node, group, status))
+                let checkpoint_index = node.checkpoint_index(group)?;
+                Ok((node, group, status, checkpoint_index))
             })
             .collect::<Result<Vec<_>>>()?;
         let mut routes = BTreeMap::new();
         let mut workers: Vec<JoinHandle<Result<()>>> = Vec::new();
-        for (node, group, status) in nodes {
+        for (node, group, status, checkpoint_index) in nodes {
             let observation = Arc::new(Mutex::new(Observation {
                 status,
+                checkpoint_index,
                 observed_at: Instant::now(),
                 polling: false,
                 stopped: false,
@@ -753,8 +767,10 @@ impl Worker {
                     completion();
                 }
                 let status = self.node.status(self.group)?;
+                let checkpoint_index = self.node.checkpoint_index(self.group)?;
                 let mut observation = self.observation.lock().unwrap();
                 observation.status = status;
+                observation.checkpoint_index = checkpoint_index;
                 observation.observed_at = Instant::now();
                 observation.polling = false;
             }
diff --git a/crates/lance-context-wal/src/replication/runtime/client.rs b/crates/lance-context-wal/src/replication/runtime/client.rs
index 59f5105c..1deae448 100644
--- a/crates/lance-context-wal/src/replication/runtime/client.rs
+++ b/crates/lance-context-wal/src/replication/runtime/client.rs
@@ -49,6 +49,8 @@ pub struct Endpoint {
     pub certificate: [u8; 32],
 }
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     /// Logical live request/response bytes, independent for the two lanes.
     /// Caller allocations and allocator/decoded metadata overhead are separate.
@@ -59,6 +61,7 @@ pub struct Options {
     /// Per server listener, shared across all independently running clients.
     pub server_connections: usize,
     /// Entire exchange, including pool/admission waits and durable completion.
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub timeout: Duration,
 }
 impl Default for Options {
@@ -73,7 +76,7 @@ impl Default for Options {
     }
 }
 impl Options {
-    fn validate(&self, max_record: usize) -> Result<()> {
+    pub(crate) fn validate(&self, max_record: usize) -> Result<()> {
         ensure!(
             (1..=32).contains(&self.connections)
                 && (1..=256).contains(&self.server_connections)
@@ -259,6 +262,9 @@ pub struct Server {
     tasks: Vec<JoinHandle<Result<()>>>,
 }
 impl Server {
+    pub fn is_finished(&self) -> bool {
+        self.tasks.iter().any(|task| task.is_finished())
+    }
     /// Trusted service certificates may access this node's configured groups.
     /// End-user authorization and table binding stay at the application ingress.
     pub fn spawn(
diff --git a/crates/lance-context-wal/src/replication/runtime/peer.rs b/crates/lance-context-wal/src/replication/runtime/peer.rs
index f4e53fe8..c04b2cd1 100644
--- a/crates/lance-context-wal/src/replication/runtime/peer.rs
+++ b/crates/lance-context-wal/src/replication/runtime/peer.rs
@@ -50,6 +50,8 @@ pub struct Peer {
 }
 
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     pub receive_bytes: usize,
     pub connections: usize,
@@ -58,7 +60,9 @@ pub struct Options {
     /// follower cannot retain the entire runtime outbox's byte allowance.
     pub peer_bytes: usize,
     /// Deadline for a complete handshake or frame, including bounded admission.
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub io_timeout: Duration,
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub reconnect_delay: Duration,
 }
 impl Default for Options {
@@ -74,7 +78,7 @@ impl Default for Options {
     }
 }
 impl Options {
-    fn validate(&self) -> Result<()> {
+    pub(crate) fn validate(&self) -> Result<()> {
         ensure!(
             (MAX_HEADER + MAX_BODY..=256 << 20).contains(&self.receive_bytes)
                 && (1..=128).contains(&self.connections)
@@ -223,6 +227,10 @@ impl Transport {
     pub fn statistics(&self) -> &Arc<Statistics> {
         &self.statistics
     }
+    /// A listener or peer supervisor ended; the embedding service must drain.
+    pub fn is_finished(&self) -> bool {
+        self.tasks.iter().any(|task| task.is_finished())
+    }
     /// Close network tasks before shutting down/reopening the worker runtime.
     pub async fn shutdown(mut self) -> Result<()> {
         let _ = self.stop.send(true);
diff --git a/crates/lance-context-wal/src/replication/runtime/peer/snapshot.rs b/crates/lance-context-wal/src/replication/runtime/peer/snapshot.rs
index 5dd2f9ec..af961150 100644
--- a/crates/lance-context-wal/src/replication/runtime/peer/snapshot.rs
+++ b/crates/lance-context-wal/src/replication/runtime/peer/snapshot.rs
@@ -43,6 +43,8 @@ pub trait Admission: Send + Sync + 'static {
 }
 
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     /// Total queued and active outgoing transfers across all destinations.
     pub outbound_transfers: usize,
@@ -51,8 +53,10 @@ pub struct Options {
     pub max_image_bytes: u64,
     /// Whole transfer deadline, not reset on every chunk. A filesystem call
     /// already in progress is still joined, not cancelled, at shutdown.
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub transfer_timeout: Duration,
     #[cfg(test)]
+    #[cfg_attr(feature = "service", serde(skip))]
     pub(super) pause_after_payload_chunk: Option<std::path::PathBuf>,
 }
 impl Default for Options {
@@ -148,6 +152,11 @@ pub struct Transport {
     statistics: Arc<Statistics>,
 }
 impl Transport {
+    pub fn is_finished(&self) -> bool {
+        self.supervisor
+            .as_ref()
+            .is_none_or(|task| task.is_finished())
+    }
     pub fn spawn(handle: Handle, listener: TcpListener, settings: Settings) -> Result<Self> {
         ensure!(
             handle.disk.is_some(),
diff --git a/crates/lance-context-wal/src/service.rs b/crates/lance-context-wal/src/service.rs
new file mode 100644
index 00000000..98883dcf
--- /dev/null
+++ b/crates/lance-context-wal/src/service.rs
@@ -0,0 +1,252 @@
+//! Independently deployable, authenticated WAL node with joined maintenance.
+//! Data ACKs still come only from the durable replicated runtime. Health reports
+//! describe local serving state, never quorum freshness or Lance visibility.
+use crate::replication::runtime::{
+    self, client,
+    peer::{self, snapshot},
+    Runtime,
+};
+use anyhow::{ensure, Context, Result};
+use std::{
+    sync::{atomic::Ordering, Arc},
+    time::Duration,
+};
+use tokio::{net::TcpListener, sync::watch, task::JoinHandle};
+
+pub mod config;
+mod health;
+pub mod maintenance;
+mod tls;
+pub use config::Config;
+
+pub(crate) mod duration_millis {
+    use serde::{Deserialize, Deserializer, Serializer};
+    use std::time::Duration;
+    pub fn serialize<S: Serializer>(value: &Duration, s: S) -> Result<S::Ok, S::Error> {
+        s.serialize_u64(u64::try_from(value.as_millis()).map_err(serde::ser::Error::custom)?)
+    }
+    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Duration, D::Error> {
+        Ok(Duration::from_millis(u64::deserialize(d)?))
+    }
+}
+
+#[derive(Default)]
+struct Parts {
+    runtime: Option<Runtime>,
+    peers: Option<peer::Transport>,
+    images: Option<snapshot::Transport>,
+    rpc: Option<client::Server>,
+}
+fn retain(error: &mut Option<anyhow::Error>, result: Result<()>) {
+    if let Err(e) = result {
+        if error.is_none() {
+            *error = Some(e);
+        }
+    }
+}
+impl Parts {
+    async fn stop_clients(&mut self, error: &mut Option<anyhow::Error>) {
+        if let Some(server) = self.rpc.take() {
+            retain(error, server.shutdown().await);
+        }
+    }
+    async fn drain(mut self) -> Result<()> {
+        let mut error = None;
+        self.stop_clients(&mut error).await;
+        if let Some(images) = self.images.take() {
+            retain(&mut error, images.shutdown().await);
+        }
+        if let Some(peers) = self.peers.take() {
+            retain(&mut error, peers.shutdown().await);
+        }
+        if let Some(runtime) = self.runtime.take() {
+            retain(&mut error, runtime.shutdown().await);
+        }
+        error.map_or(Ok(()), Err)
+    }
+}
+
+/// Dropping the service signals its supervisor without aborting it. The
+/// supervisor retains and joins maintenance, file workers and Raft owners.
+/// Await shutdown before reopening the same directory in this process.
+pub struct Service {
+    stop: watch::Sender<bool>,
+    supervisor: Option<JoinHandle<Result<()>>>,
+    handle: runtime::Handle,
+}
+impl Service {
+    pub async fn start(config: Config) -> Result<Self> {
+        config.validate()?;
+        let (peer_listener, snapshot_listener, data, control, health_listener) = tokio::try_join!(
+            TcpListener::bind(config.listen.peer),
+            TcpListener::bind(config.listen.snapshot),
+            TcpListener::bind(config.listen.data),
+            TcpListener::bind(config.listen.control),
+            TcpListener::bind(config.listen.health)
+        )?;
+        let opening = config.clone();
+        // No worker starts until credentials, immutable policy and all listeners
+        // are validated. A startup failure never silently creates new identity.
+        let (node, identity) = tokio::task::spawn_blocking(move || opening.open()).await??;
+        let mut parts = Parts::default();
+        let startup: Result<runtime::Handle> = async {
+            let (runtime, outbox) = Runtime::spawn(node, config.runtime.clone())?;
+            let handle = runtime.handle();
+            parts.runtime = Some(runtime);
+            let (client_tls, server_tls) = identity.configs(peer::ALPN);
+            parts.peers = Some(peer::Transport::spawn(
+                handle.clone(),
+                outbox,
+                peer_listener,
+                config.peers(false),
+                client_tls,
+                server_tls,
+                config.peer.clone(),
+            )?);
+            let (client_tls, server_tls) = identity.configs(snapshot::ALPN);
+            parts.images = Some(snapshot::Transport::spawn(
+                handle.clone(),
+                snapshot_listener,
+                snapshot::Settings {
+                    peers: config.peers(true),
+                    client_tls,
+                    server_tls,
+                    options: config.snapshot.clone(),
+                    admission: None,
+                },
+            )?);
+            parts
+                .peers
+                .as_mut()
+                .unwrap()
+                .attach_snapshots(parts.images.as_ref().unwrap().route())?;
+            let (_, server_tls) = identity.configs(client::ALPN);
+            parts.rpc = Some(client::Server::spawn(
+                handle.clone(),
+                data,
+                control,
+                config.trusted_clients.clone(),
+                server_tls,
+                config.client.clone(),
+            )?);
+            Ok(handle)
+        }
+        .await;
+        let handle = match startup {
+            Ok(handle) => handle,
+            Err(error) => {
+                let cleanup = parts.drain().await;
+                return Err(error.context(format!(
+                    "startup failed; cleanup: {}",
+                    cleanup.map_or_else(|e| e.to_string(), |_| "joined".into())
+                )));
+            }
+        };
+        let (stop, mut shutdown) = watch::channel(false);
+        let state = Arc::new(health::State {
+            node: config.node,
+            handle: handle.clone(),
+            groups: config.groups(),
+            draining: std::sync::atomic::AtomicBool::new(false),
+            maintenance: Arc::new(std::sync::Mutex::new(maintenance::Progress::default())),
+            peers: parts.peers.as_ref().unwrap().statistics().clone(),
+            snapshots: parts.images.as_ref().unwrap().statistics().clone(),
+        });
+        let mut maintenance = tokio::spawn(maintenance::run(
+            handle.clone(),
+            config.groups(),
+            config.maintenance.clone(),
+            state.maintenance.clone(),
+            stop.subscribe(),
+        ));
+        let mut health = tokio::spawn(health::serve(
+            health_listener,
+            state.clone(),
+            stop.subscribe(),
+        ));
+        let signal = stop.clone();
+        let supervisor = tokio::spawn(async move {
+            let mut error = None;
+            let mut tick = tokio::time::interval(Duration::from_millis(200));
+            tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+            loop {
+                tokio::select! { _=shutdown.changed()=>break, _=tick.tick()=>{} }
+                if *shutdown.borrow() {
+                    break;
+                }
+                if parts.peers.as_ref().unwrap().is_finished()
+                    || parts.images.as_ref().unwrap().is_finished()
+                    || parts.rpc.as_ref().unwrap().is_finished()
+                    || maintenance.is_finished()
+                    || health.is_finished()
+                {
+                    error = Some(anyhow::anyhow!("WAL service task stopped unexpectedly"));
+                    break;
+                }
+                for &group in &state.groups {
+                    match state.handle.observation(group) {
+                        Ok(observed) if !observed.stopped && observed.failure.is_none() => {}
+                        _ => {
+                            error = Some(anyhow::anyhow!("WAL replica worker failed"));
+                            break;
+                        }
+                    }
+                }
+                if error.is_some() {
+                    break;
+                }
+            }
+            state.draining.store(true, Ordering::Release);
+            let _ = signal.send(true);
+            // Stop client admission first, but keep owners and peer traffic
+            // alive until every already-running maintenance operation is joined.
+            parts.stop_clients(&mut error).await;
+            retain(
+                &mut error,
+                (&mut maintenance)
+                    .await
+                    .map_err(anyhow::Error::from)
+                    .and_then(|r| r),
+            );
+            retain(
+                &mut error,
+                (&mut health)
+                    .await
+                    .map_err(anyhow::Error::from)
+                    .and_then(|r| r),
+            );
+            retain(&mut error, parts.drain().await);
+            error.map_or(Ok(()), Err)
+        });
+        Ok(Self {
+            stop,
+            supervisor: Some(supervisor),
+            handle,
+        })
+    }
+    pub fn handle(&self) -> runtime::Handle {
+        self.handle.clone()
+    }
+    pub async fn wait(&mut self) -> Result<()> {
+        let result = self
+            .supervisor
+            .as_mut()
+            .context("service already joined")?
+            .await;
+        self.supervisor.take();
+        result?
+    }
+    pub async fn shutdown(mut self) -> Result<()> {
+        let _ = self.stop.send(true);
+        if self.supervisor.is_some() {
+            self.wait().await
+        } else {
+            Ok(())
+        }
+    }
+}
+impl Drop for Service {
+    fn drop(&mut self) {
+        let _ = self.stop.send(true);
+    }
+}
diff --git a/crates/lance-context-wal/src/service/config.rs b/crates/lance-context-wal/src/service/config.rs
new file mode 100644
index 00000000..8b221495
--- /dev/null
+++ b/crates/lance-context-wal/src/service/config.rs
@@ -0,0 +1,299 @@
+use super::*;
+use crate::{
+    placement::Layout,
+    raft_log::{DiskOptions, LogOptions},
+    replication::{source, Limits},
+};
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    io::Read,
+    net::SocketAddr,
+    path::{Path, PathBuf},
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Endpoint {
+    pub peer: String,
+    pub snapshot: String,
+    pub data: String,
+    pub control: String,
+    pub server_name: String,
+    pub certificate: [u8; 32],
+}
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Listen {
+    pub peer: SocketAddr,
+    pub snapshot: SocketAddr,
+    pub data: SocketAddr,
+    pub control: SocketAddr,
+    /// Local diagnostic endpoint, not an authorization or quorum oracle.
+    pub health: SocketAddr,
+}
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TlsFiles {
+    pub certificate: PathBuf,
+    pub private_key: PathBuf,
+    pub roots: PathBuf,
+}
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SigningFiles {
+    pub catalog: Option<PathBuf>,
+    pub sources: BTreeMap<u64, PathBuf>,
+}
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Config {
+    pub node: u64,
+    pub directory: PathBuf,
+    pub layout: Layout,
+    pub limits: Limits,
+    pub endpoints: BTreeMap<u64, Endpoint>,
+    pub listen: Listen,
+    pub tls: TlsFiles,
+    pub trusted_clients: BTreeSet<[u8; 32]>,
+    pub signing: SigningFiles,
+    pub disk: DiskOptions,
+    #[serde(default)]
+    pub storage: LogOptions,
+    #[serde(default)]
+    pub runtime: runtime::Options,
+    #[serde(default)]
+    pub peer: peer::Options,
+    #[serde(default)]
+    pub snapshot: snapshot::Options,
+    #[serde(default)]
+    pub client: client::Options,
+    #[serde(default)]
+    pub maintenance: maintenance::Options,
+}
+impl Config {
+    pub fn read(path: &Path) -> Result<Self> {
+        let mut config: Self = serde_json::from_slice(&read_bounded(path, 1 << 20)?)?;
+        let base = path.parent().unwrap_or_else(|| Path::new("."));
+        fn resolve(base: &Path, path: &mut PathBuf) {
+            if path.is_relative() {
+                *path = base.join(&*path);
+            }
+        }
+        resolve(base, &mut config.directory);
+        for path in [
+            &mut config.tls.certificate,
+            &mut config.tls.private_key,
+            &mut config.tls.roots,
+        ] {
+            resolve(base, path);
+        }
+        if let Some(path) = &mut config.signing.catalog {
+            resolve(base, path);
+        }
+        for path in config.signing.sources.values_mut() {
+            resolve(base, path);
+        }
+        config.validate()?;
+        Ok(config)
+    }
+    pub fn groups(&self) -> Vec<u64> {
+        self.layout
+            .groups
+            .iter()
+            .filter(|(_, voters)| voters.contains(&self.node))
+            .map(|(&group, _)| group)
+            .collect()
+    }
+    pub fn validate(&self) -> Result<()> {
+        self.layout.validate()?;
+        self.limits.identity()?;
+        self.disk.validate()?;
+        self.runtime.validate()?;
+        self.peer.validate()?;
+        self.client
+            .validate(self.limits.max_record_bytes as usize)?;
+        self.maintenance.validate()?;
+        ensure!(
+            !self.directory.as_os_str().is_empty() && self.layout.nodes.contains_key(&self.node),
+            "invalid node or directory"
+        );
+        ensure!(
+            self.endpoints.keys().eq(self.layout.nodes.keys()),
+            "endpoints must cover exactly the layout nodes"
+        );
+        let mut certificates = BTreeSet::new();
+        let mut addresses = BTreeSet::new();
+        for (&node, endpoint) in &self.endpoints {
+            ensure!(
+                endpoint.peer == self.layout.nodes[&node],
+                "advertised peer address differs from durable layout"
+            );
+            for value in [
+                &endpoint.peer,
+                &endpoint.snapshot,
+                &endpoint.data,
+                &endpoint.control,
+            ] {
+                ensure!(
+                    !value.is_empty() && value.len() <= 1024 && addresses.insert(value),
+                    "invalid or duplicate advertised address"
+                );
+            }
+            tokio_rustls::rustls::pki_types::ServerName::try_from(endpoint.server_name.clone())?;
+            ensure!(
+                endpoint.certificate != [0; 32] && certificates.insert(endpoint.certificate),
+                "invalid or duplicate node certificate"
+            );
+        }
+        ensure!(
+            !self.trusted_clients.is_empty()
+                && self.trusted_clients.len() <= 256
+                && !self.trusted_clients.contains(&[0; 32]),
+            "invalid application certificate pins"
+        );
+        let mut listens = BTreeSet::new();
+        for address in [
+            self.listen.peer,
+            self.listen.snapshot,
+            self.listen.data,
+            self.listen.control,
+            self.listen.health,
+        ] {
+            ensure!(
+                address.port() > 0 && listens.insert(address),
+                "listener addresses must be distinct fixed ports"
+            );
+        }
+        ensure!(
+            self.storage.target_file_bytes > 0
+                && self.storage.purge_threshold_bytes >= self.storage.target_file_bytes
+                && usize::try_from(self.storage.purge_threshold_bytes).is_ok(),
+            "invalid physical storage thresholds"
+        );
+        let largest = (self.limits.max_record_bytes as usize).max(1 << 20) + 4096;
+        ensure!(
+            self.peer.peer_bytes >= largest
+                && self.peer.peer_bytes + largest <= self.runtime.peer_bytes,
+            "peer credits must fit a record and reserve capacity for another voter"
+        );
+        ensure!(
+            (1..=16).contains(&self.snapshot.outbound_transfers)
+                && self.snapshot.outbound_transfers < self.runtime.peer_requests
+                && (1..=64).contains(&self.snapshot.connections)
+                && (4096..=1 << 40).contains(&self.snapshot.max_image_bytes)
+                && (Duration::from_millis(100)..=Duration::from_secs(3600))
+                    .contains(&self.snapshot.transfer_timeout),
+            "invalid snapshot options"
+        );
+        ensure!(
+            self.limits.producer_slots > 0
+                && self.limits.catalog_key.is_some()
+                && self.limits.source_keys.keys().eq(self.layout.groups.keys()),
+            "service requires bounded producers and complete publication signing policy"
+        );
+        let catalog = *self
+            .layout
+            .groups
+            .keys()
+            .next()
+            .context("no catalog group")?;
+        let hosted = self.groups();
+        ensure!(
+            self.signing.catalog.is_some() == hosted.contains(&catalog),
+            "catalog secret must be provisioned exactly on catalog replicas"
+        );
+        ensure!(
+            self.signing.sources.keys().copied().collect::<Vec<_>>() == hosted,
+            "source secrets must cover exactly this node's hosted groups"
+        );
+        Ok(())
+    }
+    pub(super) fn peers(&self, images: bool) -> BTreeMap<u64, peer::Peer> {
+        let ids: BTreeSet<_> = self
+            .groups()
+            .iter()
+            .flat_map(|group| self.layout.groups[group].iter().copied())
+            .filter(|id| *id != self.node)
+            .collect();
+        ids.into_iter()
+            .map(|id| {
+                let endpoint = &self.endpoints[&id];
+                (
+                    id,
+                    peer::Peer {
+                        address: if images {
+                            endpoint.snapshot.clone()
+                        } else {
+                            endpoint.peer.clone()
+                        },
+                        server_name: endpoint.server_name.clone(),
+                        certificate: endpoint.certificate,
+                    },
+                )
+            })
+            .collect()
+    }
+    pub(super) fn open(&self) -> Result<(crate::replication::Node, tls::Identity)> {
+        self.validate()?;
+        let identity = tls::Identity::load(&self.tls, &self.endpoints[&self.node].server_name)?;
+        ensure!(
+            identity.fingerprint == self.endpoints[&self.node].certificate,
+            "local TLS identity differs from configured node pin"
+        );
+        let catalog = self
+            .signing
+            .catalog
+            .as_ref()
+            .map(|path| source::CatalogSigner::from_pkcs8(&read_bounded(path, 16 << 10)?))
+            .transpose()?;
+        let sources = self
+            .signing
+            .sources
+            .iter()
+            .map(|(&group, path)| {
+                Ok((
+                    group,
+                    source::SourceSigner::from_pkcs8(&read_bounded(path, 16 << 10)?)?,
+                ))
+            })
+            .collect::<Result<BTreeMap<_, _>>>()?;
+        ensure!(
+            catalog
+                .as_ref()
+                .is_none_or(|key| Some(key.public_key()) == self.limits.catalog_key)
+                && sources
+                    .iter()
+                    .all(|(group, key)| self.limits.source_keys.get(group)
+                        == Some(&key.public_key())),
+            "private signing keys differ from immutable public policy"
+        );
+        let mut node = crate::replication::Node::open_with_storage_options(
+            &self.directory,
+            Arc::new(self.layout.clone()),
+            self.node,
+            self.limits.clone(),
+            self.storage,
+        )?
+        .with_disk_budget(self.disk)?;
+        if let Some(signer) = catalog {
+            node.set_catalog_signer(Arc::new(signer))?;
+        }
+        for (group, signer) in sources {
+            node.set_source_signer(group, Arc::new(signer))?;
+        }
+        Ok((node, identity))
+    }
+}
+pub(super) fn read_bounded(path: &Path, limit: usize) -> Result<Vec<u8>> {
+    let mut bytes = Vec::new();
+    std::fs::File::open(path)
+        .with_context(|| format!("open {}", path.display()))?
+        .take((limit + 1) as u64)
+        .read_to_end(&mut bytes)?;
+    ensure!(
+        bytes.len() <= limit,
+        "configuration file exceeds size limit"
+    );
+    Ok(bytes)
+}
diff --git a/crates/lance-context-wal/src/service/health.rs b/crates/lance-context-wal/src/service/health.rs
new file mode 100644
index 00000000..28678531
--- /dev/null
+++ b/crates/lance-context-wal/src/service/health.rs
@@ -0,0 +1,139 @@
+use super::*;
+use bytes::Bytes;
+use http_body_util::Full;
+use hyper::{body::Incoming, service::service_fn, Request, Response, StatusCode};
+use serde::Serialize;
+use std::{
+    collections::BTreeMap,
+    convert::Infallible,
+    sync::atomic::{AtomicBool, Ordering},
+};
+
+pub(super) struct State {
+    pub node: u64,
+    pub handle: runtime::Handle,
+    pub groups: Vec<u64>,
+    pub draining: AtomicBool,
+    pub maintenance: maintenance::Shared,
+    pub peers: Arc<peer::Statistics>,
+    pub snapshots: Arc<snapshot::Statistics>,
+}
+#[derive(Serialize)]
+struct Group {
+    status: crate::replication::Status,
+    checkpoint_index: u64,
+    observed_age_ms: u64,
+    polling: bool,
+    stopped: bool,
+}
+#[derive(Serialize)]
+struct View {
+    node: u64,
+    ready: bool,
+    groups: BTreeMap<u64, Group>,
+    disk: Option<crate::admission::Snapshot>,
+    maintenance: maintenance::Progress,
+    snapshots_sent: u64,
+    snapshots_installed: u64,
+    snapshots_failed: u64,
+    peer_drops: (u64, u64),
+    peer_sent_frames: u64,
+    peer_received_frames: u64,
+}
+impl State {
+    fn view(&self) -> Result<View> {
+        let mut ready = !self.draining.load(Ordering::Acquire);
+        let mut groups = BTreeMap::new();
+        for &group in &self.groups {
+            let o = self.handle.observation(group)?;
+            let age = o.observed_at.elapsed();
+            ready &= !o.stopped && o.failure.is_none() && age <= Duration::from_secs(15);
+            groups.insert(
+                group,
+                Group {
+                    status: o.status,
+                    checkpoint_index: o.checkpoint_index,
+                    observed_age_ms: age.as_millis().min(u64::MAX as u128) as u64,
+                    polling: o.polling,
+                    stopped: o.stopped,
+                },
+            );
+        }
+        Ok(View {
+            node: self.node,
+            ready,
+            groups,
+            disk: self.handle.disk_budget(),
+            maintenance: self
+                .maintenance
+                .lock()
+                .map_err(|_| anyhow::anyhow!("maintenance observation unavailable"))?
+                .clone(),
+            snapshots_sent: self.snapshots.sent.load(Ordering::Relaxed),
+            snapshots_installed: self.snapshots.installed.load(Ordering::Relaxed),
+            snapshots_failed: self.snapshots.failed.load(Ordering::Relaxed),
+            peer_drops: self.handle.peer_drops(),
+            peer_sent_frames: self.peers.sent_frames.load(Ordering::Relaxed),
+            peer_received_frames: self.peers.received_frames.load(Ordering::Relaxed),
+        })
+    }
+}
+async fn respond(
+    state: Arc<State>,
+    request: Request<Incoming>,
+) -> Result<Response<Full<Bytes>>, Infallible> {
+    let (code, body) = if request.method() != hyper::Method::GET {
+        (StatusCode::METHOD_NOT_ALLOWED, Bytes::new())
+    } else {
+        match request.uri().path() {
+            "/health/live" => (StatusCode::OK, Bytes::from_static(b"{\"live\":true}\n")),
+            "/health/ready" | "/status" => match state.view() {
+                Ok(view) => (
+                    if view.ready {
+                        StatusCode::OK
+                    } else {
+                        StatusCode::SERVICE_UNAVAILABLE
+                    },
+                    Bytes::from(serde_json::to_vec(&view).unwrap_or_default()),
+                ),
+                Err(_) => (StatusCode::SERVICE_UNAVAILABLE, Bytes::new()),
+            },
+            _ => (StatusCode::NOT_FOUND, Bytes::new()),
+        }
+    };
+    Ok(Response::builder()
+        .status(code)
+        .header("content-type", "application/json")
+        .body(Full::new(body))
+        .expect("fixed HTTP response"))
+}
+/// Bound even slow header readers independently of the WAL's data/control pools.
+pub(super) async fn serve(
+    listener: TcpListener,
+    state: Arc<State>,
+    mut stop: watch::Receiver<bool>,
+) -> Result<()> {
+    let slots = Arc::new(tokio::sync::Semaphore::new(16));
+    let mut tasks = tokio::task::JoinSet::new();
+    loop {
+        tokio::select! {
+            _ = stop.changed() => break,
+            _ = tasks.join_next(), if !tasks.is_empty() => {},
+            incoming = listener.accept() => {
+                let (tcp,_)=incoming?;
+                let Ok(permit)=slots.clone().try_acquire_owned() else { drop(tcp); continue; };
+                let state=state.clone();
+                tasks.spawn(async move {
+                    let _permit=permit;
+                    let service=service_fn(move |request| respond(state.clone(),request));
+                    let mut http=hyper::server::conn::http1::Builder::new();
+                    http.keep_alive(false).max_buf_size(8192);
+                    let _=tokio::time::timeout(Duration::from_secs(2),http.serve_connection(hyper_util::rt::TokioIo::new(tcp),service)).await;
+                });
+            }
+        }
+    }
+    tasks.abort_all();
+    while tasks.join_next().await.is_some() {}
+    Ok(())
+}
diff --git a/crates/lance-context-wal/src/service/maintenance.rs b/crates/lance-context-wal/src/service/maintenance.rs
new file mode 100644
index 00000000..4d341ff3
--- /dev/null
+++ b/crates/lance-context-wal/src/service/maintenance.rs
@@ -0,0 +1,192 @@
+use super::*;
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    sync::Mutex,
+    time::Instant,
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct Options {
+    pub interval_ms: u64,
+    pub checkpoint_entries: u64,
+    pub checkpoint_interval_ms: u64,
+    pub log_interval_ms: u64,
+    pub scan_entries: usize,
+}
+impl Default for Options {
+    fn default() -> Self {
+        Self {
+            interval_ms: 1000,
+            checkpoint_entries: 8192,
+            checkpoint_interval_ms: 60_000,
+            log_interval_ms: 30_000,
+            scan_entries: 512,
+        }
+    }
+}
+impl Options {
+    pub(super) fn validate(&self) -> Result<()> {
+        ensure!(
+            (10..=60_000).contains(&self.interval_ms)
+                && self.checkpoint_entries > 0
+                && (self.interval_ms..=3_600_000).contains(&self.checkpoint_interval_ms)
+                && (self.interval_ms..=3_600_000).contains(&self.log_interval_ms)
+                && (1..=4096).contains(&self.scan_entries),
+            "invalid maintenance scheduling"
+        );
+        Ok(())
+    }
+}
+#[derive(Clone, Debug, Default, Serialize)]
+pub struct Progress {
+    pub checkpoints: u64,
+    pub checkpoint_micros: u64,
+    pub collected_files: u64,
+    pub collected_bytes: u64,
+    pub log_passes: u64,
+    pub log_micros: u64,
+    pub attempts_failed: u64,
+    pub last_error: Option<String>,
+}
+pub(super) type Shared = Arc<Mutex<Progress>>;
+fn error(progress: &Shared, error: anyhow::Error) {
+    let mut progress = progress.lock().unwrap();
+    progress.attempts_failed = progress.attempts_failed.saturating_add(1);
+    progress.last_error = Some(error.to_string().chars().take(2048).collect());
+}
+fn micros(start: Instant) -> u64 {
+    start.elapsed().as_micros().min(u64::MAX as u128) as u64
+}
+
+/// One scheduler per physical node, with at most one blocking job outstanding.
+/// Cancellation stops between file operations; every submitted job is joined.
+pub(super) async fn run(
+    handle: runtime::Handle,
+    groups: Vec<u64>,
+    options: Options,
+    progress: Shared,
+    mut stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<()> {
+    let mut checkpoints: BTreeMap<_, _> = groups.iter().map(|&g| (g, Instant::now())).collect();
+    let mut last_log = Instant::now();
+    let mut suggested = BTreeSet::new();
+    let mut tick = tokio::time::interval(Duration::from_millis(options.interval_ms));
+    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        tokio::select! { _ = stop.changed() => return Ok(()), _ = tick.tick() => {} }
+        if *stop.borrow() {
+            return Ok(());
+        }
+        // Preserve the scan cursor across bounded batches. Do not repeatedly
+        // restart at the first directory entry and starve later orphan files.
+        match handle.prepare_image_collection() {
+            Ok(pending) => match pending.await {
+                Ok(mut collection) => loop {
+                    let count = options.scan_entries;
+                    let (returned, result) = tokio::task::spawn_blocking(move || {
+                        let result = collection.run_batch(count);
+                        (collection, result)
+                    })
+                    .await?;
+                    collection = returned;
+                    match result {
+                        Ok(batch) => {
+                            let mut p = progress.lock().unwrap();
+                            p.collected_files =
+                                p.collected_files.saturating_add(batch.removed_files);
+                            p.collected_bytes =
+                                p.collected_bytes.saturating_add(batch.removed_bytes);
+                            if batch.finished || *stop.borrow() {
+                                break;
+                            }
+                        }
+                        Err(e) => {
+                            error(&progress, e);
+                            break;
+                        }
+                    }
+                    tokio::task::yield_now().await;
+                },
+                Err(e) => error(&progress, e),
+            },
+            Err(e) => error(&progress, e),
+        }
+        if *stop.borrow() {
+            return Ok(());
+        }
+        for &group in &groups {
+            let observed = handle.observation(group)?;
+            ensure!(
+                !observed.stopped && observed.failure.is_none(),
+                "replica worker failed during maintenance"
+            );
+            let new = observed
+                .status
+                .applied
+                .saturating_sub(observed.checkpoint_index);
+            if new == 0
+                || (!suggested.contains(&group)
+                    && new < options.checkpoint_entries
+                    && checkpoints[&group].elapsed()
+                        < Duration::from_millis(options.checkpoint_interval_ms))
+            {
+                continue;
+            }
+            let start = Instant::now();
+            let result: Result<()> = async {
+                let capture = handle
+                    .prepare_checkpoint(group, uuid::Uuid::new_v4())?
+                    .await?;
+                let built = tokio::task::spawn_blocking(move || capture.run()).await??;
+                handle.install_checkpoint(built)?.await?;
+                Ok(())
+            }
+            .await;
+            checkpoints.insert(group, Instant::now());
+            match result {
+                Ok(()) => {
+                    suggested.remove(&group);
+                    let mut p = progress.lock().unwrap();
+                    p.checkpoints = p.checkpoints.saturating_add(1);
+                    p.checkpoint_micros = p.checkpoint_micros.saturating_add(micros(start));
+                }
+                Err(e) => error(&progress, e),
+            }
+            if *stop.borrow() {
+                return Ok(());
+            }
+        }
+        if last_log.elapsed() >= Duration::from_millis(options.log_interval_ms) {
+            let start = Instant::now();
+            let result: Result<Vec<u64>> = async {
+                let task = handle.prepare_log_maintenance()?.await?;
+                tokio::task::spawn_blocking(move || task.run()).await?
+            }
+            .await;
+            last_log = Instant::now();
+            match result {
+                Ok(next) => {
+                    suggested.extend(next.into_iter().filter(|group| groups.contains(group)));
+                    let mut p = progress.lock().unwrap();
+                    p.log_passes = p.log_passes.saturating_add(1);
+                    p.log_micros = p.log_micros.saturating_add(micros(start));
+                }
+                Err(e) => error(&progress, e),
+            }
+        }
+        if *stop.borrow() {
+            return Ok(());
+        }
+        let result: Result<()> = async {
+            let task = handle.prepare_disk_measurement()?.await?;
+            tokio::task::spawn_blocking(move || task.run()).await??;
+            Ok(())
+        }
+        .await;
+        if let Err(e) = result {
+            error(&progress, e);
+        }
+    }
+}
diff --git a/crates/lance-context-wal/src/service/tls.rs b/crates/lance-context-wal/src/service/tls.rs
new file mode 100644
index 00000000..47900103
--- /dev/null
+++ b/crates/lance-context-wal/src/service/tls.rs
@@ -0,0 +1,78 @@
+use super::*;
+use rustls::client::danger::ServerCertVerifier;
+use tokio_rustls::rustls::{
+    self,
+    pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer, ServerName, UnixTime},
+    RootCertStore,
+};
+
+pub(super) struct Identity {
+    pub fingerprint: [u8; 32],
+    client: rustls::ClientConfig,
+    server: rustls::ServerConfig,
+}
+impl Identity {
+    pub fn load(files: &config::TlsFiles, name: &str) -> Result<Self> {
+        let certificates =
+            CertificateDer::pem_slice_iter(&config::read_bounded(&files.certificate, 1 << 20)?)
+                .collect::<Result<Vec<_>, _>>()?;
+        ensure!(
+            !certificates.is_empty() && certificates.len() <= 16,
+            "invalid certificate chain length"
+        );
+        let key =
+            PrivateKeyDer::from_pem_slice(&config::read_bounded(&files.private_key, 64 << 10)?)?;
+        let roots = CertificateDer::pem_slice_iter(&config::read_bounded(&files.roots, 1 << 20)?)
+            .collect::<Result<Vec<_>, _>>()?;
+        ensure!(
+            !roots.is_empty() && roots.len() <= 128,
+            "invalid root certificate count"
+        );
+        let mut store = RootCertStore::empty();
+        for certificate in roots {
+            store.add(certificate)?;
+        }
+        let store = Arc::new(store);
+        let provider = Arc::new(rustls::crypto::ring::default_provider());
+        let verifier = rustls::client::WebPkiServerVerifier::builder_with_provider(
+            store.clone(),
+            provider.clone(),
+        )
+        .build()?;
+        verifier.verify_server_cert(
+            &certificates[0],
+            &certificates[1..],
+            &ServerName::try_from(name.to_owned())?,
+            &[],
+            UnixTime::now(),
+        )?;
+        let fingerprint = ring::digest::digest(&ring::digest::SHA256, certificates[0].as_ref())
+            .as_ref()
+            .try_into()?;
+        let verifier = rustls::server::WebPkiClientVerifier::builder_with_provider(
+            store.clone(),
+            provider.clone(),
+        )
+        .build()?;
+        let server = rustls::ServerConfig::builder_with_provider(provider.clone())
+            .with_safe_default_protocol_versions()?
+            .with_client_cert_verifier(verifier)
+            .with_single_cert(certificates.clone(), key.clone_key())?;
+        let client = rustls::ClientConfig::builder_with_provider(provider)
+            .with_safe_default_protocol_versions()?
+            .with_root_certificates(store)
+            .with_client_auth_cert(certificates, key)?;
+        Ok(Self {
+            fingerprint,
+            client,
+            server,
+        })
+    }
+    pub fn configs(&self, alpn: &[u8]) -> (Arc<rustls::ClientConfig>, Arc<rustls::ServerConfig>) {
+        let mut client = self.client.clone();
+        client.alpn_protocols = vec![alpn.to_vec()];
+        let mut server = self.server.clone();
+        server.alpn_protocols = vec![alpn.to_vec()];
+        (Arc::new(client), Arc::new(server))
+    }
+}
diff --git a/crates/lance-context-wal/tests/node_service.rs b/crates/lance-context-wal/tests/node_service.rs
new file mode 100644
index 00000000..d394ab98
--- /dev/null
+++ b/crates/lance-context-wal/tests/node_service.rs
@@ -0,0 +1,269 @@
+use anyhow::{ensure, Context, Result};
+use bytes::Bytes;
+use lance_context_wal::{
+    record::Batch,
+    replication::{
+        producer,
+        runtime::{
+            client::{self, Client, Reply, Request},
+            Command,
+        },
+        Applied, Outcome,
+    },
+    service::{Config, Service},
+};
+use std::{collections::BTreeMap, sync::Arc, time::Duration};
+use uuid::Uuid;
+#[path = "node_service/fixture.rs"]
+mod fixture;
+use fixture::Fixture;
+
+#[test]
+fn configuration_rejects_missing_signers_unknown_fields_and_listener_aliases() -> Result<()> {
+    let f = Fixture::new()?;
+    let original = &f.configs[&1];
+    let mut bad = original.clone();
+    bad.limits.producer_slots = 0;
+    assert!(bad.validate().is_err());
+    let mut bad = original.clone();
+    bad.signing.sources.clear();
+    assert!(bad.validate().is_err());
+    let mut bad = f.configs[&4].clone();
+    bad.signing.catalog = original.signing.catalog.clone();
+    assert!(bad.validate().is_err());
+    let mut bad = original.clone();
+    bad.listen.snapshot = bad.listen.peer;
+    assert!(bad.validate().is_err());
+    let mut bad = original.clone();
+    bad.endpoints.get_mut(&1).unwrap().peer = "foreign:7000".into();
+    assert!(bad.validate().is_err());
+    let mut json = serde_json::to_value(original)?;
+    json["ignored_typo"] = true.into();
+    assert!(serde_json::from_value::<Config>(json).is_err());
+    for node in 1..=6 {
+        assert_eq!(
+            Config::read(&f.root.path().join(format!("{node}.json")))?.node,
+            node
+        );
+    }
+    Ok(())
+}
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn mismatched_tls_and_signing_identity_fail_before_directory_creation() -> Result<()> {
+    let f = Fixture::new()?;
+    let mut bad = f.configs[&1].clone();
+    bad.tls.private_key = f.configs[&2].tls.private_key.clone();
+    assert!(Service::start(bad).await.is_err());
+    let mut bad = f.configs[&1].clone();
+    bad.signing
+        .sources
+        .insert(10, f.root.path().join("source-20.pk8"));
+    assert!(Service::start(bad).await.is_err());
+    assert!(!f.configs[&1].directory.exists());
+    Ok(())
+}
+async fn accepted(client: &Client, group: u64, batch: &Batch) -> Result<u64> {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            // Explicitly retry transport errors with the same tuple and bytes.
+            if let Ok(response) = client
+                .request_routed(Request::Append {
+                    group,
+                    batch: batch.clone(),
+                })
+                .await
+            {
+                match response.reply {
+                    Reply::Commit(Outcome::Committed(Applied::Accepted { index })) => {
+                        return Ok(index)
+                    }
+                    Reply::Commit(Outcome::Unknown) | Reply::Unknown | Reply::Redirect { .. } => {}
+                    other => anyhow::bail!("unexpected append reply: {other:?}"),
+                }
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await?
+}
+async fn leader(f: &Fixture, group: u64) -> Result<u64> {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            for &node in &f.layout.groups[&group] {
+                if let Ok(status) = f.status(node).await {
+                    let leader = status["groups"][group.to_string()]["status"]["leader"]
+                        .as_u64()
+                        .unwrap_or(0);
+                    if leader != 0 && status["ready"] == true {
+                        return Ok(leader);
+                    }
+                }
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await?
+}
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn actual_node_binaries_maintain_two_disjoint_groups_catch_up_and_restart() -> Result<()> {
+    let f = Fixture::new()?;
+    let result = run_cluster(&f).await;
+    if result.is_err() {
+        let path = f.root.keep();
+        eprintln!("failed service fixture retained at {}", path.display());
+    }
+    result
+}
+async fn run_cluster(f: &Fixture) -> Result<()> {
+    let mut children = (1..=6)
+        .map(|node| Ok((node, f.spawn(node, 1)?)))
+        .collect::<Result<BTreeMap<_, _>>>()?;
+    let first_leader = leader(f, 10).await?;
+    leader(f, 20).await?;
+    for child in children.values_mut() {
+        child.assert_live()?;
+    }
+    let table = Uuid::from_u128(50);
+    let producer = producer::producer_id(0, 1);
+    for (group, bucket) in [(10, 0), (20, 1)] {
+        let command = Command::Producer(producer::Command {
+            cluster: f.layout.cluster,
+            layout: f.layout.fingerprint()?,
+            group,
+            table,
+            slot: 0,
+            action: producer::Action::Open {
+                expected_generation: 0,
+                operation: Uuid::from_u128(group.into()),
+                bucket,
+            },
+        });
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                if let Ok(reply) = f
+                    .client
+                    .request_routed(Request::Commit(Box::new(command.clone())))
+                    .await
+                {
+                    if matches!(
+                        reply.reply,
+                        Reply::Commit(Outcome::Committed(Applied::Producer(
+                            producer::Reply::Opened { .. }
+                        )))
+                    ) {
+                        break;
+                    }
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await?;
+    }
+    let lagging = *(1..=3).find(|n| *n != first_leader).as_ref().unwrap();
+    children.get_mut(&lagging).unwrap().kill()?;
+    let mut expected = BTreeMap::new();
+    let mut tails = BTreeMap::new();
+    for sequence in 1..=16 {
+        for (group, bucket) in [(10, 0), (20, 1)] {
+            let batch = Batch::new(
+                table,
+                bucket,
+                producer,
+                sequence,
+                Bytes::from(vec![(group + sequence) as u8; 256 << 10]),
+            );
+            let index = accepted(&f.client, group, &batch).await?;
+            tails.insert(group, (index, batch.clone()));
+            expected.insert((group, index), batch);
+        }
+    }
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            let status = f.status(first_leader).await?;
+            if status["groups"]["10"]["checkpoint_index"]
+                .as_u64()
+                .unwrap_or(0)
+                >= tails[&10].0
+                && status["maintenance"]["log_passes"].as_u64().unwrap_or(0) > 0
+            {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        Ok::<_, anyhow::Error>(())
+    })
+    .await??;
+    children.insert(lagging, f.spawn(lagging, 2)?);
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            if let Ok(status) = f.status(lagging).await {
+                if status["snapshots_installed"].as_u64().unwrap_or(0) > 0
+                    && status["groups"]["10"]["status"]["accepted_records"] == 16
+                {
+                    break;
+                }
+            }
+            children.get_mut(&lagging).unwrap().assert_live()?;
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        Ok::<_, anyhow::Error>(())
+    })
+    .await??;
+    children.get_mut(&first_leader).unwrap().kill()?;
+    let batch = Batch::new(table, 0, producer, 17, Bytes::from(vec![0xac; 256 << 10]));
+    let index = accepted(&f.client, 10, &batch).await?;
+    tails.insert(10, (index, batch.clone()));
+    expected.insert((10, index), batch);
+    children.insert(first_leader, f.spawn(first_leader, 2)?);
+    // Allow all replicas to catch up, then exercise SIGTERM during automatic
+    // checkpoint/GC activity. Every new process reuses its original directory.
+    for (&(group, index), batch) in &expected {
+        for &node in &f.layout.groups[&group] {
+            tokio::time::timeout(Duration::from_secs(30), async {
+                loop {
+                    if let Ok(reply) = f
+                        .client
+                        .request(node, Request::Fetch { group, index })
+                        .await
+                    {
+                        if let Reply::Record(Some(meta)) = &reply.reply {
+                            assert_eq!(meta, &batch.meta);
+                            assert_eq!(reply.body, batch.body);
+                            break;
+                        }
+                    }
+                    tokio::time::sleep(Duration::from_millis(20)).await;
+                }
+            })
+            .await?;
+        }
+    }
+    for child in children.values_mut() {
+        child.terminate().await?;
+    }
+    for node in 1..=6 {
+        children.insert(node, f.spawn(node, 3)?);
+    }
+    leader(f, 10).await?;
+    leader(f, 20).await?;
+    for (&group, (index, batch)) in &tails {
+        assert_eq!(accepted(&f.client, group, batch).await?, *index);
+    }
+    for (&(group, index), batch) in &expected {
+        for &node in &f.layout.groups[&group] {
+            let reply = f
+                .client
+                .request(node, Request::Fetch { group, index })
+                .await?;
+            ensure!(
+                matches!(&reply.reply,Reply::Record(Some(meta)) if meta==&batch.meta),
+                "record metadata lost across restart"
+            );
+            assert_eq!(reply.body, batch.body);
+        }
+    }
+    for child in children.values_mut() {
+        child.terminate().await?;
+    }
+    Ok(())
+}
diff --git a/crates/lance-context-wal/tests/node_service/fixture.rs b/crates/lance-context-wal/tests/node_service/fixture.rs
new file mode 100644
index 00000000..a1c1e279
--- /dev/null
+++ b/crates/lance-context-wal/tests/node_service/fixture.rs
@@ -0,0 +1,340 @@
+use super::*;
+use lance_context_wal::{
+    placement::Layout,
+    raft_log::{DiskOptions, LogOptions},
+    replication::{
+        runtime::{
+            self,
+            peer::{self, snapshot},
+        },
+        source::{CatalogSigner, SourceSigner},
+        Limits,
+    },
+    service::{
+        config::{Endpoint, Listen, SigningFiles, TlsFiles},
+        maintenance, Config,
+    },
+};
+use ring::{rand::SystemRandom, signature::Ed25519KeyPair};
+use std::{
+    collections::BTreeSet,
+    fs,
+    net::TcpListener,
+    path::PathBuf,
+    process::{Child, Command, Stdio},
+};
+use tokio_rustls::rustls::{
+    self,
+    pki_types::{CertificateDer, PrivatePkcs8KeyDer},
+    RootCertStore,
+};
+
+pub struct Fixture {
+    pub root: tempfile::TempDir,
+    pub configs: BTreeMap<u64, Config>,
+    pub client: Arc<Client>,
+    pub layout: Arc<Layout>,
+}
+impl Fixture {
+    pub fn new() -> Result<Self> {
+        let root = tempfile::tempdir()?;
+        let mut leases = Vec::new();
+        let mut listens = BTreeMap::new();
+        for node in 1..=6 {
+            let mut ports = Vec::new();
+            for _ in 0..5 {
+                let listener = TcpListener::bind("127.0.0.1:0")?;
+                ports.push(listener.local_addr()?);
+                leases.push(listener);
+            }
+            listens.insert(
+                node,
+                Listen {
+                    peer: ports[0],
+                    snapshot: ports[1],
+                    data: ports[2],
+                    control: ports[3],
+                    health: ports[4],
+                },
+            );
+        }
+        let layout = Arc::new(Layout {
+            cluster: Uuid::from_u128(1),
+            nodes: listens
+                .iter()
+                .map(|(&id, l)| (id, l.peer.to_string()))
+                .collect(),
+            groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![4, 5, 6])]),
+            buckets: vec![10, 20, 10, 20],
+        });
+        let mut roots = RootCertStore::empty();
+        let mut roots_pem = String::new();
+        let mut certs = BTreeMap::new();
+        let mut keys = BTreeMap::new();
+        let mut fingerprints = BTreeMap::new();
+        for id in 1..=7 {
+            let certified =
+                rcgen::generate_simple_self_signed(vec![format!("wal-{id}.example.invalid")])?;
+            let der: CertificateDer<'static> = certified.cert.der().clone();
+            roots.add(der.clone())?;
+            roots_pem.push_str(&certified.cert.pem());
+            fs::write(root.path().join(format!("{id}.crt")), certified.cert.pem())?;
+            fs::write(
+                root.path().join(format!("{id}.key")),
+                certified.signing_key.serialize_pem(),
+            )?;
+            fingerprints.insert(
+                id,
+                ring::digest::digest(&ring::digest::SHA256, der.as_ref())
+                    .as_ref()
+                    .try_into()?,
+            );
+            certs.insert(id, der);
+            keys.insert(id, certified.signing_key.serialize_der());
+        }
+        fs::write(root.path().join("roots.pem"), roots_pem)?;
+        let catalog = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new())
+            .map_err(|_| anyhow::anyhow!("generate fixture catalog key"))?;
+        fs::write(root.path().join("catalog.pk8"), catalog.as_ref())?;
+        let mut source_keys = BTreeMap::new();
+        for group in [10, 20] {
+            let key = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new())
+                .map_err(|_| anyhow::anyhow!("generate fixture source key"))?;
+            fs::write(
+                root.path().join(format!("source-{group}.pk8")),
+                key.as_ref(),
+            )?;
+            source_keys.insert(group, SourceSigner::from_pkcs8(key.as_ref())?.public_key());
+        }
+        let limits = Limits {
+            catalog_key: Some(CatalogSigner::from_pkcs8(catalog.as_ref())?.public_key()),
+            source_keys,
+            producer_slots: 16,
+            max_record_bytes: 1 << 20,
+            ..Limits::default()
+        };
+        let endpoints: BTreeMap<_, _> = listens
+            .iter()
+            .map(|(&id, l)| {
+                (
+                    id,
+                    Endpoint {
+                        peer: l.peer.to_string(),
+                        snapshot: l.snapshot.to_string(),
+                        data: l.data.to_string(),
+                        control: l.control.to_string(),
+                        server_name: format!("wal-{id}.example.invalid"),
+                        certificate: fingerprints[&id],
+                    },
+                )
+            })
+            .collect();
+        let mut configs = BTreeMap::new();
+        for (&node, listen) in &listens {
+            let groups: Vec<_> = layout
+                .groups
+                .iter()
+                .filter(|(_, voters)| voters.contains(&node))
+                .map(|(&group, _)| group)
+                .collect();
+            let config = Config {
+                node,
+                directory: root.path().join(format!("node-{node}")),
+                layout: (*layout).clone(),
+                limits: limits.clone(),
+                endpoints: endpoints.clone(),
+                listen: listen.clone(),
+                tls: TlsFiles {
+                    certificate: root.path().join(format!("{node}.crt")),
+                    private_key: root.path().join(format!("{node}.key")),
+                    roots: root.path().join("roots.pem"),
+                },
+                trusted_clients: BTreeSet::from([fingerprints[&7]]),
+                signing: SigningFiles {
+                    catalog: groups
+                        .contains(&10)
+                        .then(|| root.path().join("catalog.pk8")),
+                    sources: groups
+                        .iter()
+                        .map(|&group| (group, root.path().join(format!("source-{group}.pk8"))))
+                        .collect(),
+                },
+                disk: DiskOptions {
+                    max_bytes: 512 << 20,
+                    data_bytes: 128 << 20,
+                    headroom_bytes: 256 << 20,
+                },
+                storage: LogOptions {
+                    target_file_bytes: 1 << 20,
+                    purge_threshold_bytes: 2 << 20,
+                },
+                runtime: runtime::Options::default(),
+                peer: peer::Options::default(),
+                snapshot: snapshot::Options {
+                    max_image_bytes: 64 << 20,
+                    transfer_timeout: Duration::from_secs(30),
+                    ..snapshot::Options::default()
+                },
+                client: client::Options {
+                    timeout: Duration::from_secs(2),
+                    ..client::Options::default()
+                },
+                maintenance: maintenance::Options {
+                    interval_ms: 25,
+                    checkpoint_entries: 4,
+                    checkpoint_interval_ms: 100,
+                    log_interval_ms: 100,
+                    scan_entries: 8,
+                },
+            };
+            config.validate()?;
+            fs::write(
+                root.path().join(format!("{node}.json")),
+                serde_json::to_vec_pretty(&config)?,
+            )?;
+            configs.insert(node, config);
+        }
+        let provider = Arc::new(rustls::crypto::ring::default_provider());
+        let mut tls = rustls::ClientConfig::builder_with_provider(provider)
+            .with_safe_default_protocol_versions()?
+            .with_root_certificates(roots)
+            .with_client_auth_cert(
+                vec![certs[&7].clone()],
+                PrivatePkcs8KeyDer::from(keys[&7].clone()).into(),
+            )?;
+        tls.alpn_protocols = vec![client::ALPN.to_vec()];
+        let endpoints = endpoints
+            .iter()
+            .map(|(&id, e)| {
+                (
+                    id,
+                    client::Endpoint {
+                        data_address: e.data.clone(),
+                        control_address: e.control.clone(),
+                        server_name: e.server_name.clone(),
+                        certificate: e.certificate,
+                    },
+                )
+            })
+            .collect();
+        let client = Arc::new(Client::new(
+            &layout,
+            &limits,
+            endpoints,
+            Arc::new(tls),
+            client::Options {
+                timeout: Duration::from_secs(2),
+                ..client::Options::default()
+            },
+        )?);
+        drop(leases);
+        Ok(Self {
+            root,
+            configs,
+            client,
+            layout,
+        })
+    }
+    pub fn spawn(&self, node: u64, incarnation: u64) -> Result<ChildNode> {
+        let log = self
+            .root
+            .path()
+            .join(format!("node-{node}-{incarnation}.log"));
+        let child = Command::new(env!("CARGO_BIN_EXE_context-wal-node"))
+            .args([
+                "--config",
+                self.root
+                    .path()
+                    .join(format!("{node}.json"))
+                    .to_str()
+                    .unwrap(),
+                "--io-threads",
+                "2",
+            ])
+            .stdin(Stdio::null())
+            .stdout(fs::File::create(&log)?)
+            .stderr(fs::OpenOptions::new().append(true).open(&log)?)
+            .spawn()?;
+        Ok(ChildNode {
+            child: Some(child),
+            log,
+        })
+    }
+    pub async fn status(&self, node: u64) -> Result<serde_json::Value> {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        tokio::time::timeout(Duration::from_secs(3), async {
+            let mut tcp = tokio::net::TcpStream::connect(self.configs[&node].listen.health).await?;
+            tcp.write_all(b"GET /status HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
+                .await?;
+            let mut bytes = Vec::new();
+            tcp.take(64 << 10).read_to_end(&mut bytes).await?;
+            let start = bytes
+                .windows(4)
+                .position(|p| p == b"\r\n\r\n")
+                .context("health response header")?
+                + 4;
+            Ok(serde_json::from_slice(&bytes[start..])?)
+        })
+        .await?
+    }
+}
+pub struct ChildNode {
+    child: Option<Child>,
+    log: PathBuf,
+}
+impl ChildNode {
+    pub fn assert_live(&mut self) -> Result<()> {
+        if let Some(status) = self.child.as_mut().context("child missing")?.try_wait()? {
+            anyhow::bail!(
+                "node exited {status}: {}",
+                fs::read_to_string(&self.log).unwrap_or_default()
+            );
+        }
+        Ok(())
+    }
+    pub fn kill(&mut self) -> Result<()> {
+        if let Some(mut child) = self.child.take() {
+            child.kill()?;
+            child.wait()?;
+        }
+        Ok(())
+    }
+    pub async fn terminate(&mut self) -> Result<()> {
+        let Some(child) = self.child.as_mut() else {
+            return Ok(());
+        };
+        #[cfg(unix)]
+        unsafe {
+            ensure!(
+                libc::kill(child.id() as i32, libc::SIGTERM) == 0,
+                "signal child"
+            );
+        }
+        #[cfg(not(unix))]
+        child.kill()?;
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                if let Some(status) = self.child.as_mut().unwrap().try_wait()? {
+                    ensure!(
+                        status.success(),
+                        "node shutdown failed {status}: {}",
+                        fs::read_to_string(&self.log).unwrap_or_default()
+                    );
+                    self.child.take();
+                    return Ok::<_, anyhow::Error>(());
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await??;
+        Ok(())
+    }
+}
+impl Drop for ChildNode {
+    fn drop(&mut self) {
+        if let Some(mut child) = self.child.take() {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+    }
+}
diff --git a/docs/design/wal-node-service.md b/docs/design/wal-node-service.md
new file mode 100644
index 00000000..5e410830
--- /dev/null
+++ b/docs/design/wal-node-service.md
@@ -0,0 +1,98 @@
+# Independent WAL node service
+
+`context-wal-node` starts the durable runtime, authenticated peer replication,
+separate snapshot transfer, and separate client data/control listeners. It also
+runs automatic checkpoint, image collection, disk measurement and native log
+maintenance. Build the service without Arrow or Lance dependencies:
+
+```sh
+cargo build --release -p lance-context-wal --features service --bin context-wal-node
+context-wal-node --config /etc/wal/node.json --io-threads 2 --blocking-threads 4
+```
+
+The service supplies the storage side of ingestion. HTTP ingestion and Lance
+materialization are separate application components; this executable does not
+publish rows into Lance by itself.
+
+## Configuration and identity
+
+The JSON configuration is `service::Config`; unknown top-level and transport
+option fields are rejected. `tests/node_service/fixture.rs` builds a complete
+configuration for six real processes in two independent three-voter groups.
+
+Required fields are the node ID, data directory, complete immutable `layout` and
+replication `limits`, advertised `endpoints`, local `listen` addresses, TLS file
+paths, trusted application certificate pins, per-group signing file paths and
+node disk budget. `storage`, `runtime`, `peer`, `snapshot`, `client` and
+`maintenance` have bounded defaults and accept explicit overrides. Durations in
+transport/runtime JSON are integer milliseconds; memory/disk fields are bytes.
+All paths relative to a configuration file resolve against its containing
+directory. The executable reads at most 1 MiB of configuration.
+
+Each advertised node has distinct peer, snapshot, data and control addresses, a
+TLS server name and a SHA-256 leaf-certificate pin. Addresses are stable service
+names; changing the durable layout is a migration. Listeners may bind wildcard
+addresses, but must have distinct, nonzero ports. Only voters sharing a hosted
+group have peer queues and certificate authorization on a physical node; adding
+unrelated groups does not create a full mesh of peer tasks.
+
+TLS certificate chains and root bundles use PEM, as does the local private key.
+Catalog/source attestation files contain Ed25519 PKCS#8 DER. Existing persistent
+keys must match the public keys in the immutable replication policy. The local
+TLS certificate is checked against its configured pin, DNS name, roots and
+private key before opening storage. Startup never generates replacement keys,
+producer policy or a new layout. Catalog keys are supplied only to catalog
+replicas; source private keys cover exactly the groups hosted by each node.
+
+Provision files through per-node mounted configuration and secrets. A node
+requires its original persistent directory across process/Pod replacement.
+For Kubernetes, use stable DNS, persistent volumes, and a termination grace
+period sufficient for the largest native maintenance operation. Startup must
+permit all voters to resolve each other before readiness; a headless service
+normally needs `publishNotReadyAddresses: true`. Configure one process per
+physical engine, not one process per virtual bucket.
+
+## Automatic maintenance and shutdown
+
+Snapshot transport is attached before maintenance starts. Each node has one
+maintenance scheduler and at most one of its blocking jobs outstanding.
+Checkpointing captures applied state on the owner thread, builds a complete
+application image outside it, and durably installs that image before logically
+compacting the covered log. Retained acknowledged payloads remain in the image.
+The scheduler uses both an entry threshold and a maximum interval, and acts on
+native-engine checkpoint suggestions without treating them as deletion permits.
+
+Image collection retains a scan cursor across bounded batches and protects
+installed images and live transfers/readers. Physical native log maintenance
+runs once per node, with explicit rewrite headroom. It can pause persistence
+while holding the engine's maintenance gate; it has no hard execution deadline.
+Counters expose its duration separately from checkpoint work. Admission errors
+or a busy image permit are observed and retried; a failed replica/transport task
+causes the service to drain and exit.
+
+SIGTERM/SIGINT stops client admission, joins maintenance and snapshot file work,
+then stops and joins peer/runtime owners. Cancelling a shutdown waiter does not
+abort the owning supervisor or free the engine while filesystem work is still
+running. Unanswered writes remain unknown, and clients must retry their exact
+persisted producer/sequence/body tuple.
+
+Disk admission is an application control window, not a filesystem quota or a
+promise of unlimited ingest. Replication can consume reserved headroom. Without
+publication/source retirement, acknowledged data must accumulate and admission
+must eventually stop; checkpoints cannot discard that data to improve a score.
+
+## Observation and qualification
+
+The diagnostic HTTP listener serves `/health/live`, `/health/ready` and `/status`.
+It allows at most 16 connections, bounds header buffers, disables keepalive and
+uses a two-second connection deadline, independently of WAL RPC capacity.
+Readiness reports active local workers and fresh local observations; it is not
+a quorum lease or proof of Lance visibility. Status includes each group's log
+and checkpoint positions, disk accounting and maintenance/transfer counters.
+
+The process test starts the actual executable on six independent directories,
+stops a follower until automatic checkpointing has compacted its missing log,
+requires a snapshot to catch it up, kills a leader, then stops/restarts every
+process with the same configuration. Exact last-batch retries and complete
+metadata/payload reads are checked on every replica. These are correctness and
+lifecycle checks, not sustained throughput results.