diff --git a/Cargo.lock b/Cargo.lock
index c7ac9567..70785143 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4191,6 +4191,7 @@ dependencies = [
  "lance",
  "lance-context-core",
  "lance-context-wal",
+ "lance-io",
  "lance-table",
  "object_store 0.14.2",
  "rcgen",
diff --git a/crates/lance-context-ingest/Cargo.toml b/crates/lance-context-ingest/Cargo.toml
index bc54f5e3..18db9f93 100644
--- a/crates/lance-context-ingest/Cargo.toml
+++ b/crates/lance-context-ingest/Cargo.toml
@@ -10,7 +10,7 @@ description = "Durable rollout ingress over the independent lance-context WAL"
 worker = ["materializer", "dep:etcd-client"]
 http = ["dep:axum", "dep:http-body-util"]
 http-client = ["http", "dep:reqwest", "lance-context-wal/client-journal"]
-materializer = ["lance-context-wal/lance-integration", "dep:lance", "dep:lance-table", "dep:object_store", "dep:futures"]
+materializer = ["lance-context-wal/lance-integration", "dep:lance", "dep:lance-table", "dep:lance-io", "dep:object_store", "dep:futures"]
 
 [dependencies]
 etcd-client = { version = "0.19", features = ["tls"], optional = true }
@@ -19,6 +19,7 @@ http-body-util = { version = "0.1", optional = true }
 reqwest = { version = "0.12", optional = true, default-features = false, features = ["json", "rustls-tls"] }
 lance = { version = "13.0.0", optional = true }
 lance-table = { version = "13.0.0", optional = true }
+lance-io = { version = "13.0.0", optional = true }
 object_store = { version = "0.14.2", optional = true }
 futures = { version = "0.3", optional = true }
 anyhow = "1"
diff --git a/crates/lance-context-ingest/src/bootstrap.rs b/crates/lance-context-ingest/src/bootstrap.rs
new file mode 100644
index 00000000..0ba69a59
--- /dev/null
+++ b/crates/lance-context-ingest/src/bootstrap.rs
@@ -0,0 +1,279 @@
+//! Idempotent provisioning of a NEW managed rollout generation. Persist the
+//! binding, URI and complete operation identity before the first call. This is
+//! not an importer for existing datasets or a legacy writer migration.
+use super::*;
+use crate::materializer::Storage;
+use arrow_array::RecordBatchIterator;
+use lance::{
+    dataset::{builder::DatasetBuilder, WriteMode, WriteParams},
+    Dataset,
+};
+use lance_context_wal::replication::publication as p;
+use lance_io::{
+    object_store::StorageOptionsAccessor,
+    object_store::{ObjectStore, ObjectStoreParams},
+};
+use lance_table::io::commit::{ConditionalPutCommitHandler, ManifestNamingScheme};
+use std::collections::HashMap;
+
+const GENESIS: &str = "lance-context.ingest.genesis.v1";
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Identity {
+    pub owner: Uuid,
+    pub acquisition: Uuid,
+    pub manifest: Uuid,
+}
+
+impl Storage {
+    /// Resolve object-store paths with Lance's own URI/credential handling.
+    /// Storage options may select workload identity; no credentials enter WAL.
+    pub async fn resolve(
+        uri: String,
+        options: HashMap<String, String>,
+        session: Arc<lance::session::Session>,
+    ) -> Result<Self> {
+        ensure!(!uri.is_empty() && uri.len() <= 8192, "invalid dataset URI");
+        let (_, base) =
+            ObjectStore::from_uri_and_params(session.store_registry(), &uri, &parameters(&options))
+                .await?;
+        Ok(Self {
+            uri,
+            base,
+            options,
+            session,
+        })
+    }
+}
+fn parameters(options: &HashMap<String, String>) -> ObjectStoreParams {
+    ObjectStoreParams {
+        storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options(
+            options.clone(),
+        ))),
+        ..Default::default()
+    }
+}
+async fn state(client: &Client, scope: &Scope) -> Result<Option<Box<p::Table>>> {
+    let group = client.catalog_group();
+    let response = client
+        .request_routed(Request::Query {
+            group,
+            query: query::Query::Publisher { table: scope.table },
+        })
+        .await?;
+    let Reply::Query(query::Outcome::Observed {
+        cut,
+        value: query::Value::Publisher(view),
+    }) = response.reply
+    else {
+        bail!("bootstrap quorum observation unavailable");
+    };
+    ensure!(
+        cut.cluster == scope.cluster
+            && cut.layout == scope.layout
+            && cut.group == group
+            && cut.term > 0
+            && cut.quorum_index > 0
+            && cut.applied >= cut.quorum_index,
+        "foreign bootstrap observation"
+    );
+    if let Some(view) = &view {
+        ensure!(
+            view.destination == scope.destination && view.ingest == scope.contract,
+            "dataset generation already bound differently"
+        );
+    }
+    Ok(view)
+}
+async fn commit(client: &Client, command: Command) -> Result<Applied> {
+    let response = client
+        .request_routed(Request::Commit(Box::new(command)))
+        .await?;
+    let Reply::Commit(Outcome::Committed(applied)) = response.reply else {
+        bail!("bootstrap outcome unknown; retry the same persisted identity");
+    };
+    Ok(applied)
+}
+
+/// Provision the reserved generation at an application-authorized URI. Only a
+/// newly created empty dataset carrying this exact bootstrap identity can become
+/// the catalog's genesis. A timeout never generates replacement operation IDs.
+pub async fn provision(
+    client: Arc<Client>,
+    layout: Arc<Layout>,
+    binding: TableBinding,
+    storage: Storage,
+    identity: Identity,
+) -> Result<Storage> {
+    let scope = Scope {
+        cluster: layout.cluster,
+        layout: layout.fingerprint()?,
+        table: binding.table,
+        destination: binding.destination.clone(),
+        incarnation: binding.incarnation,
+        contract: binding.contract()?,
+    };
+    ensure!(
+        client.identity() == (scope.cluster, scope.layout)
+            && !scope.table.is_nil()
+            && !scope.incarnation.is_nil()
+            && !identity.owner.is_nil()
+            && !identity.acquisition.is_nil()
+            && !identity.manifest.is_nil()
+            && !scope.destination.namespace.is_nil()
+            && !scope.destination.dataset.is_empty()
+            && client.producer_slots() > 0
+            && scope.contract.max_record_bytes as usize <= client.max_record_bytes(),
+        "invalid bootstrap identity"
+    );
+    let existing = state(&client, &scope).await?;
+    if let Some(manifests) = existing.as_ref().and_then(|v| v.manifests.as_ref()) {
+        ensure!(
+            manifests.incarnation == scope.incarnation
+                && manifests.bootstrap_request == identity.manifest
+                && manifests.initial_version == 1,
+            "bootstrap identity differs from durable generation"
+        );
+        // After initialization, use only the catalog-selected manifest. The
+        // canonical genesis path may be old; never rediscover visibility there.
+        let reader = p::lance::ManagedManifest::reader(
+            Arc::new(client.manifest_catalog()),
+            storage.base.clone(),
+            &layout,
+            scope.table,
+            scope.incarnation,
+        )?;
+        let dataset = DatasetBuilder::from_uri(&storage.uri)
+            .with_storage_options(storage.options.clone())
+            .with_session(storage.session.clone())
+            .with_commit_handler(Arc::new(reader))
+            .load()
+            .await?;
+        let schema: arrow_schema::Schema = dataset.schema().into();
+        ensure!(
+            schema == rollout_schema(),
+            "managed generation schema differs"
+        );
+        return Ok(storage);
+    }
+    let applied = commit(
+        &client,
+        Command::Acquire(p::Acquire {
+            table: scope.table,
+            destination: scope.destination.clone(),
+            ingest: scope.contract.clone(),
+            owner: identity.owner,
+            request: identity.acquisition,
+            expected_epoch: 0,
+        }),
+    )
+    .await?;
+    let Applied::Publisher(p::Reply::Acquired(authority)) = applied else {
+        bail!("bootstrap acquisition rejected: {applied:?}");
+    };
+    ensure!(
+        authority.owner == identity.owner
+            && authority.table == scope.table
+            && authority.epoch == 1
+            && authority.cluster == scope.cluster
+            && authority.layout == scope.layout
+            && authority.group == client.catalog_group(),
+        "foreign bootstrap authority"
+    );
+    let (object, transaction) = genesis(&storage, &scope, &identity).await?;
+    let applied = commit(
+        &client,
+        Command::Bootstrap(p::BootstrapManifest {
+            authority,
+            incarnation: scope.incarnation,
+            request: identity.manifest,
+            version: 1,
+            object: object.clone(),
+            transaction,
+        }),
+    )
+    .await?;
+    let Applied::Publisher(p::Reply::ManifestReserved(selected)) = applied else {
+        bail!("genesis selection not confirmed: {applied:?}");
+    };
+    ensure!(
+        selected.version == 1 && selected.object == object && selected.transaction == transaction,
+        "catalog selected another genesis"
+    );
+    Ok(storage)
+}
+
+pub(crate) async fn genesis(
+    storage: &Storage,
+    scope: &Scope,
+    identity: &Identity,
+) -> Result<(p::ManifestObject, Uuid)> {
+    let marker = serde_json::to_string(&(scope, identity))?;
+    let parameters = WriteParams {
+        mode: WriteMode::Create,
+        store_params: Some(parameters(&storage.options)),
+        session: Some(storage.session.clone()),
+        commit_handler: Some(Arc::new(ConditionalPutCommitHandler)),
+        transaction_properties: Some(Arc::new(HashMap::from([(
+            GENESIS.to_owned(),
+            marker.clone(),
+        )]))),
+        ..Default::default()
+    };
+    let schema = Arc::new(rollout_schema());
+    let empty = RecordBatch::new_empty(schema.clone());
+    let dataset = match Dataset::write(
+        RecordBatchIterator::new(vec![Ok(empty)], schema),
+        &storage.uri,
+        Some(parameters),
+    )
+    .await
+    {
+        Ok(dataset) => dataset,
+        Err(lance::Error::DatasetAlreadyExists { .. }) => {
+            DatasetBuilder::from_uri(&storage.uri)
+                .with_version(1)
+                .with_storage_options(storage.options.clone())
+                .with_session(storage.session.clone())
+                .with_commit_handler(Arc::new(ConditionalPutCommitHandler))
+                .load()
+                .await?
+        }
+        Err(error) => return Err(error.into()),
+    };
+    let schema: arrow_schema::Schema = dataset.schema().into();
+    let transaction = dataset
+        .read_transaction_by_version(1)
+        .await?
+        .context("genesis transaction missing")?;
+    ensure!(
+        dataset.version().version == 1
+            && schema == rollout_schema()
+            && dataset.count_rows(None).await? == 0
+            && transaction
+                .transaction_properties
+                .as_ref()
+                .and_then(|p| p.get(GENESIS))
+                == Some(&marker),
+        "refusing to adopt a foreign or nonempty genesis"
+    );
+    let location = dataset.manifest_location();
+    ensure!(
+        location.path.parent().and_then(|p| p.parent()).as_ref() == Some(&storage.base),
+        "genesis path differs from configured storage"
+    );
+    let object = p::ManifestObject {
+        path: format!(
+            "_versions/{}",
+            location
+                .path
+                .filename()
+                .context("genesis filename missing")?
+        ),
+        bytes: location.size.context("genesis size missing")?,
+        v2_names: location.naming_scheme == ManifestNamingScheme::V2,
+    };
+    let transaction = Uuid::parse_str(&transaction.uuid)?;
+    Ok((object, transaction))
+}
diff --git a/crates/lance-context-ingest/src/lib.rs b/crates/lance-context-ingest/src/lib.rs
index 6e48498f..8cc69117 100644
--- a/crates/lance-context-ingest/src/lib.rs
+++ b/crates/lance-context-ingest/src/lib.rs
@@ -29,6 +29,8 @@ use serde::{Deserialize, Serialize};
 use std::{collections::HashSet, sync::Arc, time::Duration};
 use uuid::Uuid;
 
+#[cfg(feature = "materializer")]
+pub mod bootstrap;
 #[cfg(feature = "http")]
 pub mod http;
 #[cfg(feature = "materializer")]
@@ -38,7 +40,8 @@ pub mod receipt;
 pub mod worker;
 
 /// Resolved by the application's authorized catalog, not an uploaded URI/UUID.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct TableBinding {
     pub table: Uuid,
     pub destination: Destination,
diff --git a/crates/lance-context-ingest/src/tests.rs b/crates/lance-context-ingest/src/tests.rs
index d069b698..8486481f 100644
--- a/crates/lance-context-ingest/src/tests.rs
+++ b/crates/lance-context-ingest/src/tests.rs
@@ -1,15 +1,22 @@
 use super::*;
 use arrow_array::{
     new_null_array, types::Int8Type, ArrayRef, DictionaryArray, Int32Array, Int8Array,
-    LargeBinaryArray, RecordBatch, RecordBatchIterator, TimestampMicrosecondArray,
+    LargeBinaryArray, RecordBatch, TimestampMicrosecondArray,
 };
 use arrow_ipc::writer::FileWriter;
-use lance::{dataset::WriteParams, Dataset};
+use lance::Dataset;
 use lance_context_wal::replication::publication as p;
-use lance_table::io::commit::{ConditionalPutCommitHandler, ManifestNamingScheme};
 use receipt::ReceiptKeys;
 use std::collections::BTreeMap;
+#[cfg(not(feature = "materializer"))]
+use {
+    arrow_array::RecordBatchIterator,
+    lance::dataset::WriteParams,
+    lance_table::io::commit::{ConditionalPutCommitHandler, ManifestNamingScheme},
+};
 
+#[cfg(feature = "materializer")]
+mod bootstrap;
 #[cfg(all(feature = "http-client", feature = "materializer"))]
 mod http;
 #[cfg(feature = "materializer")]
@@ -181,6 +188,30 @@ async fn validates_whole_atomic_batch_and_preserves_original_buffer() -> Result<
     Ok(())
 }
 
+#[cfg(feature = "materializer")]
+async fn bootstrap(
+    c: &network::Cluster,
+    path: &std::path::Path,
+    binding: &TableBinding,
+) -> Result<()> {
+    let storage = crate::materializer::Storage::resolve(
+        path.to_str().context("test URI")?.into(),
+        Default::default(),
+        Arc::new(lance::session::Session::default()),
+    )
+    .await?;
+    crate::bootstrap::provision(
+        c.client.clone(),
+        layout(),
+        binding.clone(),
+        storage,
+        bootstrap::identity(),
+    )
+    .await?;
+    Ok(())
+}
+
+#[cfg(not(feature = "materializer"))]
 async fn bootstrap(
     c: &network::Cluster,
     path: &std::path::Path,
diff --git a/crates/lance-context-ingest/src/tests/bootstrap.rs b/crates/lance-context-ingest/src/tests/bootstrap.rs
new file mode 100644
index 00000000..18014c77
--- /dev/null
+++ b/crates/lance-context-ingest/src/tests/bootstrap.rs
@@ -0,0 +1,176 @@
+use super::*;
+use crate::{
+    bootstrap::{provision, Identity},
+    materializer::Storage,
+};
+use arrow_array::RecordBatchIterator;
+use lance::dataset::WriteParams;
+use lance_table::io::commit::ConditionalPutCommitHandler;
+
+pub(super) fn identity() -> Identity {
+    Identity {
+        owner: Uuid::from_u128(9000),
+        acquisition: Uuid::from_u128(9001),
+        manifest: Uuid::from_u128(9002),
+    }
+}
+async fn view(c: &network::Cluster, table: Uuid) -> Result<Box<p::Table>> {
+    let response = c
+        .client
+        .request_routed(Request::Query {
+            group: c.client.catalog_group(),
+            query: query::Query::Publisher { table },
+        })
+        .await?;
+    let Reply::Query(query::Outcome::Observed {
+        value: query::Value::Publisher(Some(view)),
+        ..
+    }) = response.reply
+    else {
+        bail!("missing publisher state");
+    };
+    Ok(view)
+}
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn production_bootstrap_restarts_idempotently_and_refuses_foreign_empty_genesis() -> Result<()>
+{
+    let root = tempfile::tempdir()?;
+    let path = root.path().join("managed.lance");
+    let storage = Storage::resolve(
+        path.to_str().unwrap().into(),
+        Default::default(),
+        Arc::new(lance::session::Session::default()),
+    )
+    .await?;
+    let c = network::Cluster::open_signed(root.path(), layout()).await?;
+    provision(
+        c.client.clone(),
+        layout(),
+        binding(),
+        storage.clone(),
+        identity(),
+    )
+    .await?;
+    let original = view(&c, binding().table).await?;
+    assert_eq!(original.authority.epoch, 1);
+    assert_eq!(original.manifests.as_ref().unwrap().head, 1);
+    provision(
+        c.client.clone(),
+        layout(),
+        binding(),
+        storage.clone(),
+        identity(),
+    )
+    .await?;
+    assert_eq!(view(&c, binding().table).await?, original);
+    c.shutdown().await?;
+    let c = network::Cluster::open_signed(root.path(), layout()).await?;
+    provision(
+        c.client.clone(),
+        layout(),
+        binding(),
+        storage.clone(),
+        identity(),
+    )
+    .await?;
+    assert_eq!(view(&c, binding().table).await?, original);
+    let mut foreign_identity = identity();
+    foreign_identity.manifest = Uuid::new_v4();
+    assert!(provision(
+        c.client.clone(),
+        layout(),
+        binding(),
+        storage,
+        foreign_identity
+    )
+    .await
+    .is_err());
+    assert_eq!(view(&c, binding().table).await?, original);
+
+    let path = root.path().join("foreign.lance");
+    let foreign = Storage::resolve(
+        path.to_str().unwrap().into(),
+        Default::default(),
+        Arc::new(lance::session::Session::default()),
+    )
+    .await?;
+    Dataset::write(
+        RecordBatchIterator::new(vec![Ok(batch(&[], &[])?)], Arc::new(rollout_schema())),
+        &foreign.uri,
+        Some(WriteParams {
+            commit_handler: Some(Arc::new(ConditionalPutCommitHandler)),
+            ..Default::default()
+        }),
+    )
+    .await?;
+    let mut other = binding();
+    other.table = Uuid::new_v4();
+    other.destination.dataset = "foreign.lance".into();
+    assert!(provision(
+        c.client.clone(),
+        layout(),
+        other.clone(),
+        foreign.clone(),
+        identity()
+    )
+    .await
+    .is_err());
+    assert!(view(&c, other.table).await?.manifests.is_none());
+    let untouched = Dataset::open(&foreign.uri).await?;
+    assert_eq!(untouched.version().version, 1);
+    assert_eq!(untouched.count_rows(None).await?, 0);
+    c.shutdown().await
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn bootstrap_reuses_original_genesis_after_restart_before_catalog_selection() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let path = root.path().join("interrupted.lance");
+    let storage = Storage::resolve(
+        path.to_str().unwrap().into(),
+        Default::default(),
+        Arc::new(lance::session::Session::default()),
+    )
+    .await?;
+    let c = network::Cluster::open_signed(root.path(), layout()).await?;
+    let b = binding();
+    let id = identity();
+    let response = c
+        .client
+        .request_routed(Request::Commit(Box::new(Command::Acquire(p::Acquire {
+            table: b.table,
+            destination: b.destination.clone(),
+            ingest: b.contract()?,
+            owner: id.owner,
+            request: id.acquisition,
+            expected_epoch: 0,
+        }))))
+        .await?;
+    assert!(matches!(
+        response.reply,
+        Reply::Commit(Outcome::Committed(Applied::Publisher(p::Reply::Acquired(
+            _
+        ))))
+    ));
+    let scope = Scope {
+        cluster: layout().cluster,
+        layout: layout().fingerprint()?,
+        table: b.table,
+        destination: b.destination.clone(),
+        incarnation: b.incarnation,
+        contract: b.contract()?,
+    };
+    let (object, transaction) = crate::bootstrap::genesis(&storage, &scope, &id).await?;
+    assert!(view(&c, b.table).await?.manifests.is_none());
+    // The actual genesis file/transaction exists, but no manifest was selected.
+    // Reopen all original replica directories and retry the persisted identity.
+    c.shutdown().await?;
+    let c = network::Cluster::open_signed(root.path(), layout()).await?;
+    provision(c.client.clone(), layout(), b.clone(), storage, id).await?;
+    let state = view(&c, b.table).await?;
+    assert_eq!(state.authority.epoch, 1);
+    let selected = &state.manifests.as_ref().unwrap().versions[&1];
+    assert_eq!(selected.object, object);
+    assert_eq!(selected.transaction, transaction);
+    c.shutdown().await
+}
diff --git a/docs/design/wal-ingest-bootstrap.md b/docs/design/wal-ingest-bootstrap.md
new file mode 100644
index 00000000..d0d097de
--- /dev/null
+++ b/docs/design/wal-ingest-bootstrap.md
@@ -0,0 +1,39 @@
+# Provisioning a managed rollout generation
+
+`bootstrap::provision` creates a new empty rollout generation and selects its
+genesis manifest through the durable WAL catalog. The application must reserve
+and persist the authorized table binding, dataset URI, publisher owner, acquisition
+request and manifest request UUIDs before calling it. A timeout is retried with
+that same configuration. This is not an existing-table importer or a legacy WAL
+migration.
+
+`Storage::resolve` uses Lance's own object-store registry and URI/credential
+handling to derive the manifest base path. Credentials stay in application storage
+options and never enter replicated metadata. Readers and publishers can share
+the same Lance session and object-store cache.
+
+Initialization acquires the initial WAL publisher epoch, creates an empty Lance
+dataset in create-only mode, and writes the complete binding and operation identity
+into the genesis transaction properties. If the dataset already exists, it opens
+version 1 and requires the exact marker, rollout schema and zero rows before it
+can select that manifest. An unrelated empty dataset is also rejected. Other
+write failures remain errors; there is no overwrite or automatic file deletion.
+
+After catalog initialization, retries use only the catalog-selected manifest.
+They do not read a canonical genesis path to discover visibility, and they do not
+take ownership from a running publisher. A different table contract, incarnation
+or manifest request is rejected. The operation does not generate replacement IDs
+on restart or silently migrate an existing namespace binding.
+
+The application still needs to publish the completed binding into its authorized
+namespace and provide it to gateways and materializer workers. This initializer
+alone is not a dynamic namespace registry or a deployment executable.
+
+Focused qualification passed seven ordinary integration tests, the two real-etcd
+worker tests and all three checked feature Clippy configurations. The bootstrap
+tests use actual TLS WAL replicas and Lance datasets, including all-original-directory restart before
+genesis selection, exact original genesis transaction reuse, completed-operation
+retries, and rejection of a foreign empty genesis. The existing HTTP/materializer
+suite also uses this initializer when its materializer feature is enabled.
+
+Tracking: #378.