diff --git a/crates/core/src/import.rs b/crates/core/src/import.rs
index 6fabecf4f..1e9fa7004 100644
--- a/crates/core/src/import.rs
+++ b/crates/core/src/import.rs
@@ -155,3 +155,26 @@ where
 
     Ok(WalSeed::Seeded(cursor))
 }
+
+/// The durable result of importing one batch.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ReplayProgress {
+    /// The batch committed normally at this state-store position.
+    Committed { position: ChainPoint },
+    /// The configured stopping epoch fired after its anchoring block committed.
+    Boundary { position: ChainPoint },
+}
+
+impl ReplayProgress {
+    /// The committed state cursor reported by this import.
+    pub fn position(&self) -> &ChainPoint {
+        match self {
+            Self::Committed { position } | Self::Boundary { position } => position,
+        }
+    }
+
+    /// Whether the configured stopping boundary was reached.
+    pub fn is_boundary(&self) -> bool {
+        matches!(self, Self::Boundary { .. })
+    }
+}
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index ac582ca1d..b69fa9d57 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -45,7 +45,7 @@ pub mod wal;
 pub mod work_unit;
 
 pub use bootstrap::BootstrapExt;
-pub use import::{seed_wal_from_state, ImportExt, WalSeed, WalSeedError};
+pub use import::{seed_wal_from_state, ImportExt, ReplayProgress, WalSeed, WalSeedError};
 pub use submit::SubmitExt;
 pub use sync::SyncExt;
 pub use work_unit::{MempoolUpdate, WorkUnit};
diff --git a/crates/snapshot/src/backfill.rs b/crates/snapshot/src/backfill.rs
index dcd12afdf..f43dbcc29 100644
--- a/crates/snapshot/src/backfill.rs
+++ b/crates/snapshot/src/backfill.rs
@@ -56,29 +56,23 @@
 //!
 //! This module is orchestration only: it composes the mithril fetch, the
 //! import lifecycle, and the publish path `snapshot publish --repo` uses, and
-//! changes none of them. Everything a *process* owns stays outside it, on
-//! [`Driver`]'s seams — the tokio runtime the mithril calls are driven on, the
-//! shutdown token the signal handler cancels, the renderers, and the steps a
-//! [`Domain`] does not expose: opening the stores without a domain, building
-//! and tearing down one that stops at a chosen epoch, and publishing a plan.
-//! That split is what keeps the shutdown semantics the binary's, where the
-//! signals arrive.
+//! changes none of them. Process setup and signal handling stay in the host.
+//! The driver owns source acquisition, publication order and cancellation.
+//! A workspace supplies non-advancing profile inspection and an exclusive
+//! replay session. Engine initialization and persistence are not host policy.
 
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use dolos_core::config::{MithrilConfig, RootConfig};
-use dolos_core::{
-    seed_wal_from_state, BlockSlot, Domain, DomainError, Genesis, ImportExt as _, StateStore as _,
-    WalSeedError,
-};
+use dolos_core::{BlockSlot, ChainPoint, Genesis, RawBlock, ReplayProgress};
 use dolos_mithril as mithril;
 use dolos_mithril::mithril_client::feedback::FeedbackReceiver;
 use itertools::Itertools as _;
 use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 
-use crate::{export::Plan, planning, retry};
+use crate::{export::Plan, planning, retry, source::SnapshotSource};
 
 /// Blocks handed to `import_blocks` per batch.
 const IMPORT_CHUNK: usize = 100;
@@ -107,32 +101,23 @@ const INTERRUPTED: &str =
 /// What the daemon refused, or what refused it.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    /// A seam the caller supplied failed: opening the stores, building a
-    /// domain, or publishing. Its rendering is the caller's — the daemon only
+    /// A seam the caller supplied failed: opening a workspace, replaying,
+    /// or publishing. Its rendering is the caller's — the daemon only
     /// says which step it was in.
     #[error("{0}")]
     Caller(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
 
-    #[error("reading the state cursor")]
-    Cursor(#[source] dolos_core::StateError),
-
-    #[error("loading the chain summary")]
-    ChainSummary(#[source] dolos_core::ChainError),
-
     #[error("reading snapshot.state_epochs")]
     RetainedEpochs(#[source] crate::Error),
 
     #[error("planning the publish")]
     Planning(#[source] crate::Error),
 
-    #[error("seeding the WAL from the state cursor")]
-    WalSeed(#[source] WalSeedError),
-
-    #[error("pruning excess history")]
-    Housekeeping(#[source] DomainError),
-
-    #[error("importing an immutable block chunk")]
-    Import(#[source] DomainError),
+    #[error("replay failed ({replay}) and shutting down also failed ({shutdown})")]
+    ReplayAndShutdown {
+        replay: Box<Error>,
+        shutdown: Box<Error>,
+    },
 
     #[error("iterating the local immutable db: {0}")]
     ImmutableDb(String),
@@ -181,16 +166,21 @@ impl Error {
     }
 }
 
-/// The three stores a publish reads, opened without a domain.
-///
-/// [`Driver::publish_pending`] runs *before* anything opens a domain and
-/// cannot use one: the WAL reseed it performs is the very thing that makes
-/// the next domain open legal, and a domain assembled first would refuse with
-/// `InconsistentState` instead.
-pub struct Stores<D: Domain> {
-    pub wal: D::Wal,
-    pub state: D::State,
-    pub archive: D::Archive,
+/// A dataset that can be inspected without advancing its logical state.
+pub trait Workspace {
+    type Session: Session;
+
+    fn snapshot(&self) -> impl SnapshotSource + '_;
+    fn start(self, target_epoch: u64) -> Result<Self::Session, Error>;
+    fn finish(self) -> Result<(), Error>;
+}
+
+/// The operations backfill needs from an exclusive replay engine.
+pub trait Session {
+    fn committed_position(&self) -> Result<Option<ChainPoint>, Error>;
+    fn import_blocks(&mut self, blocks: Vec<RawBlock>) -> Result<ReplayProgress, Error>;
+    fn prune_history(&mut self) -> Result<u64, Error>;
+    fn finish(self) -> Result<(), Error>;
 }
 
 /// Where the replay's own progress goes.
@@ -219,7 +209,7 @@ impl Replay for () {}
 /// A seam rather than a call into [`publisher`](crate::publisher), because
 /// `snapshot publish --repo` renders the same publish and one telling of that
 /// order is what [`publisher::Publisher`](crate::publisher::Publisher) is for.
-pub trait Publish<D: Domain> {
+pub trait Publish {
     /// Say what is about to be published. Once per iteration, outside the
     /// retry, so a transient failure does not repeat the report.
     fn announce(&self, plan: &Plan) -> Result<(), Error> {
@@ -229,7 +219,7 @@ pub trait Publish<D: Domain> {
 
     /// Publish the plan. Retried in place by the daemon, so it must be safe to
     /// simply run again.
-    fn publish(&self, plan: &Plan, archive: &D::Archive, state: &D::State) -> Result<(), Error>;
+    fn publish(&self, plan: &Plan, source: &dyn SnapshotSource) -> Result<(), Error>;
 }
 
 /// How a run ended, for a caller that has something to say about it.
@@ -446,7 +436,7 @@ fn cleanup_consumed(
 /// The seams are `&dyn` rather than generic parameters because there is one
 /// caller and the daemon is not on any hot path: a monomorphized copy per
 /// closure would buy nothing and cost a signature nobody can read.
-pub struct Driver<'a, D: Domain> {
+pub struct Driver<'a, W: Workspace> {
     /// The node's configuration, for the retained-epoch parameters a plan
     /// carries.
     pub config: &'a RootConfig,
@@ -492,28 +482,26 @@ pub struct Driver<'a, D: Domain> {
     /// Where the replay's progress goes.
     pub replay: &'a dyn Replay,
 
-    /// Open the three stores a publish reads, without assembling a domain.
-    pub open_stores: &'a dyn Fn() -> Result<Stores<D>, Error>,
-
-    /// Build a domain whose `stop_epoch` is the given epoch.
-    ///
-    /// The seam a [`Domain`] does not cover: the stop epoch is baked in at
-    /// build time and the daemon rebuilds the domain per boundary, so
-    /// construction is the caller's.
-    pub build_domain: &'a dyn Fn(u64) -> Result<D, Error>,
-
-    /// Drain a domain's background work before its handle drops.
-    ///
-    /// Beside [`Driver::build_domain`] and for the same reason: a domain's
-    /// teardown is not on the [`Domain`] trait either, so the half of its
-    /// lifecycle that flushes is the caller's too.
-    pub shutdown_domain: &'a dyn Fn(&D) -> Result<(), Error>,
+    /// Open a dataset for inspection without advancing it.
+    pub open_workspace: &'a dyn Fn() -> Result<W, Error>,
 
     /// Where the planned sequence goes.
-    pub publish: &'a dyn Publish<D>,
+    pub publish: &'a dyn Publish,
 }
 
-impl<D: Domain> Driver<'_, D> {
+fn finish_extend<T>(replay: Result<T, Error>, shutdown: Result<(), Error>) -> Result<T, Error> {
+    match (replay, shutdown) {
+        (Ok(advance), Ok(())) => Ok(advance),
+        (Err(replay), Ok(())) => Err(replay),
+        (Ok(_), Err(shutdown)) => Err(shutdown),
+        (Err(replay), Err(shutdown)) => Err(Error::ReplayAndShutdown {
+            replay: Box::new(replay),
+            shutdown: Box::new(shutdown),
+        }),
+    }
+}
+
+impl<W: Workspace> Driver<'_, W> {
     /// Where the replay reads from.
     fn immutable_dir(&self) -> PathBuf {
         self.download_dir.join("immutable")
@@ -563,18 +551,8 @@ impl<D: Domain> Driver<'_, D> {
 
     /// Publish the sequence the cursor stands at, and name the next target.
     ///
-    /// Also reseeds the WAL from the state cursor before anything else opens
-    /// the domain: `import_blocks` skips the WAL by design, so a run that
-    /// died mid-import left the state ahead of it, and the next domain open
-    /// would refuse with `InconsistentState`.
-    ///
-    /// These stores are dropped rather than shut down, where [`Self::extend`]
-    /// takes the trouble — and the asymmetry is the write, not an oversight.
-    /// The one write here is the WAL reseed, whose only backend is redb, whose
-    /// `shutdown` is a no-op because a redb commit is already durable and its
-    /// drop cleans up without blocking. Everything the publish touches after
-    /// that it only reads, so fjall has no flush of ours to drain — which is
-    /// the whole reason `extend` shuts its domain down after a bulk import.
+    /// Inspection and publication do not advance the dataset. The workspace
+    /// is finalized on both successful and failed publication attempts.
     ///
     /// Nothing *inside* the publish observes [`Driver::cancel`]: a stele goes
     /// out over minutes of store walking and uploading with no seam to check a
@@ -586,9 +564,13 @@ impl<D: Domain> Driver<'_, D> {
     /// below polls it: a shutdown during a backoff ends the run on the failure
     /// in hand rather than after the remaining patience.
     fn publish_pending(&self) -> Result<Step, Error> {
-        let stores = (self.open_stores)()?;
+        let workspace = (self.open_workspace)()?;
+        let result = self.plan_pending(&workspace.snapshot());
+        finish_extend(result, workspace.finish())
+    }
 
-        let cursor = stores.state.read_cursor().map_err(Error::Cursor)?;
+    fn plan_pending(&self, source: &dyn SnapshotSource) -> Result<Step, Error> {
+        let cursor = source.committed_position().map_err(Error::Planning)?;
 
         let Some(cursor) = cursor else {
             return Ok(Step::Extend {
@@ -597,17 +579,12 @@ impl<D: Domain> Driver<'_, D> {
             });
         };
 
-        // An undefined cursor is deliberately not a refusal here: `plan` below
-        // refuses the same state as an unanchored point, with the sentence
-        // that names the command's own subject.
-        if cursor.is_fully_defined() {
-            seed_wal_from_state(&stores.state, &stores.wal).map_err(Error::WalSeed)?;
-        }
-
-        let summary = dolos_cardano::eras::load_chain_summary_from_state(&stores.state)
-            .map_err(Error::ChainSummary)?;
-
-        let (epoch, _) = summary.slot_epoch(cursor.slot());
+        let epoch = source
+            .epoch()
+            .map_err(Error::Planning)?
+            .ok_or(Error::UnanchoredCursor {
+                slot: cursor.slot(),
+            })?;
 
         // Nothing publishable yet: a sequence-0 stele would be epoch 0's
         // mid-epoch sliver, which no consumer chains from.
@@ -620,12 +597,9 @@ impl<D: Domain> Driver<'_, D> {
 
         let retained = planning::retained_epochs(self.config).map_err(Error::RetainedEpochs)?;
 
-        let plan = crate::export::plan(
-            &stores.state,
-            u64::from(self.genesis.network_magic()),
-            retained,
-        )
-        .map_err(Error::Planning)?;
+        let plan = source
+            .plan(u64::from(self.genesis.network_magic()), retained)
+            .map_err(Error::Planning)?;
 
         // Retried here rather than allowed to end the process, because the
         // process ending is the most expensive recovery this driver has and a
@@ -645,7 +619,7 @@ impl<D: Domain> Driver<'_, D> {
         retry::transient(
             "publishing the pending sequence",
             &|| self.aborted(),
-            || self.publish.publish(&plan, &stores.archive, &stores.state),
+            || self.publish.publish(&plan, source),
         )?;
 
         if self.until_epoch.is_some_and(|until| plan.sequence >= until) {
@@ -660,32 +634,29 @@ impl<D: Domain> Driver<'_, D> {
         })
     }
 
-    /// Replay toward `target`'s boundary inside a domain that stops there.
+    /// Replay toward the requested boundary inside an exclusive session.
     fn extend(
         &self,
         target: u64,
         prune: bool,
         slots_per_immutable_file: u64,
     ) -> Result<Advance, Error> {
-        let domain = (self.build_domain)(target)?;
-
-        let result = self.advance_domain(&domain, prune, slots_per_immutable_file);
+        let workspace = (self.open_workspace)()?;
+        let mut session = workspace.start(target)?;
+        let result = self.advance_session(&mut session, prune, slots_per_immutable_file);
 
         // Shut down even when the replay failed: fjall in particular has
         // background work to flush before the handle drops.
-        let shutdown = (self.shutdown_domain)(&domain);
+        let shutdown = session.finish();
 
-        let advance = result?;
-        shutdown?;
-
-        Ok(advance)
+        finish_extend(result, shutdown)
     }
 
     /// Import what is on disk, fetching windows from mithril whenever the
     /// files run out, until the boundary, the aggregator's tip, or a signal.
-    fn advance_domain(
+    fn advance_session(
         &self,
-        domain: &D,
+        session: &mut W::Session,
         prune: bool,
         slots_per_immutable_file: u64,
     ) -> Result<Advance, Error> {
@@ -694,9 +665,7 @@ impl<D: Domain> Driver<'_, D> {
         // After the publish and before the next epoch goes in, never between
         // a boundary and its publish.
         if prune {
-            let rounds = domain
-                .drain_housekeeping(None)
-                .map_err(Error::Housekeeping)?;
+            let rounds = session.prune_history()?;
 
             info!(rounds, "housekeeping drained");
         }
@@ -708,12 +677,10 @@ impl<D: Domain> Driver<'_, D> {
                 break Advance::Cancelled;
             }
 
-            match self.import_available(domain, &immutable_dir)? {
+            match self.import_available(session, &immutable_dir)? {
                 Import::Boundary => {
-                    let cursor_slot = domain
-                        .state()
-                        .read_cursor()
-                        .map_err(Error::Cursor)?
+                    let cursor_slot = session
+                        .committed_position()?
                         .map(|cursor| cursor.slot())
                         .unwrap_or_default();
 
@@ -748,11 +715,7 @@ impl<D: Domain> Driver<'_, D> {
 
             let highest = mithril::highest_existing_immutable(&immutable_dir);
 
-            let cursor_slot = domain
-                .state()
-                .read_cursor()
-                .map_err(Error::Cursor)?
-                .map(|cursor| cursor.slot());
+            let cursor_slot = session.committed_position()?.map(|cursor| cursor.slot());
 
             let resume = resume_file(highest, cursor_slot, slots_per_immutable_file);
 
@@ -830,17 +793,17 @@ impl<D: Domain> Driver<'_, D> {
             }
         };
 
-        // Whatever ended the replay, the chunks it committed are in the state
-        // and the WAL must agree before the next domain open.
-        seed_wal_from_state(domain.state(), domain.wal()).map_err(Error::WalSeed)?;
-
         self.replay.round_finished();
 
         Ok(outcome)
     }
 
     /// Import everything on disk past the cursor, in chunks.
-    fn import_available(&self, domain: &D, immutable_dir: &Path) -> Result<Import, Error> {
+    fn import_available(
+        &self,
+        session: &mut W::Session,
+        immutable_dir: &Path,
+    ) -> Result<Import, Error> {
         use pallas::network::miniprotocols::Point;
 
         // Before the first download the immutable dir does not exist at all;
@@ -858,7 +821,7 @@ impl<D: Domain> Driver<'_, D> {
             return Ok(Import::Exhausted);
         }
 
-        let cursor = domain.state().read_cursor().map_err(Error::Cursor)?;
+        let cursor = session.committed_position()?;
 
         // A cursor with no hash is `ChainPoint::Slot`, which the pallas
         // conversion refuses — and it refuses with `()`, so an `unwrap` here
@@ -867,8 +830,7 @@ impl<D: Domain> Driver<'_, D> {
         // to the boundary slot alone, and the boundary block's own commit right
         // after it. `export::plan` refuses the same state first, as an
         // unanchored point, so the driver ordinarily fails there rather than
-        // here; this says the same thing the WAL seed says, for the path that
-        // reaches it anyway.
+        // here.
         let point: Point = match cursor {
             None => Point::Origin,
             Some(cursor) => {
@@ -900,10 +862,10 @@ impl<D: Domain> Driver<'_, D> {
 
             let batch: Vec<_> = batch.into_iter().map(Arc::new).collect();
 
-            match domain.import_blocks(batch) {
-                Ok(last) => self.replay.reached(last),
-                Err(DomainError::StopEpochReached) => return Ok(Import::Boundary),
-                Err(e) => return Err(Error::Import(e)),
+            let progress = session.import_blocks(batch)?;
+            self.replay.reached(progress.position().slot());
+            if progress.is_boundary() {
+                return Ok(Import::Boundary);
             }
 
             if self.aborted() {
@@ -1055,6 +1017,21 @@ mod tests {
         assert!(!fetch_advanced(None, None));
     }
 
+    #[test]
+    fn replay_and_shutdown_failures_are_both_preserved() {
+        let Err(error) = finish_extend::<Advance>(Err(Error::Interrupted), Err(Error::EmptyWindow))
+        else {
+            panic!("simultaneous failures unexpectedly succeeded");
+        };
+
+        let Error::ReplayAndShutdown { replay, shutdown } = error else {
+            panic!("simultaneous failures lost their combined error");
+        };
+
+        assert!(matches!(*replay, Error::Interrupted));
+        assert!(matches!(*shutdown, Error::EmptyWindow));
+    }
+
     #[test]
     fn windows_advance_from_the_highest_existing_file() {
         // a fresh dir starts at the beginning, one window deep
diff --git a/crates/snapshot/src/lib.rs b/crates/snapshot/src/lib.rs
index 25003698a..e8c7feb78 100644
--- a/crates/snapshot/src/lib.rs
+++ b/crates/snapshot/src/lib.rs
@@ -73,6 +73,7 @@ pub mod planning;
 pub mod publisher;
 pub mod registry;
 pub mod restore;
+pub mod source;
 
 /// The free-space policy, which is [`stelae_driver`]'s: it is one rule over
 /// paths and byte counts and knows nothing about what fills them. Re-exported
diff --git a/crates/snapshot/src/source.rs b/crates/snapshot/src/source.rs
new file mode 100644
index 000000000..cb3e4da14
--- /dev/null
+++ b/crates/snapshot/src/source.rs
@@ -0,0 +1,74 @@
+//! Read-only access to the Dolos snapshot profile.
+
+use dolos_core::{ArchiveStore, ChainPoint, StateStore};
+use stelae::progress::Observer;
+
+use crate::{
+    export::{self, Plan},
+    publisher::Publisher,
+    registry::{Preview, Published},
+    Error, RetainedEpochs,
+};
+
+/// Profile operations available without exposing mutable storage handles.
+pub trait SnapshotSource {
+    fn committed_position(&self) -> Result<Option<ChainPoint>, Error>;
+    fn epoch(&self) -> Result<Option<u64>, Error>;
+    fn plan(&self, network_magic: u64, retained: RetainedEpochs) -> Result<Plan, Error>;
+    fn preview(&self, publisher: &Publisher, plan: &Plan) -> Result<Preview, Error>;
+    fn publish(
+        &self,
+        publisher: &Publisher,
+        plan: &Plan,
+        observer: &Observer,
+    ) -> Result<Published, Error>;
+}
+
+/// A borrowed profile view. The stores remain private and cannot escape it.
+pub struct StoreSnapshot<'a, A, S> {
+    archive: &'a A,
+    state: &'a S,
+}
+
+impl<'a, A: ArchiveStore, S: StateStore> StoreSnapshot<'a, A, S> {
+    pub fn new(archive: &'a A, state: &'a S) -> Self {
+        Self { archive, state }
+    }
+}
+
+impl<A: ArchiveStore, S: StateStore> SnapshotSource for StoreSnapshot<'_, A, S> {
+    fn committed_position(&self) -> Result<Option<ChainPoint>, Error> {
+        Ok(self.state.read_cursor()?)
+    }
+
+    fn epoch(&self) -> Result<Option<u64>, Error> {
+        let Some(position) = self.committed_position()? else {
+            return Ok(None);
+        };
+        if !position.is_fully_defined() {
+            return Err(Error::UnanchoredPoint(format!(
+                "cursor at slot {} has no block hash",
+                position.slot()
+            )));
+        }
+        let summary = dolos_cardano::eras::load_chain_summary_from_state(self.state)?;
+        Ok(Some(summary.slot_epoch(position.slot()).0))
+    }
+
+    fn plan(&self, network_magic: u64, retained: RetainedEpochs) -> Result<Plan, Error> {
+        export::plan(self.state, network_magic, retained)
+    }
+
+    fn preview(&self, publisher: &Publisher, plan: &Plan) -> Result<Preview, Error> {
+        publisher.preview(plan, self.archive)
+    }
+
+    fn publish(
+        &self,
+        publisher: &Publisher,
+        plan: &Plan,
+        observer: &Observer,
+    ) -> Result<Published, Error> {
+        publisher.publish(plan, self.archive, self.state, observer)
+    }
+}
diff --git a/docs/headless-replay.md b/docs/headless-replay.md
new file mode 100644
index 000000000..0cc5b7494
--- /dev/null
+++ b/docs/headless-replay.md
@@ -0,0 +1,70 @@
+# Embedding Dolos replay
+
+Use `dolos` with default service features disabled when the host only needs
+Cardano replay. The `headless_replay` example runs against a local immutable
+block directory:
+
+```sh
+cargo run --no-default-features --example headless_replay -- \
+  ./dolos.toml ./snapshot/immutable 500
+```
+
+## Contract
+
+- `ReplayWorkspace::open(config, genesis)` opens a dataset without running
+  ledger initialization, replaying pending work, or pruning history.
+- `workspace.snapshot()` lends read-only Dolos profile operations:
+  committed position, epoch, publication planning, preview and publication.
+  No domain or writable storage handle escapes the view.
+- `workspace.start(stop_epoch)` consumes the inspection workspace and explicitly
+  permits initialization and processing. Publish any pending boundary first.
+- `session.import_blocks(blocks)` processes a nonempty batch of trusted immutable
+  blocks. It returns `ReplayProgress::Committed` or `Boundary`, each carrying the
+  committed input position. A stopping boundary includes its anchoring block.
+  Resume input from that position: the rest of a submitted batch may be unprocessed.
+- A boundary is terminal for that session. Further import calls return the same
+  boundary without processing input. To advance, finish and start another session.
+- After a replay execution error, the session rejects further input. Finish and reopen;
+  errors indicating an incomplete ledger transition may require explicit repair.
+  An empty batch is rejected before execution and does not invalidate the session.
+- `session.finish()` consumes the handle, persists completed work and releases
+  resources. It returns `Result<(), BulkReplayError>`, not storage bookkeeping.
+  `session.run(operation)` performs finalization on both ordinary success and
+  error returns, preserving both errors when necessary.
+- `session.prune_history()` is explicit host policy. Neither replay completion
+  nor finalization prunes automatically. Publish required history before pruning.
+
+Sessions are not cloneable and do not implement `Domain`. Import requires
+exclusive mutable access. Borrowing a profile view prevents advancing or
+finishing the session while the view remains in use.
+
+## Publisher ordering
+
+```text
+open workspace
+  → inspect/plan/publish pending snapshot
+  → finish inspection workspace
+open workspace
+  → start replay for next boundary
+  → prune already-published history if policy permits
+  → import trusted blocks to boundary
+  → finish
+repeat
+```
+
+The current backfill driver follows this order through its narrow `Workspace`,
+`Session` and `Publish` interfaces. Source acquisition, retries, signals and OCI
+publication policy remain the host's responsibilities.
+
+## Internal implementation and limits
+
+Dolos still uses its existing `ImportExt` execution path. Checkpoint maintenance
+is private to the engine; an embedding application neither selects recovery
+actions nor interprets their results. Normal node construction continues through
+`DomainBuilder` and retains the existing startup integrity policy.
+
+This is not a new atomic import or general corruption-repair mechanism.
+Interrupted epoch transitions, inconsistent archives and other pre-existing
+storage limitations are not repaired by this facade. Dropping a handle,
+panicking or terminating the process is not equivalent to successful
+`finish()`. Internal error details remain available in diagnostic source chains.
diff --git a/examples/headless_replay.rs b/examples/headless_replay.rs
new file mode 100644
index 000000000..31d3d2f3b
--- /dev/null
+++ b/examples/headless_replay.rs
@@ -0,0 +1,106 @@
+//! Minimal external host for Dolos domain construction and bulk replay.
+//!
+//! Build without Dolos service features:
+//!
+//! ```text
+//! cargo run --no-default-features --example headless_replay -- \
+//!   ./dolos.toml ./snapshot/immutable 500
+//! ```
+//!
+//! The final argument is an optional stopping epoch. Source acquisition,
+//! progress rendering, publication, and housekeeping intentionally stay in
+//! the host application.
+
+use std::error::Error;
+use std::path::Path;
+use std::sync::Arc;
+
+use dolos::core::{Genesis, RawBlock};
+use dolos::engine::{BulkReplaySession, ReplayProgress};
+
+type AnyError = Box<dyn Error + Send + Sync + 'static>;
+
+fn main() -> Result<(), AnyError> {
+    let mut args = std::env::args_os().skip(1);
+    let config_path = args.next().ok_or("missing path to dolos.toml")?;
+    let immutable_path = args.next().ok_or("missing immutable directory")?;
+    let stop_epoch = args
+        .next()
+        .map(|value| value.to_string_lossy().parse::<u64>())
+        .transpose()?;
+
+    let config: dolos::core::config::RootConfig = config::Config::builder()
+        .add_source(config::File::from(Path::new(&config_path)))
+        .build()?
+        .try_deserialize()?;
+
+    let genesis = Arc::new(Genesis::from_file_paths(
+        &config.genesis.byron_path,
+        &config.genesis.shelley_path,
+        &config.genesis.alonzo_path,
+        &config.genesis.conway_path,
+        config.genesis.force_protocol,
+    )?);
+
+    let session = BulkReplaySession::open(&config, genesis, stop_epoch)?;
+
+    session
+        .run(|session| {
+            let cursor = session.committed_position()?;
+            let point = match cursor {
+                Some(cursor) => {
+                    let slot = cursor.slot();
+                    cursor.try_into().map_err(|()| {
+                        format!("state cursor at slot {slot} is not anchored to a block hash")
+                    })?
+                }
+                None => pallas::network::miniprotocols::Point::Origin,
+            };
+
+            let mut blocks = pallas::interop::hardano::storage::immutable::read_blocks_from_point(
+                Path::new(&immutable_path),
+                point.clone(),
+            )?;
+
+            if point != pallas::network::miniprotocols::Point::Origin {
+                blocks.next();
+            }
+
+            let mut batch: Vec<RawBlock> = Vec::with_capacity(100);
+
+            for block in blocks {
+                batch.push(Arc::new(block?));
+
+                if batch.len() == 100 && import_batch(session, &mut batch)? {
+                    return Ok::<_, AnyError>(());
+                }
+            }
+
+            if !batch.is_empty() {
+                import_batch(session, &mut batch)?;
+            }
+
+            Ok(())
+        })
+        .map_err(|error| std::io::Error::other(error.to_string()))?;
+
+    Ok(())
+}
+
+fn import_batch(
+    session: &mut BulkReplaySession,
+    batch: &mut Vec<RawBlock>,
+) -> Result<bool, AnyError> {
+    let progress = session.import_blocks(std::mem::take(batch))?;
+
+    match progress {
+        ReplayProgress::Committed { position } => {
+            eprintln!("committed through {position}");
+            Ok(false)
+        }
+        ReplayProgress::Boundary { position } => {
+            eprintln!("stopping boundary committed at {position}");
+            Ok(true)
+        }
+    }
+}
diff --git a/src/bin/dolos/bootstrap/mithril.rs b/src/bin/dolos/bootstrap/mithril.rs
index 7a9a9fa31..5f85f195b 100644
--- a/src/bin/dolos/bootstrap/mithril.rs
+++ b/src/bin/dolos/bootstrap/mithril.rs
@@ -7,7 +7,6 @@
 //! root library assembles.
 
 use dolos_core::config::RootConfig;
-use dolos_core::ImportExt;
 use dolos_mithril::{fetch_snapshot, Fetch};
 use itertools::Itertools;
 use miette::{Context, IntoDiagnostic};
@@ -15,6 +14,7 @@ use std::{path::Path, sync::Arc};
 use tracing::{info, warn};
 
 use crate::feedback::{Feedback, MithrilFeedback};
+use dolos::engine::BulkReplaySession;
 use dolos::prelude::*;
 
 #[derive(Debug, clap::Args, Clone)]
@@ -66,18 +66,13 @@ impl Default for Args {
     }
 }
 
-fn define_starting_point<S: dolos_core::StateStore>(
+fn define_starting_point(
     args: &Args,
-    state: &S,
+    cursor: Option<ChainPoint>,
 ) -> Result<pallas::network::miniprotocols::Point, miette::Error> {
     if let Some(point) = &args.start_from {
         Ok(point.clone().try_into().unwrap())
     } else {
-        let cursor = state
-            .read_cursor()
-            .into_diagnostic()
-            .context("reading state cursor")?;
-
         let point = cursor
             .map(|c| c.try_into().unwrap())
             .unwrap_or(pallas::network::miniprotocols::Point::Origin);
@@ -88,8 +83,9 @@ fn define_starting_point<S: dolos_core::StateStore>(
 
 /// Inner import function that can return errors.
 /// The outer function ensures shutdown is called regardless of success/failure.
-fn do_import<D: dolos_core::Domain>(
-    domain: &D,
+fn do_import(
+    cursor: Option<ChainPoint>,
+    mut import: impl FnMut(Vec<RawBlock>) -> miette::Result<BlockSlot>,
     args: &Args,
     immutable_path: &Path,
     feedback: &Feedback,
@@ -100,7 +96,7 @@ fn do_import<D: dolos_core::Domain>(
         .context("reading immutable db tip")?
         .ok_or(miette::miette!("immutable db has no tip"))?;
 
-    let cursor = define_starting_point(args, domain.state())?;
+    let cursor = define_starting_point(args, cursor)?;
 
     let mut iter = pallas::interop::hardano::storage::immutable::read_blocks_from_point(
         immutable_path,
@@ -131,9 +127,7 @@ fn do_import<D: dolos_core::Domain>(
         // around throughout the pipeline
         let batch: Vec<_> = batch.into_iter().map(Arc::new).collect();
 
-        let last = domain
-            .import_blocks(batch)
-            .map_err(|e| miette::miette!(e.to_string()))?;
+        let last = import(batch)?;
 
         progress.set_position(last);
     }
@@ -150,17 +144,29 @@ fn import_hardano_into_domain(
     feedback: &Feedback,
     chunk_size: usize,
 ) -> Result<(), miette::Error> {
-    let domain = crate::common::setup_domain(config)?;
-
-    let result = do_import(&domain, args, immutable_path, feedback, chunk_size);
-
-    // Always shutdown the domain before it goes out of scope, regardless of
-    // whether import succeeded or failed.
-    if let Err(e) = domain.shutdown() {
-        tracing::error!("error during domain shutdown: {}", e);
-    }
-
-    result
+    let genesis = Arc::new(crate::common::open_genesis_files(&config.genesis)?);
+    let session = BulkReplaySession::open(config, genesis, None)
+        .map_err(|error| miette::miette!("opening the bulk-replay session: {error}"))?;
+
+    session
+        .run(|session| {
+            let cursor = session.committed_position().into_diagnostic()?;
+            do_import(
+                cursor,
+                |blocks| {
+                    let progress = session.import_blocks(blocks).into_diagnostic()?;
+                    if progress.is_boundary() {
+                        return Err(miette::miette!("{}", DomainError::StopEpochReached));
+                    }
+                    Ok(progress.position().slot())
+                },
+                args,
+                immutable_path,
+                feedback,
+                chunk_size,
+            )
+        })
+        .map_err(|error| miette::miette!("{error}"))
 }
 
 pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Result<()> {
@@ -223,7 +229,7 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res
 #[cfg(test)]
 mod tests {
     use super::*;
-    use dolos_core::{ArchiveStore, Domain};
+    use dolos_core::{ArchiveStore, Domain, ImportExt};
     use dolos_testing::{
         blocks::write_immutable_fixture,
         synthetic::{build_synthetic_blocks, SyntheticBlockConfig},
@@ -231,6 +237,23 @@ mod tests {
     };
     use pallas::ledger::traverse::MultiEraBlock;
 
+    fn import_fixture<D: Domain>(
+        domain: &D,
+        args: &Args,
+        path: &Path,
+        feedback: &Feedback,
+        chunk_size: usize,
+    ) -> miette::Result<()> {
+        do_import(
+            domain.state().read_cursor().into_diagnostic()?,
+            |blocks| domain.import_blocks(blocks).into_diagnostic(),
+            args,
+            path,
+            feedback,
+            chunk_size,
+        )
+    }
+
     #[test]
     fn mithril_import_resumes_through_the_automatic_writer() {
         let (blocks, _, config) = build_synthetic_blocks(SyntheticBlockConfig {
@@ -253,7 +276,7 @@ mod tests {
         };
         let dir = tempfile::tempdir().unwrap();
         write_immutable_fixture(dir.path(), &blocks);
-        do_import(&domain, &args, dir.path(), &Feedback::hidden(), 3).unwrap();
+        import_fixture(&domain, &args, dir.path(), &Feedback::hidden(), 3).unwrap();
         let stats = domain.archive().append_stats();
         assert!(stats.serial_batches >= 1, "{stats:?}");
         assert_eq!(
@@ -275,7 +298,7 @@ mod tests {
                 .collect::<Vec<_>>()
         );
         let before = domain.archive().append_stats();
-        do_import(
+        import_fixture(
             &domain,
             &Args::default(),
             dir.path(),
diff --git a/src/bin/dolos/common.rs b/src/bin/dolos/common.rs
index 27a1500c3..cf8ae3e6d 100644
--- a/src/bin/dolos/common.rs
+++ b/src/bin/dolos/common.rs
@@ -1,5 +1,4 @@
-use dolos_core::config::{ChainConfig, GenesisConfig, LoggingConfig, RootConfig, TelemetryConfig};
-use dolos_core::BootstrapExt;
+use dolos_core::config::{GenesisConfig, LoggingConfig, RootConfig, TelemetryConfig};
 use futures_util::{stream::FuturesUnordered, StreamExt};
 use miette::{Context as _, IntoDiagnostic};
 use opentelemetry::trace::TracerProvider as _;
@@ -13,6 +12,7 @@ use tracing_subscriber::{filter::Targets, prelude::*};
 
 use dolos::adapters::DomainAdapter;
 use dolos::core::Genesis;
+use dolos::engine::{DomainBuildError, DomainBuilder};
 use dolos::prelude::*;
 use dolos::storage;
 
@@ -77,48 +77,31 @@ pub fn setup_domain_with_stop_epoch(
     config: &RootConfig,
     stop_epoch: Option<u64>,
 ) -> miette::Result<DomainAdapter> {
-    let stores = open_data_stores(config).map_err(|e| match e {
-        Error::WalError(WalError::IncompatibleVersion { found, expected }) => miette::miette!(
+    let genesis = Arc::new(open_genesis_files(&config.genesis)?);
+
+    DomainBuilder::new(config, genesis)
+        .stop_epoch(stop_epoch)
+        .build()
+        .map_err(render_domain_build_error)
+}
+
+/// Keep actionable executable diagnostics at the UI boundary while the
+/// library returns typed construction errors.
+fn render_domain_build_error(error: DomainBuildError) -> miette::Report {
+    match error {
+        DomainBuildError::Storage(Error::WalError(WalError::IncompatibleVersion {
+            found,
+            expected,
+        })) => miette::miette!(
             help = format!(
                 "WAL was created by a newer dolos version (v{found}) than this binary supports (v{expected}); upgrade dolos or run `dolos bootstrap --force` to wipe storage and re-bootstrap",
             ),
             "incompatible WAL version: found v{found}, expected v{expected}",
         ),
-        other => miette::miette!("{other}"),
-    })?;
-    let genesis = Arc::new(open_genesis_files(&config.genesis)?);
-    let mempool = stores.mempool.clone();
-    let (tip_broadcast, _) = tokio::sync::broadcast::channel(100);
-    let chain = config.chain.clone();
-
-    let ChainConfig::Cardano(mut chain_config) = chain;
-
-    if stop_epoch.is_some() {
-        chain_config.stop_epoch = stop_epoch;
-    }
-
-    let chain = dolos_cardano::CardanoLogic::initialize::<DomainAdapter>(
-        chain_config,
-        &stores.state,
-        &genesis,
-    )
-    .into_diagnostic()?;
-
-    let domain = DomainAdapter {
-        storage_config: Arc::new(config.storage.clone()),
-        sync_config: Arc::new(config.sync.clone()),
-        genesis,
-        chain: Arc::new(std::sync::RwLock::new(chain)),
-        wal: stores.wal,
-        state: stores.state,
-        archive: stores.archive,
-        mempool,
-        tip_broadcast,
-    };
-
-    // this will make sure the domain is correctly initialized and in a valid state.
-    domain.bootstrap().map_err(|e| match e {
-        dolos_core::DomainError::InconsistentState { ref wal, ref state } => {
+        DomainBuildError::Bootstrap(dolos_core::DomainError::InconsistentState {
+            ref wal,
+            ref state,
+        }) => {
             let msg = match (wal, state) {
                 (Some(w), Some(s)) => format!(
                     "state (slot {}) is ahead of WAL (slot {})",
@@ -137,10 +120,8 @@ pub fn setup_domain_with_stop_epoch(
             };
             miette::miette!(help = help, "{msg}")
         }
-        other => miette::miette!("{other:?}"),
-    })?;
-
-    Ok(domain)
+        other => miette::miette!("{other}"),
+    }
 }
 
 pub fn setup_tracing_error_only() -> miette::Result<()> {
diff --git a/src/bin/dolos/snapshot/backfill.rs b/src/bin/dolos/snapshot/backfill.rs
index 65cc24a16..ee31da4aa 100644
--- a/src/bin/dolos/snapshot/backfill.rs
+++ b/src/bin/dolos/snapshot/backfill.rs
@@ -23,7 +23,8 @@ use miette::{bail, Context as _, IntoDiagnostic as _};
 use tokio_util::sync::CancellationToken;
 
 use crate::feedback::Feedback;
-use dolos::adapters::{ArchiveStoreBackend, DomainAdapter, StateStoreBackend};
+use dolos::engine::ReplayWorkspace;
+use dolos_snapshot::source::SnapshotSource;
 
 /// Where the mithril window lands when the operator names nowhere: beside the
 /// stores, so the bytes stay on the data mount.
@@ -169,26 +170,14 @@ impl RepositoryArm<'_> {
     }
 }
 
-impl backfill::Publish<DomainAdapter> for RepositoryArm<'_> {
+impl backfill::Publish for RepositoryArm<'_> {
     fn announce(&self, plan: &Plan) -> Result<(), backfill::Error> {
         super::report_plan(plan).map_err(backfill::Error::caller)
     }
 
-    fn publish(
-        &self,
-        plan: &Plan,
-        archive: &ArchiveStoreBackend,
-        state: &StateStoreBackend,
-    ) -> Result<(), backfill::Error> {
-        super::publish::to_repository(
-            self.config,
-            &self.settings(),
-            plan,
-            archive,
-            state,
-            self.feedback,
-        )
-        .map_err(backfill::Error::caller)
+    fn publish(&self, plan: &Plan, source: &dyn SnapshotSource) -> Result<(), backfill::Error> {
+        super::publish::to_repository(self.config, &self.settings(), plan, source, self.feedback)
+            .map_err(backfill::Error::caller)
     }
 }
 
@@ -231,7 +220,9 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res
         feedback,
     };
 
-    let driver = backfill::Driver::<DomainAdapter> {
+    let genesis = Arc::new(genesis);
+
+    let driver = backfill::Driver::<ReplayWorkspace> {
         config,
         genesis: &genesis,
         mithril,
@@ -249,26 +240,8 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res
                 >)
         },
         replay: &replay,
-        open_stores: &|| {
-            let stores = crate::common::open_data_stores(config)
-                .into_diagnostic()
-                .context("opening the data stores")
-                .map_err(backfill::Error::caller)?;
-
-            Ok(backfill::Stores::<DomainAdapter> {
-                wal: stores.wal,
-                state: stores.state,
-                archive: stores.archive,
-            })
-        },
-        build_domain: &|target| {
-            crate::common::setup_domain_with_stop_epoch(config, Some(target))
-                .map_err(backfill::Error::caller)
-        },
-        shutdown_domain: &|domain: &DomainAdapter| {
-            domain
-                .shutdown()
-                .map_err(|e| backfill::Error::caller(format!("shutting down the domain: {e}")))
+        open_workspace: &|| {
+            ReplayWorkspace::open(config, genesis.clone()).map_err(backfill::Error::caller)
         },
         publish: &publish,
     };
diff --git a/src/bin/dolos/snapshot/publish.rs b/src/bin/dolos/snapshot/publish.rs
index 65b6fdb4a..10d251cfd 100644
--- a/src/bin/dolos/snapshot/publish.rs
+++ b/src/bin/dolos/snapshot/publish.rs
@@ -2,7 +2,7 @@ use std::path::PathBuf;
 
 use clap::Parser;
 use dolos_core::config::RootConfig;
-use dolos_core::{ArchiveStore, StateStore};
+use dolos_snapshot::source::{SnapshotSource, StoreSnapshot};
 use miette::{Context as _, IntoDiagnostic as _};
 
 use dolos_snapshot::{
@@ -132,8 +132,7 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res
                 config,
                 &publish,
                 &plan,
-                &stores.archive,
-                &stores.state,
+                &StoreSnapshot::new(&stores.archive, &stores.state),
                 feedback,
             )
         }
@@ -194,12 +193,11 @@ fn to_directory(
 /// than trust: how much of this stele was inherited rather than built, and how
 /// much of it moved. Both are numbers the code counted, not an inference from a
 /// duration.
-pub(super) fn to_repository<A: ArchiveStore, S: StateStore>(
+pub(super) fn to_repository(
     config: &RootConfig,
     publish: &RepositoryPublish,
     plan: &export::Plan,
-    archive: &A,
-    state: &S,
+    source: &dyn SnapshotSource,
     feedback: &Feedback,
 ) -> miette::Result<()> {
     let repo = publish.repo;
@@ -230,8 +228,8 @@ pub(super) fn to_repository<A: ArchiveStore, S: StateStore>(
         .context("sizing the staging directory")?;
 
     if publish.dry_run {
-        let preview = publisher
-            .preview(plan, archive)
+        let preview = source
+            .preview(&publisher, plan)
             .into_diagnostic()
             .context("planning the publish")?;
 
@@ -253,8 +251,8 @@ pub(super) fn to_repository<A: ArchiveStore, S: StateStore>(
     // under the report.
     let progress = SteleProgress::publishing(feedback);
 
-    let published = publisher
-        .publish(plan, archive, state, &progress.observer())
+    let published = source
+        .publish(&publisher, plan, &progress.observer())
         .into_diagnostic()
         .context("publishing the stele")?;
 
diff --git a/src/engine.rs b/src/engine.rs
new file mode 100644
index 000000000..3e399e5fd
--- /dev/null
+++ b/src/engine.rs
@@ -0,0 +1,500 @@
+//! Headless node construction and exclusive bulk replay.
+//!
+//! Hosts provide resolved configuration, trusted blocks and stopping policy.
+//! Opening a replay workspace does not advance the ledger. Inspect or export
+//! its snapshot before explicitly starting replay. Finalization persists the
+//! committed position and releases resources without pruning history.
+
+mod checkpoint;
+
+use std::fmt;
+use std::sync::Arc;
+
+use dolos_core::config::{ChainConfig, RootConfig};
+pub use dolos_core::ReplayProgress;
+use dolos_core::{
+    BootstrapExt as _, ChainLogic as _, ChainPoint, Domain as _, DomainError, Genesis,
+    ImportExt as _, RawBlock, StateStore as _,
+};
+use dolos_snapshot::source::{SnapshotSource, StoreSnapshot};
+
+use crate::adapters::{ArchiveStoreBackend, DomainAdapter, StateStoreBackend, WalAdapter};
+use crate::storage;
+
+type Stores = storage::Stores<dolos_cardano::CardanoDelta>;
+type Cause = Box<dyn std::error::Error + Send + Sync + 'static>;
+
+/// A failure while assembling and bootstrapping a Dolos domain.
+#[derive(Debug, thiserror::Error)]
+pub enum DomainBuildError {
+    #[error("opening configured Dolos stores: {0}")]
+    Storage(#[source] crate::prelude::Error),
+
+    #[error("initializing Cardano ledger logic: {0}")]
+    Chain(#[source] dolos_core::ChainError),
+
+    #[error("bootstrapping the Dolos domain: {0}")]
+    Bootstrap(#[source] DomainError),
+}
+
+/// Assemble a [`DomainAdapter`] from explicit configuration and genesis.
+///
+/// The builder never loads files or environment variables. An embedding
+/// application decides how configuration and genesis are obtained, then hands
+/// the resolved values here. `stop_epoch` overrides only the cloned chain
+/// configuration used by this domain; the caller's [`RootConfig`] is not
+/// mutated.
+pub struct DomainBuilder<'a> {
+    config: &'a RootConfig,
+    genesis: Arc<Genesis>,
+    stop_epoch: Option<u64>,
+}
+
+impl<'a> DomainBuilder<'a> {
+    /// Start a domain build with no stopping-epoch override.
+    pub fn new(config: &'a RootConfig, genesis: Arc<Genesis>) -> Self {
+        Self {
+            config,
+            genesis,
+            stop_epoch: None,
+        }
+    }
+
+    /// Override the configured Cardano stopping epoch for this domain.
+    pub fn stop_epoch(mut self, stop_epoch: Option<u64>) -> Self {
+        self.stop_epoch = stop_epoch;
+        self
+    }
+
+    /// Construct a normal node domain, including initialization and integrity
+    /// checks.
+    pub fn build(&self) -> Result<DomainAdapter, DomainBuildError> {
+        let stores = storage::open_data_stores(self.config).map_err(DomainBuildError::Storage)?;
+        self.build_with_stores(&stores)
+    }
+
+    fn build_with_stores(&self, stores: &Stores) -> Result<DomainAdapter, DomainBuildError> {
+        let ChainConfig::Cardano(mut chain_config) = self.config.chain.clone();
+
+        if let Some(stop_epoch) = self.stop_epoch {
+            chain_config.stop_epoch = Some(stop_epoch);
+        }
+
+        let chain = dolos_cardano::CardanoLogic::initialize::<DomainAdapter>(
+            chain_config,
+            &stores.state,
+            &self.genesis,
+        )
+        .map_err(DomainBuildError::Chain)?;
+
+        let (tip_broadcast, _) = tokio::sync::broadcast::channel(100);
+
+        let domain = DomainAdapter {
+            storage_config: Arc::new(self.config.storage.clone()),
+            sync_config: Arc::new(self.config.sync.clone()),
+            genesis: self.genesis.clone(),
+            chain: Arc::new(std::sync::RwLock::new(chain)),
+            wal: stores.wal.clone(),
+            state: stores.state.clone(),
+            archive: stores.archive.clone(),
+            mempool: stores.mempool.clone(),
+            tip_broadcast,
+        };
+
+        domain.bootstrap().map_err(DomainBuildError::Bootstrap)?;
+
+        Ok(domain)
+    }
+}
+
+/// An operation failed. Storage-specific details remain in the error source
+/// chain rather than becoming part of the replay protocol.
+#[derive(Debug, thiserror::Error)]
+#[error("{operation}: {source}")]
+pub struct BulkReplayError {
+    operation: &'static str,
+    #[source]
+    source: Cause,
+}
+
+impl BulkReplayError {
+    fn new(operation: &'static str, source: impl Into<Cause>) -> Self {
+        Self {
+            operation,
+            source: source.into(),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("{primary}; finalization also failed: {cleanup}")]
+struct WithCleanup {
+    #[source]
+    primary: Cause,
+    cleanup: Cause,
+}
+
+fn complete<T>(
+    result: Result<T, BulkReplayError>,
+    cleanup: Result<(), BulkReplayError>,
+) -> Result<T, BulkReplayError> {
+    match (result, cleanup) {
+        (Ok(value), Ok(())) => Ok(value),
+        (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
+        (Err(primary), Err(cleanup)) => Err(BulkReplayError::new(
+            "finalizing replay",
+            WithCleanup {
+                primary: Box::new(primary),
+                cleanup: Box::new(cleanup),
+            },
+        )),
+    }
+}
+
+fn flush(
+    wal: &WalAdapter,
+    state: &StateStoreBackend,
+    archive: &ArchiveStoreBackend,
+) -> Result<(), BulkReplayError> {
+    let wal = wal
+        .shutdown()
+        .map_err(|error| BulkReplayError::new("finishing replay", error));
+    let state = state
+        .shutdown()
+        .map_err(|error| BulkReplayError::new("finishing replay", error));
+    let archive = archive
+        .shutdown()
+        .map_err(|error| BulkReplayError::new("finishing replay", error));
+    complete(complete(wal, state), archive)
+}
+
+/// An exclusively owned replay dataset, opened without advancing chain state.
+///
+/// Use the borrowed profile view to inspect or publish pending data, then
+/// consume the workspace with `start` to advance it or `finish` to release it.
+/// Neither this handle nor its profile view exposes writable stores.
+#[must_use = "finish the workspace or consume it by starting replay"]
+pub struct ReplayWorkspace<'a> {
+    config: &'a RootConfig,
+    genesis: Arc<Genesis>,
+    stores: Stores,
+}
+
+impl<'a> ReplayWorkspace<'a> {
+    /// Open the configured dataset without running ledger initialization,
+    /// replaying blocks, or pruning history.
+    pub fn open(config: &'a RootConfig, genesis: Arc<Genesis>) -> Result<Self, BulkReplayError> {
+        let stores = storage::open_data_stores(config)
+            .map_err(|error| BulkReplayError::new("opening replay workspace", error))?;
+        Ok(Self {
+            config,
+            genesis,
+            stores,
+        })
+    }
+
+    /// Borrow read-only Dolos profile operations, not storage handles.
+    pub fn snapshot(&self) -> impl SnapshotSource + '_ {
+        StoreSnapshot::new(&self.stores.archive, &self.stores.state)
+    }
+
+    /// Begin processing, including any pending ledger initialization.
+    ///
+    /// Call only after publishing any pending boundary. The optional stopping
+    /// epoch overrides the configured value; `None` retains that configuration.
+    pub fn start(self, stop_epoch: Option<u64>) -> Result<BulkReplaySession, BulkReplayError> {
+        let result = checkpoint::reconcile(&self.stores.state, &self.stores.wal)
+            .map_err(|error| BulkReplayError::new("preparing replay", error))
+            .and_then(|()| {
+                DomainBuilder::new(self.config, self.genesis.clone())
+                    .stop_epoch(stop_epoch)
+                    .build_with_stores(&self.stores)
+                    .map_err(|error| BulkReplayError::new("starting replay", error))
+            });
+        match result {
+            Ok(domain) => Ok(BulkReplaySession {
+                domain,
+                boundary: None,
+                failed: false,
+            }),
+            Err(error) => complete(Err(error), self.finish()),
+        }
+    }
+
+    /// Release the inspection workspace without advancing or pruning it.
+    pub fn finish(self) -> Result<(), BulkReplayError> {
+        flush(&self.stores.wal, &self.stores.state, &self.stores.archive)
+    }
+}
+
+/// An exclusive session for trusted immutable blocks.
+///
+/// Session ownership cannot be duplicated:
+/// ```compile_fail
+/// fn duplicate(session: dolos::engine::BulkReplaySession) {
+///     let other = session.clone();
+/// }
+/// ```
+///
+/// A replay session does not expose the live-node domain interface:
+/// ```compile_fail
+/// fn live_domain<D: dolos::core::Domain>() {}
+/// live_domain::<dolos::engine::BulkReplaySession>();
+/// ```
+///
+/// A profile view prevents advancement while it is in use:
+/// ```compile_fail
+/// use dolos_snapshot::source::SnapshotSource;
+/// fn inspect(session: &mut dolos::engine::BulkReplaySession) {
+///     let snapshot = session.snapshot();
+///     session.import_blocks(vec![]).unwrap();
+///     snapshot.committed_position().unwrap();
+/// }
+/// ```
+///
+/// Finishing consumes the handle:
+/// ```compile_fail
+/// fn finish(mut session: dolos::engine::BulkReplaySession) {
+///     session.finish().unwrap();
+///     session.import_blocks(vec![]).unwrap();
+/// }
+/// ```
+///
+/// The session is neither cloneable nor a `Domain`: processing requires a
+/// mutable borrow, and finishing consumes it. After a boundary, further import
+/// calls report that same boundary without accepting more input. After an
+/// execution failure, finish and reopen rather than continuing a damaged
+/// session. An empty batch is rejected before execution and does not invalidate
+/// the session.
+///
+/// `run` finalizes on ordinary success and error returns. Dropping a session,
+/// panicking or killing the process is not a substitute for successful
+/// finalization; interrupted ledger transitions may require explicit repair.
+#[must_use = "finish the session or use run to finalize an operation"]
+pub struct BulkReplaySession {
+    domain: DomainAdapter,
+    boundary: Option<ChainPoint>,
+    failed: bool,
+}
+
+impl BulkReplaySession {
+    /// Open and immediately start replay when no pending export needs
+    /// inspection.
+    pub fn open(
+        config: &RootConfig,
+        genesis: Arc<Genesis>,
+        stop_epoch: Option<u64>,
+    ) -> Result<Self, BulkReplayError> {
+        ReplayWorkspace::open(config, genesis)?.start(stop_epoch)
+    }
+
+    /// Read the last committed input position.
+    pub fn committed_position(&self) -> Result<Option<ChainPoint>, BulkReplayError> {
+        self.domain
+            .state()
+            .read_cursor()
+            .map_err(|error| BulkReplayError::new("reading committed position", error))
+    }
+
+    /// Borrow profile operations over the currently committed dataset.
+    pub fn snapshot(&self) -> impl SnapshotSource + '_ {
+        StoreSnapshot::new(self.domain.archive(), self.domain.state())
+    }
+
+    /// Process a nonempty batch and report its committed position or boundary.
+    ///
+    /// A boundary can stop partway through a batch. Resume the source from the
+    /// returned position, not from the last block submitted.
+    pub fn import_blocks(
+        &mut self,
+        blocks: Vec<RawBlock>,
+    ) -> Result<ReplayProgress, BulkReplayError> {
+        if self.failed {
+            return Err(BulkReplayError::new(
+                "importing blocks",
+                "session failed; finish and reopen it",
+            ));
+        }
+        if let Some(position) = &self.boundary {
+            return Ok(ReplayProgress::Boundary {
+                position: position.clone(),
+            });
+        }
+        if blocks.is_empty() {
+            return Err(BulkReplayError::new(
+                "importing blocks",
+                "batch must not be empty",
+            ));
+        }
+        let boundary = match self.domain.import_blocks(blocks) {
+            Ok(_) => false,
+            Err(DomainError::StopEpochReached) => true,
+            Err(error) => {
+                self.failed = true;
+                return Err(BulkReplayError::new("importing blocks", error));
+            }
+        };
+        let position = match self.committed_position() {
+            Ok(Some(position)) => position,
+            result => {
+                self.failed = true;
+                return Err(result.err().unwrap_or_else(|| {
+                    BulkReplayError::new("importing blocks", "no committed position")
+                }));
+            }
+        };
+        if boundary {
+            self.boundary = Some(position.clone());
+            Ok(ReplayProgress::Boundary { position })
+        } else {
+            Ok(ReplayProgress::Committed { position })
+        }
+    }
+
+    /// Explicitly apply the configured history retention policy.
+    ///
+    /// The host must first publish any data it intends to preserve. Import and
+    /// finalization never call this operation automatically.
+    pub fn prune_history(&mut self) -> Result<u64, BulkReplayError> {
+        if self.failed || self.boundary.is_some() {
+            return Err(BulkReplayError::new(
+                "pruning history",
+                "finish the session before starting the next replay round",
+            ));
+        }
+        self.domain
+            .drain_housekeeping(None)
+            .map_err(|error| BulkReplayError::new("pruning history", error))
+    }
+
+    /// Persist completed work and release resources without advancing or
+    /// pruning.
+    ///
+    /// Successful finalization makes the committed position usable by a later
+    /// replay session or normal node startup. Resource finalization is
+    /// attempted even when the dataset cannot be made resumable.
+    pub fn finish(self) -> Result<(), BulkReplayError> {
+        let result = checkpoint::reconcile(self.domain.state(), self.domain.wal())
+            .map_err(|error| BulkReplayError::new("finishing replay", error));
+        let cleanup = flush(
+            self.domain.wal(),
+            self.domain.state(),
+            self.domain.archive(),
+        );
+        complete(result, cleanup)
+    }
+
+    /// Run an operation and finalize on both successful and failed returns.
+    pub fn run<T, E>(
+        mut self,
+        operation: impl FnOnce(&mut Self) -> Result<T, E>,
+    ) -> Result<T, BulkReplayRunError<E>> {
+        let result = operation(&mut self);
+        let finish = self.finish();
+        match (result, finish) {
+            (Ok(value), Ok(())) => Ok(value),
+            (Err(operation), Ok(())) => Err(BulkReplayRunError::Operation(operation)),
+            (Ok(_), Err(finish)) => Err(BulkReplayRunError::Finish(finish)),
+            (Err(operation), Err(finish)) => {
+                Err(BulkReplayRunError::OperationAndFinish { operation, finish })
+            }
+        }
+    }
+}
+
+/// Preserve both failures when an operation and its finalization fail.
+#[derive(Debug)]
+pub enum BulkReplayRunError<E> {
+    Operation(E),
+    Finish(BulkReplayError),
+    OperationAndFinish {
+        operation: E,
+        finish: BulkReplayError,
+    },
+}
+
+impl<E: fmt::Display> fmt::Display for BulkReplayRunError<E> {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Operation(error) => write!(formatter, "bulk replay failed: {error}"),
+            Self::Finish(error) => write!(formatter, "finishing bulk replay failed: {error}"),
+            Self::OperationAndFinish { operation, finish } => write!(
+                formatter,
+                "bulk replay failed ({operation}) and finalization also failed ({finish})"
+            ),
+        }
+    }
+}
+
+impl<E: std::error::Error + 'static> std::error::Error for BulkReplayRunError<E> {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Operation(error)
+            | Self::OperationAndFinish {
+                operation: error, ..
+            } => Some(error),
+            Self::Finish(error) => Some(error),
+        }
+    }
+}
+
+#[cfg(feature = "mithril")]
+impl dolos_snapshot::backfill::Workspace for ReplayWorkspace<'_> {
+    type Session = BulkReplaySession;
+
+    fn snapshot(&self) -> impl SnapshotSource + '_ {
+        self.snapshot()
+    }
+
+    fn start(self, target: u64) -> Result<Self::Session, dolos_snapshot::backfill::Error> {
+        self.start(Some(target))
+            .map_err(dolos_snapshot::backfill::Error::caller)
+    }
+
+    fn finish(self) -> Result<(), dolos_snapshot::backfill::Error> {
+        self.finish()
+            .map_err(dolos_snapshot::backfill::Error::caller)
+    }
+}
+
+#[cfg(feature = "mithril")]
+impl dolos_snapshot::backfill::Session for BulkReplaySession {
+    fn committed_position(&self) -> Result<Option<ChainPoint>, dolos_snapshot::backfill::Error> {
+        self.committed_position()
+            .map_err(dolos_snapshot::backfill::Error::caller)
+    }
+
+    fn import_blocks(
+        &mut self,
+        blocks: Vec<RawBlock>,
+    ) -> Result<ReplayProgress, dolos_snapshot::backfill::Error> {
+        self.import_blocks(blocks)
+            .map_err(dolos_snapshot::backfill::Error::caller)
+    }
+
+    fn prune_history(&mut self) -> Result<u64, dolos_snapshot::backfill::Error> {
+        self.prune_history()
+            .map_err(dolos_snapshot::backfill::Error::caller)
+    }
+
+    fn finish(self) -> Result<(), dolos_snapshot::backfill::Error> {
+        self.finish()
+            .map_err(dolos_snapshot::backfill::Error::caller)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn finalization_preserves_both_errors() {
+        let result = complete::<()>(
+            Err(BulkReplayError::new("operation", "first failure")),
+            Err(BulkReplayError::new("finish", "second failure")),
+        );
+        let error = result.unwrap_err().to_string();
+        assert!(error.contains("first failure"));
+        assert!(error.contains("second failure"));
+    }
+}
diff --git a/src/engine/checkpoint.rs b/src/engine/checkpoint.rs
new file mode 100644
index 000000000..fdd3eeca3
--- /dev/null
+++ b/src/engine/checkpoint.rs
@@ -0,0 +1,131 @@
+use dolos_core::{BlockSlot, ChainPoint, StateError, StateStore, WalError, WalStore};
+
+#[derive(Debug, thiserror::Error)]
+pub(super) enum CheckpointError {
+    #[error("reading replay position")]
+    ReadState(#[source] StateError),
+    #[error("reading replay checkpoint")]
+    ReadWal(#[source] WalError),
+    #[error("state cursor at slot {slot} has no block hash")]
+    UnanchoredState { slot: BlockSlot },
+    #[error("checkpoint {wal} and state {state} disagree")]
+    Diverged { wal: ChainPoint, state: ChainPoint },
+    #[error("checkpoint {wal} exists but state has no cursor")]
+    MissingState { wal: ChainPoint },
+    #[error("persisting replay checkpoint")]
+    ResetWal(#[source] WalError),
+}
+
+pub(super) fn reconcile<S: StateStore, W: WalStore>(
+    state: &S,
+    wal: &W,
+) -> Result<(), CheckpointError> {
+    let position = state.read_cursor().map_err(CheckpointError::ReadState)?;
+    let tip = wal
+        .find_tip()
+        .map_err(CheckpointError::ReadWal)?
+        .map(|(point, _)| point);
+
+    if let Some(position) = position.as_ref() {
+        if !position.is_fully_defined() {
+            return Err(CheckpointError::UnanchoredState {
+                slot: position.slot(),
+            });
+        }
+    }
+
+    match (tip, position) {
+        (None, None) => Ok(()),
+        (Some(ChainPoint::Origin), None) => Ok(()),
+        (Some(wal), None) => Err(CheckpointError::MissingState { wal }),
+        (Some(wal), Some(state)) if wal == state => Ok(()),
+        (Some(wal), Some(state)) if wal.slot() == state.slot() => {
+            Err(CheckpointError::Diverged { wal, state })
+        }
+        (Some(wal), Some(state)) if wal.slot() > state.slot() => Ok(()),
+        (_, Some(position)) => wal.reset_to(&position).map_err(CheckpointError::ResetWal),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dolos_core::{Domain as _, StateWriter as _};
+    use dolos_testing::toy_domain::ToyDomain;
+
+    fn set_position(domain: &ToyDomain, position: ChainPoint) {
+        let writer = domain.state().start_writer().unwrap();
+        writer.set_cursor(position).unwrap();
+        writer.commit().unwrap();
+    }
+
+    fn tip(domain: &ToyDomain) -> Option<ChainPoint> {
+        domain.wal().find_tip().unwrap().map(|(point, _)| point)
+    }
+
+    #[test]
+    fn missing_or_stale_checkpoint_is_reconciled_without_moving_state() {
+        for previous in [None, Some(ChainPoint::Specific(5, [1; 32].into()))] {
+            let domain = ToyDomain::new(None, None);
+            if let Some(previous) = previous {
+                domain.wal().reset_to(&previous).unwrap();
+            }
+            let position = ChainPoint::Specific(10, [2; 32].into());
+            set_position(&domain, position.clone());
+            reconcile(domain.state(), domain.wal()).unwrap();
+            assert_eq!(tip(&domain), Some(position.clone()));
+            assert_eq!(
+                domain.state().read_cursor().unwrap(),
+                Some(position.clone())
+            );
+            reconcile(domain.state(), domain.wal()).unwrap();
+            assert_eq!(tip(&domain), Some(position));
+        }
+    }
+
+    #[test]
+    fn later_checkpoint_is_left_for_normal_bootstrap() {
+        let domain = ToyDomain::new(None, None);
+        let state = ChainPoint::Specific(5, [1; 32].into());
+        let checkpoint = ChainPoint::Specific(10, [2; 32].into());
+        set_position(&domain, state.clone());
+        domain.wal().reset_to(&checkpoint).unwrap();
+        reconcile(domain.state(), domain.wal()).unwrap();
+        assert_eq!(tip(&domain), Some(checkpoint));
+        assert_eq!(domain.state().read_cursor().unwrap(), Some(state));
+    }
+
+    #[test]
+    fn divergent_and_unanchored_positions_are_not_overwritten() {
+        let domain = ToyDomain::new(None, None);
+        let checkpoint = ChainPoint::Specific(10, [2; 32].into());
+        domain.wal().reset_to(&checkpoint).unwrap();
+        set_position(&domain, ChainPoint::Specific(10, [3; 32].into()));
+        assert!(matches!(
+            reconcile(domain.state(), domain.wal()),
+            Err(CheckpointError::Diverged { .. })
+        ));
+        assert_eq!(tip(&domain), Some(checkpoint.clone()));
+        set_position(&domain, ChainPoint::Slot(11));
+        assert!(matches!(
+            reconcile(domain.state(), domain.wal()),
+            Err(CheckpointError::UnanchoredState { .. })
+        ));
+        assert_eq!(tip(&domain), Some(checkpoint));
+    }
+
+    #[test]
+    fn missing_state_is_only_allowed_for_a_fresh_or_origin_checkpoint() {
+        let state = dolos_core::builtin::MemoryStateStore::new();
+        let wal = dolos_redb3::wal::RedbWalStore::<dolos_cardano::CardanoDelta>::memory().unwrap();
+        reconcile(&state, &wal).unwrap();
+        wal.reset_to(&ChainPoint::Origin).unwrap();
+        reconcile(&state, &wal).unwrap();
+        wal.reset_to(&ChainPoint::Specific(10, [2; 32].into()))
+            .unwrap();
+        assert!(matches!(
+            reconcile(&state, &wal),
+            Err(CheckpointError::MissingState { .. })
+        ));
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 36bbe017b..17554cfa3 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,5 +1,6 @@
 pub mod adapters;
 pub mod cli;
+pub mod engine;
 pub mod prelude;
 pub mod relay;
 pub mod serve;
diff --git a/tests/engine.rs b/tests/engine.rs
new file mode 100644
index 000000000..81668768f
--- /dev/null
+++ b/tests/engine.rs
@@ -0,0 +1,305 @@
+mod node;
+
+#[cfg(feature = "mithril")]
+#[test]
+fn backfill_publishes_before_advancing_and_releases_failed_publications() {
+    use dolos_snapshot::{backfill, export::Plan, source::SnapshotSource};
+    use std::cell::Cell;
+
+    struct Publisher {
+        expected: ChainPoint,
+        fail: bool,
+        called: Cell<bool>,
+        cancel: tokio_util::sync::CancellationToken,
+    }
+
+    impl backfill::Publish for Publisher {
+        fn publish(&self, plan: &Plan, source: &dyn SnapshotSource) -> Result<(), backfill::Error> {
+            assert_eq!(plan.sequence, 1);
+            assert_eq!(
+                source.committed_position().unwrap(),
+                Some(self.expected.clone())
+            );
+            self.called.set(true);
+            if self.fail {
+                self.cancel.cancel();
+                Err(backfill::Error::caller(io::Error::other(
+                    "publication failed",
+                )))
+            } else {
+                Ok(())
+            }
+        }
+    }
+
+    let mut node = node::Node::new();
+    let mut genesis = dolos_cardano::include::preview::load();
+    genesis.force_protocol = Some(9);
+    let epoch_one = genesis.shelley.epoch_length.unwrap() as u64;
+    let genesis = Arc::new(genesis);
+    let (before, block_before) = make_conway_block_with_prev(epoch_one - 1, None, 1);
+    let (boundary, block_boundary) = make_conway_block_with_prev(epoch_one, before.hash(), 2);
+    node.config.chain = ChainConfig::Cardano(Default::default());
+    let mut session = BulkReplaySession::open(&node.config, genesis.clone(), Some(1)).unwrap();
+    session
+        .import_blocks(vec![block_before, block_boundary])
+        .unwrap();
+    session.finish().unwrap();
+
+    let mithril = dolos::core::config::MithrilConfig {
+        aggregator: "http://unused.invalid".to_owned(),
+        genesis_key: String::new(),
+        ancillary_key: None,
+    };
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    for fail in [true, false] {
+        let cancel = tokio_util::sync::CancellationToken::new();
+        let publish = Publisher {
+            expected: boundary.clone(),
+            fail,
+            called: Cell::new(false),
+            cancel: cancel.clone(),
+        };
+        let open_workspace = || {
+            ReplayWorkspace::open(&node.config, genesis.clone()).map_err(backfill::Error::caller)
+        };
+        let driver = backfill::Driver {
+            config: &node.config,
+            genesis: &genesis,
+            mithril: &mithril,
+            download_dir: node.root.path().join("unused"),
+            window: 1,
+            until_epoch: Some(1),
+            skip_validation: false,
+            runtime: runtime.handle().clone(),
+            cancel,
+            mithril_feedback: &|| None,
+            replay: &(),
+            open_workspace: &open_workspace,
+            publish: &publish,
+        };
+        let result = driver.run();
+        assert_eq!(result.is_err(), fail);
+        assert!(publish.called.get());
+        let workspace = ReplayWorkspace::open(&node.config, genesis.clone()).unwrap();
+        assert_eq!(
+            workspace.snapshot().committed_position().unwrap(),
+            Some(boundary.clone())
+        );
+        workspace.finish().unwrap();
+    }
+}
+
+use std::io;
+use std::sync::Arc;
+
+use dolos::core::config::ChainConfig;
+use dolos::core::{ChainPoint, Domain as _, RawBlock, StateStore, StateWriter as _};
+use dolos::engine::{
+    BulkReplayRunError, BulkReplaySession, DomainBuilder, ReplayProgress, ReplayWorkspace,
+};
+use dolos_snapshot::source::SnapshotSource as _;
+use dolos_testing::blocks::make_conway_block_with_prev;
+use dolos_testing::synthetic::{build_synthetic_blocks, SyntheticBlockConfig};
+
+fn fixture() -> (node::Node, Arc<dolos::core::Genesis>, Vec<RawBlock>) {
+    let mut node = node::Node::new();
+    let genesis = Arc::new(dolos_cardano::include::preview::load());
+    let (blocks, _, chain) = build_synthetic_blocks(SyntheticBlockConfig {
+        block_count: 4,
+        slot: 100,
+        ..Default::default()
+    });
+    node.config.chain = ChainConfig::Cardano(chain);
+    (node, genesis, blocks)
+}
+
+#[test]
+fn public_builder_cleanly_initializes_configured_stores() {
+    let (node, genesis, _) = fixture();
+    let domain = DomainBuilder::new(&node.config, genesis).build().unwrap();
+    assert_eq!(domain.state().read_cursor().unwrap(), None);
+    domain.shutdown().unwrap();
+}
+
+#[test]
+fn inspection_does_not_initialize_a_fresh_workspace() {
+    let (node, genesis, _) = fixture();
+    let workspace = ReplayWorkspace::open(&node.config, genesis).unwrap();
+    assert_eq!(workspace.snapshot().committed_position().unwrap(), None);
+    assert_eq!(workspace.snapshot().epoch().unwrap(), None);
+    workspace.finish().unwrap();
+}
+
+#[test]
+fn configured_boundary_survives_inspection_and_requires_explicit_advancement() {
+    let mut node = node::Node::new();
+    let mut genesis = dolos_cardano::include::preview::load();
+    genesis.force_protocol = Some(9);
+    let epoch_one = genesis.shelley.epoch_length.unwrap() as u64;
+    let genesis = Arc::new(genesis);
+    let (before, block_before) = make_conway_block_with_prev(epoch_one - 1, None, 1);
+    let (boundary, block_boundary) = make_conway_block_with_prev(epoch_one, before.hash(), 2);
+    let (after, block_after) = make_conway_block_with_prev(epoch_one + 1, boundary.hash(), 3);
+    node.config.chain = ChainConfig::Cardano(Default::default());
+
+    let mut session = BulkReplaySession::open(&node.config, genesis.clone(), Some(1)).unwrap();
+    let progress = session
+        .import_blocks(vec![block_before, block_boundary, block_after.clone()])
+        .unwrap();
+    assert_eq!(
+        progress,
+        ReplayProgress::Boundary {
+            position: boundary.clone()
+        }
+    );
+    assert_eq!(
+        session.import_blocks(vec![block_after.clone()]).unwrap(),
+        progress
+    );
+    assert!(session.prune_history().is_err());
+    session.finish().unwrap();
+
+    for _attempt in 0..2 {
+        let workspace = ReplayWorkspace::open(&node.config, genesis.clone()).unwrap();
+        let snapshot = workspace.snapshot();
+        assert_eq!(
+            snapshot.committed_position().unwrap(),
+            Some(boundary.clone())
+        );
+        assert_eq!(snapshot.epoch().unwrap(), Some(1));
+        let plan = snapshot
+            .plan(
+                u64::from(genesis.network_magic()),
+                dolos_snapshot::planning::retained_epochs(&node.config).unwrap(),
+            )
+            .unwrap();
+        assert_eq!(plan.sequence, 1);
+        drop(snapshot);
+        workspace.finish().unwrap();
+    }
+
+    let workspace = ReplayWorkspace::open(&node.config, genesis).unwrap();
+    let mut session = workspace.start(Some(2)).unwrap();
+    assert_eq!(
+        session.import_blocks(vec![block_after]).unwrap().position(),
+        &after
+    );
+    session.finish().unwrap();
+}
+
+#[test]
+fn unfinished_replay_resumes_from_its_committed_input() {
+    let (node, genesis, blocks) = fixture();
+    let mut session = BulkReplaySession::open(&node.config, genesis.clone(), None).unwrap();
+    let committed = session
+        .import_blocks(blocks[..2].to_vec())
+        .unwrap()
+        .position()
+        .clone();
+    drop(session);
+
+    let workspace = ReplayWorkspace::open(&node.config, genesis.clone()).unwrap();
+    assert_eq!(
+        workspace.snapshot().committed_position().unwrap(),
+        Some(committed.clone())
+    );
+    let mut resumed = workspace.start(None).unwrap();
+    assert_eq!(
+        resumed.committed_position().unwrap(),
+        Some(committed.clone())
+    );
+    let final_position = resumed
+        .import_blocks(blocks[2..].to_vec())
+        .unwrap()
+        .position()
+        .clone();
+    assert!(final_position.slot() > committed.slot());
+    resumed.finish().unwrap();
+
+    let domain = DomainBuilder::new(&node.config, genesis).build().unwrap();
+    assert_eq!(domain.state().read_cursor().unwrap(), Some(final_position));
+    domain.shutdown().unwrap();
+}
+
+#[test]
+fn operation_failure_still_finalizes_for_normal_node_startup() {
+    let (node, genesis, blocks) = fixture();
+    let session = BulkReplaySession::open(&node.config, genesis.clone(), None).unwrap();
+    let mut committed = None;
+    let result = session.run(|session| {
+        committed = Some(session.import_blocks(blocks).unwrap().position().clone());
+        Err::<(), _>(io::Error::other("failure after import"))
+    });
+    assert!(matches!(result, Err(BulkReplayRunError::Operation(_))));
+    let domain = DomainBuilder::new(&node.config, genesis).build().unwrap();
+    assert_eq!(domain.state().read_cursor().unwrap(), committed);
+    domain.shutdown().unwrap();
+}
+
+#[test]
+fn success_finalizes_for_normal_node_startup() {
+    let (node, genesis, blocks) = fixture();
+    let session = BulkReplaySession::open(&node.config, genesis.clone(), None).unwrap();
+    let committed = session
+        .run(|session| session.import_blocks(blocks))
+        .unwrap()
+        .position()
+        .clone();
+    let domain = DomainBuilder::new(&node.config, genesis).build().unwrap();
+    assert_eq!(domain.state().read_cursor().unwrap(), Some(committed));
+    domain.shutdown().unwrap();
+}
+
+#[test]
+fn empty_input_does_not_poison_a_session() {
+    let (node, genesis, blocks) = fixture();
+    let mut session = BulkReplaySession::open(&node.config, genesis, None).unwrap();
+    assert!(session.import_blocks(vec![]).is_err());
+    assert_eq!(session.committed_position().unwrap(), None);
+    session.import_blocks(blocks).unwrap();
+    session.finish().unwrap();
+}
+
+#[test]
+fn import_failure_requires_reopening() {
+    let (node, genesis, blocks) = fixture();
+    let mut session = BulkReplaySession::open(&node.config, genesis.clone(), None).unwrap();
+    assert!(session.import_blocks(vec![Arc::new(vec![0xff])]).is_err());
+    assert!(session
+        .import_blocks(blocks.clone())
+        .unwrap_err()
+        .to_string()
+        .contains("finish and reopen"));
+    session.finish().unwrap();
+    let mut session = BulkReplaySession::open(&node.config, genesis, None).unwrap();
+    session.import_blocks(blocks).unwrap();
+    session.finish().unwrap();
+}
+
+#[test]
+fn unanchored_position_is_not_silently_repaired_and_failed_start_releases_stores() {
+    let (node, genesis, _) = fixture();
+    let stores =
+        dolos::storage::open_data_stores::<dolos_cardano::CardanoDelta>(&node.config).unwrap();
+    let writer = stores.state.start_writer().unwrap();
+    writer.set_cursor(ChainPoint::Slot(42)).unwrap();
+    writer.commit().unwrap();
+    stores.state.shutdown().unwrap();
+    drop(stores);
+
+    let workspace = ReplayWorkspace::open(&node.config, genesis.clone()).unwrap();
+    assert!(workspace.snapshot().epoch().is_err());
+    let error = workspace.start(None).err().unwrap();
+    assert!(error.to_string().contains("no block hash"));
+
+    let workspace = ReplayWorkspace::open(&node.config, genesis).unwrap();
+    assert_eq!(
+        workspace.snapshot().committed_position().unwrap(),
+        Some(ChainPoint::Slot(42))
+    );
+    workspace.finish().unwrap();
+}