diff --git a/Cargo.lock b/Cargo.lock
index 72ecd6c..08145f5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4228,6 +4228,7 @@ dependencies = [
  "raft",
  "raft-engine",
  "raft-proto",
+ "ring",
  "serde",
  "serde_json",
  "sha2 0.10.9",
diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml
index cf0ac64..7f23285 100644
--- a/crates/lance-context-wal/Cargo.toml
+++ b/crates/lance-context-wal/Cargo.toml
@@ -8,7 +8,7 @@ description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
 client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"]
-replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "uuid/v4", "serde/rc"]
+replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "dep:ring", "uuid/v4", "serde/rc"]
 raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"]
 vectored-io = ["dep:bytes", "dep:tokio"]
 runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"]
@@ -21,6 +21,7 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de
 lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
+ring = { version = "0.17", optional = true }
 bincode = { version = "1.3", optional = true }
 slog = { version = "2", optional = true }
 tokio = { version = "1", features = ["io-util"], optional = true }
diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md
index 13dfa7c..5674918 100644
--- a/crates/lance-context-wal/README.md
+++ b/crates/lance-context-wal/README.md
@@ -178,8 +178,10 @@ They share pending capacity with proposals and do not fetch bulk payloads.
 
 [Immutable publication plans](../../docs/design/wal-publication-plans.md) retain
 the exact source vector and Lance target through takeover, snapshots and restart.
-They share bounded metadata between views; source pins, outcome resolution and
-retirement remain separate integration steps.
+They share bounded metadata between views. [Source pins](../../docs/design/wal-source-pins.md)
+verify signed catalog quorum observations against complete source prefixes and
+survive snapshots and restart. Outcome resolution and retirement remain separate
+integration steps; neither a plan nor a pin permits an external Lance commit.
 
 The optional `raft-log` feature exposes `raft_log::PhysicalLog` and its cloneable
 `GroupLog` handles. One raft-engine directory hosts all groups assigned to a node
diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs
index bea1a9d..7415dd4 100644
--- a/crates/lance-context-wal/src/replication.rs
+++ b/crates/lance-context-wal/src/replication.rs
@@ -16,6 +16,7 @@ pub mod checkpoint;
 mod images;
 pub mod publication;
 pub mod query;
+pub mod source;
 mod state;
 #[cfg(test)]
 mod tests;
@@ -42,6 +43,9 @@ use state::State;
 /// policy. These bounds do not account for queues in the caller's transport.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Limits {
+    /// Public Ed25519 key for catalog dependency attestations. Key changes are
+    /// application-policy migrations; private key bytes are never persisted here.
+    pub catalog_key: Option<[u8; 32]>,
     pub max_tables: u64,
     pub max_active_plans: u64,
     pub max_plan_source_bytes: u64,
@@ -54,6 +58,7 @@ pub struct Limits {
 impl Default for Limits {
     fn default() -> Self {
         Self {
+            catalog_key: None,
             max_tables: 1024,
             max_active_plans: 128,
             max_plan_source_bytes: 64 * 1024 * 1024,
@@ -69,6 +74,7 @@ impl Limits {
     fn identity(&self) -> Result<Vec<u8>> {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
+                && self.catalog_key.is_none_or(|key| key != [0; 32])
                 && (1..=256).contains(&self.max_active_plans)
                 && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes)
                 && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
@@ -80,7 +86,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v3",
+            "lance-context/replicated-append/v4",
             self,
         ))?)
     }
@@ -109,6 +115,7 @@ pub enum Applied {
         expected_sequence: u64,
     },
     Capacity,
+    Source(source::Reply),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -170,6 +177,7 @@ struct Group {
 /// or retain storage handles outside it. An internal/disk error fences this node;
 /// drop it, reconcile outstanding requests and reopen the original directories.
 pub struct Node {
+    signer: Option<Arc<source::CatalogSigner>>,
     layout: Arc<Layout>,
     layout_hash: [u8; 32],
     policy: [u8; 32],
@@ -188,6 +196,11 @@ impl Node {
         self.groups
             .into_iter()
             .map(|(id, group)| Self {
+                signer: if id == publication::control_group(&self.layout) {
+                    self.signer.clone()
+                } else {
+                    None
+                },
                 layout: self.layout.clone(),
                 layout_hash: self.layout_hash,
                 policy: self.policy,
@@ -263,6 +276,7 @@ impl Node {
             );
         }
         Ok(Self {
+            signer: None,
             layout,
             layout_hash,
             policy,
@@ -283,6 +297,20 @@ impl Node {
         Ok(())
     }
 
+    /// Install the catalog's private dependency-attestation key. The persisted
+    /// policy pins its public key. Source-only replicas need only the public key.
+    pub fn set_catalog_signer(&mut self, signer: Arc<source::CatalogSigner>) -> Result<()> {
+        self.healthy()?;
+        ensure!(
+            self.groups
+                .contains_key(&publication::control_group(&self.layout))
+                && self.limits.catalog_key == Some(signer.public_key()),
+            "catalog signer does not match hosted scope and policy"
+        );
+        self.signer = Some(signer);
+        Ok(())
+    }
+
     pub fn tick(&mut self) -> Result<()> {
         self.healthy()?;
         for group in self.groups.values_mut() {
@@ -334,6 +362,12 @@ impl Node {
             "request token already pending"
         );
         query.validate(&self.layout, group)?;
+        if matches!(query, query::Query::Pin { .. }) {
+            ensure!(
+                self.signer.is_some(),
+                "catalog dependency signer unavailable"
+            );
+        }
         let g = self.groups.get_mut(&group).context("unhosted group")?;
         ensure!(g.raw.raft.state == StateRole::Leader, "not leader");
         // raft-rs 0.7 silently drops ReadIndex before this condition holds.
@@ -365,6 +399,14 @@ impl Node {
         self.propose(publication::control_group(&self.layout), request, batch)
     }
 
+    /// Verify a catalog quorum-read attestation and persist the exact complete
+    /// table prefix in its source group. Completion, never enqueue, pins it.
+    pub fn pin_source(&mut self, request: u64, permit: source::Permit) -> Result<()> {
+        let group = permit.selection().group;
+        let batch = permit.batch(&self.layout, &self.limits)?;
+        self.propose(group, request, batch)
+    }
+
     /// Local applied view, not a quorum read or proof that an owner is current.
     pub fn publisher(&self, table: Uuid) -> Result<Option<&publication::Table>> {
         self.healthy()?;
@@ -560,6 +602,15 @@ impl Node {
                 },
                 &group.state,
                 group.raw.raft.state == StateRole::Leader,
+                |query, cut, state| {
+                    query.evaluate(
+                        state,
+                        cut,
+                        &self.layout,
+                        &self.limits,
+                        self.signer.as_deref(),
+                    )
+                },
             )?);
             events
                 .messages
diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs
index bf49ac7..bb4abc6 100644
--- a/crates/lance-context-wal/src/replication/checkpoint.rs
+++ b/crates/lance-context-wal/src/replication/checkpoint.rs
@@ -14,8 +14,8 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 3;
-const MAGIC: &[u8; 8] = b"LCWALCP3";
+const FORMAT: u32 = 4;
+const MAGIC: &[u8; 8] = b"LCWALCP4";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -72,6 +72,7 @@ struct Descriptor {
     records: u64,
     producers: u64,
     tables: u64,
+    pins: u64,
     metadata_bytes: u64,
     payload_bytes: u64,
     metadata_digest: [u8; 32],
@@ -84,6 +85,7 @@ fn metadata_limit(limits: &Limits) -> u64 {
     (limits.max_records + limits.max_producers) * 268
         + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4)
         + limits.max_tables.min(limits.max_active_plans) * publication::MAX_COMMAND_BYTES as u64
+        + limits.max_tables.min(limits.max_active_plans) * (source::MAX_PIN_BYTES as u64 + 4)
         + 1024
 }
 
@@ -97,6 +99,7 @@ impl Descriptor {
                 && self.producers <= limits.max_producers
                 && self.producers <= self.records
                 && self.tables <= limits.max_tables
+                && self.pins <= limits.max_tables.min(limits.max_active_plans)
                 && (group == publication::control_group(layout) || self.tables == 0)
                 && self.metadata_bytes <= metadata_limit(limits)
                 && self.payload_bytes <= self.records * limits.max_record_bytes,
@@ -372,12 +375,12 @@ fn read_frame<const N: usize>(input: &mut impl Read) -> Result<([u8; N], usize)>
     input.read_exact(&mut bytes[..length])?;
     Ok((bytes, length))
 }
-fn read_table_frame(input: &mut impl Read) -> Result<Vec<u8>> {
+fn read_dynamic_frame(input: &mut impl Read, maximum: usize) -> Result<Vec<u8>> {
     let mut length = [0; 4];
     input.read_exact(&mut length)?;
     let length = u32::from_le_bytes(length) as usize;
     ensure!(
-        length > 0 && length <= publication::MAX_TABLE_BYTES,
+        length > 0 && length <= maximum,
         "oversized publisher image frame"
     );
     // Plan metadata is larger than a record header. Allocate its actual bounded
@@ -442,6 +445,7 @@ pub(super) fn build(
         records: state.accepted.len() as u64,
         producers: state.producers.len() as u64,
         tables: state.catalog.tables.len() as u64,
+        pins: state.pins.len() as u64,
         metadata_bytes: 0,
         payload_bytes,
         metadata_digest: [0; 32],
@@ -454,6 +458,7 @@ pub(super) fn build(
     write_u64(&mut metadata, descriptor.records)?;
     write_u64(&mut metadata, descriptor.producers)?;
     write_u64(&mut metadata, descriptor.tables)?;
+    write_u64(&mut metadata, descriptor.pins)?;
     for (index, meta) in &state.accepted {
         write_meta(&mut metadata, *index, meta)?;
     }
@@ -463,6 +468,9 @@ pub(super) fn build(
     for table in state.catalog.tables.values() {
         write_frame(&mut metadata, &publication::encode_table(table)?)?;
     }
+    for pin in state.pins.values() {
+        write_frame(&mut metadata, &source::encode_pin(pin)?)?;
+    }
     (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?;
     let mut pack = Writer::new(&paths[1], payload_bytes)?;
     for (index, meta) in &state.accepted {
@@ -509,7 +517,8 @@ pub(super) fn load(
     ensure!(
         read_u64(&mut metadata)? == descriptor.records
             && read_u64(&mut metadata)? == descriptor.producers
-            && read_u64(&mut metadata)? == descriptor.tables,
+            && read_u64(&mut metadata)? == descriptor.tables
+            && read_u64(&mut metadata)? == descriptor.pins,
         "image metadata count mismatch"
     );
     let mut state = State {
@@ -539,7 +548,7 @@ pub(super) fn load(
     }
     let mut previous = None;
     for _ in 0..descriptor.tables {
-        let bytes = read_table_frame(&mut metadata)?;
+        let bytes = read_dynamic_frame(&mut metadata, publication::MAX_TABLE_BYTES)?;
         let table = publication::decode_table(&bytes)?;
         let id = table.authority.table;
         ensure!(
@@ -549,6 +558,18 @@ pub(super) fn load(
         state.catalog.tables.insert(id, table);
         previous = Some(id);
     }
+    let mut previous = None;
+    for _ in 0..descriptor.pins {
+        let bytes = read_dynamic_frame(&mut metadata, source::MAX_PIN_BYTES)?;
+        let pin = source::decode_pin(&bytes)?;
+        let table = pin.permit.selection().table;
+        ensure!(
+            previous.is_none_or(|old| old < table),
+            "unordered or duplicate source pin"
+        );
+        state.pins.insert(table, Arc::new(pin));
+        previous = Some(table);
+    }
     let metadata_file = metadata.finish(descriptor.metadata_bytes, descriptor.metadata_digest)?;
     state.validate_checkpoint(layout, group, limits, descriptor.binding.index)?;
     let (locations, payload_bytes) = locations(&state)?;
diff --git a/crates/lance-context-wal/src/replication/checkpoint/tests.rs b/crates/lance-context-wal/src/replication/checkpoint/tests.rs
index aa912d8..1a35474 100644
--- a/crates/lance-context-wal/src/replication/checkpoint/tests.rs
+++ b/crates/lance-context-wal/src/replication/checkpoint/tests.rs
@@ -396,7 +396,7 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() ->
     let original = fs::read(&path)?;
     // Locate frames from their encoded lengths, without assuming a fixed header.
     let binding_len = u32::from_le_bytes(original[8..12].try_into()?) as usize;
-    let mut offset = 12 + binding_len + 3 * 8;
+    let mut offset = 12 + binding_len + 4 * 8;
     for _ in 0..state.accepted.len() + state.producers.len() {
         offset += 8; // accepted index
         let len = u32::from_le_bytes(original[offset..offset + 4].try_into()?) as usize;
diff --git a/crates/lance-context-wal/src/replication/publication/plan.rs b/crates/lance-context-wal/src/replication/publication/plan.rs
index dfac4f6..f8a713e 100644
--- a/crates/lance-context-wal/src/replication/publication/plan.rs
+++ b/crates/lance-context-wal/src/replication/publication/plan.rs
@@ -14,6 +14,49 @@ pub struct SourceRange {
     pub through: u64,
     pub records: Vec<(u64, BatchMeta)>,
 }
+impl SourceRange {
+    pub(in crate::replication) fn validate(
+        &self,
+        layout: &Layout,
+        table: Uuid,
+        group: u64,
+        limits: &Limits,
+    ) -> Result<()> {
+        ensure!(
+            layout.groups.contains_key(&group)
+                && self.after < self.through
+                && self.through < u64::MAX
+                && !self.records.is_empty()
+                && self.records.len() <= MAX_INPUTS
+                && self.records.windows(2).all(|w| w[0].0 < w[1].0),
+            "invalid publication source range"
+        );
+        let mut bytes = 0u64;
+        for (index, meta) in &self.records {
+            ensure!(
+                *index > self.after
+                    && *index <= self.through
+                    && meta.stream == table
+                    && meta.kind == RecordKind::Data
+                    && !meta.producer.is_nil()
+                    && meta.sequence > 0
+                    && meta.sequence < u64::MAX
+                    && meta.body_bytes > 0
+                    && meta.body_bytes <= limits.max_record_bytes
+                    && layout.group_for_bucket(meta.bucket)? == group,
+                "invalid publication source metadata"
+            );
+            bytes = bytes
+                .checked_add(meta.body_bytes)
+                .context("source byte sum overflow")?;
+        }
+        ensure!(
+            bytes <= limits.max_plan_source_bytes,
+            "publication source byte limit"
+        );
+        Ok(())
+    }
+}
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Plan {
@@ -59,30 +102,8 @@ impl Plan {
             count = count
                 .checked_add(range.records.len())
                 .context("plan input count overflow")?;
-            ensure!(
-                count <= MAX_INPUTS
-                    && layout.groups.contains_key(group)
-                    && range.after < range.through
-                    && range.through < u64::MAX
-                    && !range.records.is_empty()
-                    && range.records.windows(2).all(|w| w[0].0 < w[1].0),
-                "invalid publication source range"
-            );
-            for (index, meta) in &range.records {
-                ensure!(
-                    *index > range.after
-                        && *index <= range.through
-                        && meta.stream == table
-                        && meta.kind == RecordKind::Data
-                        && !meta.producer.is_nil()
-                        && meta.sequence > 0
-                        && meta.sequence < u64::MAX
-                        && meta.body_bytes > 0
-                        && meta.body_bytes <= limits.max_record_bytes
-                        && layout.group_for_bucket(meta.bucket)? == *group,
-                    "invalid publication source metadata"
-                );
-            }
+            ensure!(count <= MAX_INPUTS, "publication input count limit");
+            range.validate(layout, table, *group, limits)?;
         }
         ensure!(
             self.source_bytes()? <= limits.max_plan_source_bytes,
diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs
index e1a5cb1..8c15d26 100644
--- a/crates/lance-context-wal/src/replication/query.rs
+++ b/crates/lance-context-wal/src/replication/query.rs
@@ -20,6 +20,15 @@ pub enum Query {
         bucket: u32,
         producer: Uuid,
     },
+    Pin {
+        table: Uuid,
+        transaction: Uuid,
+        plan_index: u64,
+        source_group: u64,
+    },
+    SourcePin {
+        table: Uuid,
+    },
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -28,6 +37,8 @@ pub enum Value {
     Accepted(Page),
     Producer(Option<(BatchMeta, u64)>),
     CursorAhead { applied: u64 },
+    Pin(Option<Box<source::Permit>>),
+    SourcePin(Option<Arc<source::Pin>>),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -61,6 +72,21 @@ pub struct Completion {
 impl Query {
     pub(super) fn validate(&self, layout: &Layout, group: u64) -> Result<()> {
         match self {
+            Self::Pin {
+                table,
+                transaction,
+                plan_index,
+                source_group,
+            } => ensure!(
+                !table.is_nil()
+                    && !transaction.is_nil()
+                    && *plan_index > 0
+                    && *plan_index < u64::MAX
+                    && layout.groups.contains_key(source_group)
+                    && group == publication::control_group(layout),
+                "invalid catalog dependency query"
+            ),
+            Self::SourcePin { table } => ensure!(!table.is_nil(), "nil source table"),
             Self::Publisher { table } => ensure!(
                 !table.is_nil() && group == publication::control_group(layout),
                 "invalid publisher query scope"
@@ -81,8 +107,45 @@ impl Query {
         Ok(())
     }
 
-    fn evaluate(&self, state: &State) -> Result<Value> {
+    pub(super) fn evaluate(
+        &self,
+        state: &State,
+        cut: &Cut,
+        layout: &Layout,
+        limits: &Limits,
+        signer: Option<&source::CatalogSigner>,
+    ) -> Result<Value> {
         Ok(match self {
+            Self::Pin {
+                table,
+                transaction,
+                plan_index,
+                source_group,
+            } => {
+                let selected = state.catalog.tables.get(table).filter(|table| {
+                    table.progress.pending.as_ref().is_some_and(|p| {
+                        p.index == *plan_index
+                            && p.plan.transaction == *transaction
+                            && p.plan.sources.contains_key(source_group)
+                    })
+                });
+                Value::Pin(
+                    selected
+                        .map(|table| {
+                            source::Permit::issue(
+                                signer.context("catalog dependency signer unavailable")?,
+                                cut.clone(),
+                                layout,
+                                limits,
+                                table,
+                                *source_group,
+                            )
+                            .map(Box::new)
+                        })
+                        .transpose()?,
+                )
+            }
+            Self::SourcePin { table } => Value::SourcePin(state.pins.get(table).cloned()),
             Self::Publisher { table } => {
                 Value::Publisher(state.catalog.tables.get(table).cloned().map(Box::new))
             }
@@ -169,7 +232,13 @@ impl Tracker {
         pending.confirmed = Some(index);
         Ok(())
     }
-    pub fn finish(&mut self, cut: Cut, state: &State, leader: bool) -> Result<Vec<Completion>> {
+    pub fn finish(
+        &mut self,
+        cut: Cut,
+        state: &State,
+        leader: bool,
+        evaluate: impl Fn(&Query, &Cut, &State) -> Result<Value>,
+    ) -> Result<Vec<Completion>> {
         let done: Vec<_> = self
             .pending
             .iter()
@@ -186,12 +255,14 @@ impl Tracker {
             let outcome = if !leader || p.term != cut.term {
                 Outcome::Unknown
             } else {
+                let observed_cut = Cut {
+                    quorum_index: p.confirmed.expect("confirmed completed read"),
+                    ..cut.clone()
+                };
+                let value = evaluate(&p.query, &observed_cut, state)?;
                 Outcome::Observed {
-                    cut: Cut {
-                        quorum_index: p.confirmed.expect("confirmed completed read"),
-                        ..cut.clone()
-                    },
-                    value: p.query.evaluate(state)?,
+                    cut: observed_cut,
+                    value,
                 }
             };
             results.push(Completion {
diff --git a/crates/lance-context-wal/src/replication/query/tests.rs b/crates/lance-context-wal/src/replication/query/tests.rs
index 0b9078e..0464bce 100644
--- a/crates/lance-context-wal/src/replication/query/tests.rs
+++ b/crates/lance-context-wal/src/replication/query/tests.rs
@@ -33,6 +33,15 @@ fn pump(nodes: &mut [Node]) -> Result<Events> {
     }
     anyhow::bail!("query message pump did not settle")
 }
+fn evaluate(query: &Query, cut: &Cut, state: &State) -> Result<Value> {
+    query.evaluate(
+        state,
+        cut,
+        &super::super::tests::layout(),
+        &Limits::default(),
+        None,
+    )
+}
 
 #[test]
 fn new_leader_rejects_reads_until_its_first_current_term_entry_commits() -> Result<()> {
@@ -199,25 +208,34 @@ fn read_contexts_do_not_cross_request_reuse_reopen_or_unapplied_cuts() -> Result
         applied: 4,
     };
     tracker.confirm(&context, 5)?;
-    assert!(tracker.finish(cut.clone(), &state, true)?.is_empty());
+    assert!(tracker
+        .finish(cut.clone(), &state, true, evaluate)?
+        .is_empty());
     state.applied = 5;
     cut.applied = 5;
-    assert_eq!(tracker.finish(cut.clone(), &state, true)?.len(), 1);
+    assert_eq!(
+        tracker.finish(cut.clone(), &state, true, evaluate)?.len(),
+        1
+    );
     let next = tracker.enqueue(10, 2, query.clone())?;
     assert_ne!(next, context);
     tracker.confirm(&context, 5)?;
-    assert!(tracker.finish(cut.clone(), &state, true)?.is_empty());
+    assert!(tracker
+        .finish(cut.clone(), &state, true, evaluate)?
+        .is_empty());
     cut.term = 3;
     assert_eq!(
-        tracker.finish(cut.clone(), &state, true)?[0].outcome,
+        tracker.finish(cut.clone(), &state, true, evaluate)?[0].outcome,
         Outcome::Unknown
     );
     let mut reopened = Tracker::default();
     assert_ne!(reopened.enqueue(10, 3, query)?, context);
     reopened.confirm(&context, 5)?;
-    assert!(reopened.finish(cut.clone(), &state, true)?.is_empty());
+    assert!(reopened
+        .finish(cut.clone(), &state, true, evaluate)?
+        .is_empty());
     assert_eq!(
-        reopened.finish(cut, &state, false)?[0].outcome,
+        reopened.finish(cut, &state, false, evaluate)?[0].outcome,
         Outcome::Unknown
     );
     Ok(())
diff --git a/crates/lance-context-wal/src/replication/source.rs b/crates/lance-context-wal/src/replication/source.rs
new file mode 100644
index 0000000..63e881e
--- /dev/null
+++ b/crates/lance-context-wal/src/replication/source.rs
@@ -0,0 +1,296 @@
+//! Signed catalog observations and durable, complete source-prefix pins.
+//! Attestations are issued only after safe ReadIndex completion. They are not
+//! publisher leases or permission to commit/delete external Lance objects.
+use super::*;
+use crate::{ingest_contract::IngestContract, record::RecordKind};
+use bincode::Options;
+use bytes::Bytes;
+use ring::signature::{Ed25519KeyPair, KeyPair, UnparsedPublicKey, ED25519};
+
+#[cfg(test)]
+mod tests;
+
+pub(super) const MAX_BYTES: usize = 136 * 1024;
+pub(super) const MAX_PIN_BYTES: usize = MAX_BYTES + 8;
+const DOMAIN: &str = "lance-context/catalog-source-pin/v1";
+
+fn codec(limit: usize) -> impl Options {
+    bincode::DefaultOptions::new()
+        .with_fixint_encoding()
+        .with_little_endian()
+        .with_limit(limit as u64)
+        .reject_trailing_bytes()
+}
+
+/// Private key holder for trusted catalog replicas. It has no public signing
+/// operation and is neither serializable nor printable. Provision the same key
+/// to catalog replicas; persist only its public key in `Limits`.
+pub struct CatalogSigner(Ed25519KeyPair);
+impl CatalogSigner {
+    pub fn from_pkcs8(bytes: &[u8]) -> Result<Self> {
+        Ok(Self(Ed25519KeyPair::from_pkcs8(bytes).map_err(|_| {
+            anyhow::anyhow!("invalid catalog signing key")
+        })?))
+    }
+    pub fn public_key(&self) -> [u8; 32] {
+        self.0
+            .public_key()
+            .as_ref()
+            .try_into()
+            .expect("Ed25519 key")
+    }
+}
+
+/// Stable across refreshed observations and publisher takeover. The catalog
+/// index identifies a plan; it is never compared with another group's index.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Selection {
+    pub table: Uuid,
+    pub transaction: Uuid,
+    pub plan_index: u64,
+    pub read_version: u64,
+    pub group: u64,
+    pub ingest: IngestContract,
+    pub range: publication::SourceRange,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Permit {
+    policy: [u8; 32],
+    cut: query::Cut,
+    selection: Selection,
+    // Two fixed arrays preserve a fixed-size serde representation.
+    signature: ([u8; 32], [u8; 32]),
+}
+impl Permit {
+    pub fn selection(&self) -> &Selection {
+        &self.selection
+    }
+    pub fn catalog_cut(&self) -> &query::Cut {
+        &self.cut
+    }
+    fn signed_bytes(&self) -> Result<Vec<u8>> {
+        Ok(codec(MAX_BYTES).serialize(&(DOMAIN, self.policy, &self.cut, &self.selection))?)
+    }
+    pub(super) fn issue(
+        signer: &CatalogSigner,
+        cut: query::Cut,
+        layout: &Layout,
+        limits: &Limits,
+        table: &publication::Table,
+        group: u64,
+    ) -> Result<Self> {
+        ensure!(
+            limits.catalog_key == Some(signer.public_key()),
+            "catalog signing policy mismatch"
+        );
+        let pending = table.progress.pending.as_ref().context("no pending plan")?;
+        let range = pending
+            .plan
+            .sources
+            .get(&group)
+            .context("source not selected")?
+            .clone();
+        let mut permit = Self {
+            policy: Sha256::digest(limits.identity()?).into(),
+            cut,
+            selection: Selection {
+                table: table.authority.table,
+                transaction: pending.plan.transaction,
+                plan_index: pending.index,
+                read_version: pending.plan.read_version,
+                group,
+                ingest: table.ingest.clone(),
+                range,
+            },
+            signature: ([0; 32], [0; 32]),
+        };
+        let signature = signer.0.sign(&permit.signed_bytes()?);
+        permit.signature = (
+            signature.as_ref()[..32].try_into()?,
+            signature.as_ref()[32..].try_into()?,
+        );
+        permit.verify(layout, limits, group)?;
+        Ok(permit)
+    }
+
+    fn verify(&self, layout: &Layout, limits: &Limits, group: u64) -> Result<()> {
+        let key = limits
+            .catalog_key
+            .context("source pins require a catalog verification key")?;
+        ensure!(
+            self.policy == <[u8; 32]>::from(Sha256::digest(limits.identity()?)),
+            "foreign pin policy"
+        );
+        let mut signature = [0; 64];
+        signature[..32].copy_from_slice(&self.signature.0);
+        signature[32..].copy_from_slice(&self.signature.1);
+        UnparsedPublicKey::new(&ED25519, key)
+            .verify(&self.signed_bytes()?, &signature)
+            .map_err(|_| anyhow::anyhow!("invalid catalog pin attestation"))?;
+        let s = &self.selection;
+        ensure!(
+            self.cut.cluster == layout.cluster
+                && self.cut.layout == layout.fingerprint()?
+                && self.cut.group == publication::control_group(layout)
+                && self.cut.term > 0
+                && self.cut.quorum_index > 0
+                && self.cut.quorum_index <= self.cut.applied
+                && self.cut.applied < u64::MAX
+                && s.plan_index > 0
+                && s.plan_index <= self.cut.applied
+                && !s.table.is_nil()
+                && !s.transaction.is_nil()
+                && s.group == group
+                && s.read_version > 0
+                && s.read_version < u64::MAX,
+            "invalid catalog pin scope or cut"
+        );
+        s.ingest.validate()?;
+        ensure!(
+            u64::from(s.ingest.max_record_bytes) <= limits.max_record_bytes
+                && u64::from(s.ingest.max_record_bytes) <= limits.max_plan_source_bytes,
+            "unserviceable pin contract"
+        );
+        s.range.validate(layout, s.table, group, limits)?;
+        ensure!(
+            s.range
+                .records
+                .iter()
+                .all(|(_, m)| m.body_bytes <= u64::from(s.ingest.max_record_bytes))
+                && (group != self.cut.group || s.range.through < s.plan_index),
+            "pin source exceeds its contract or catalog plan"
+        );
+        Ok(())
+    }
+
+    pub(super) fn batch(self, layout: &Layout, limits: &Limits) -> Result<Batch> {
+        self.verify(layout, limits, self.selection.group)?;
+        let bytes = codec(MAX_BYTES).serialize(&self)?;
+        ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command");
+        let mut batch = Batch::new(
+            self.selection.table,
+            0,
+            self.selection.transaction,
+            self.selection.plan_index,
+            Bytes::from(bytes),
+        );
+        batch.meta.kind = RecordKind::Source;
+        Ok(batch)
+    }
+}
+
+pub(super) fn decode(bytes: &[u8]) -> Result<Permit> {
+    ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command");
+    let permit = codec(MAX_BYTES).deserialize(bytes)?;
+    ensure!(
+        codec(MAX_BYTES).serialize(&permit)? == bytes,
+        "noncanonical source pin"
+    );
+    Ok(permit)
+}
+pub(super) fn validate_record(
+    layout: &Layout,
+    group: u64,
+    limits: &Limits,
+    meta: &BatchMeta,
+    body: &[u8],
+) -> Result<()> {
+    let permit = decode(body)?;
+    permit.verify(layout, limits, group)?;
+    let s = permit.selection();
+    ensure!(
+        meta.kind == RecordKind::Source
+            && meta.stream == s.table
+            && meta.producer == s.transaction
+            && meta.sequence == s.plan_index
+            && meta.bucket == 0,
+        "source pin envelope mismatch"
+    );
+    Ok(())
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Pin {
+    pub index: u64,
+    pub permit: Permit,
+}
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum Rejection {
+    Pending,
+    Prefix,
+    Capacity,
+}
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum Reply {
+    Pinned { index: u64 },
+    Rejected(Rejection),
+}
+
+fn complete_prefix(accepted: &BTreeMap<u64, BatchMeta>, selection: &Selection, index: u64) -> bool {
+    let range = &selection.range;
+    // No release/cursor advancement exists in this increment. The retained
+    // source prefix starts at zero until durable resolution is integrated.
+    range.after == 0
+        && range.through < index
+        && accepted
+            .range((
+                std::ops::Bound::Excluded(range.after),
+                std::ops::Bound::Included(range.through),
+            ))
+            .filter(|(_, meta)| meta.stream == selection.table)
+            .map(|(i, meta)| (*i, meta))
+            .eq(range.records.iter().map(|(i, meta)| (*i, meta)))
+}
+pub(super) fn apply(state: &mut State, permit: Permit, index: u64, limits: &Limits) -> Reply {
+    let selection = permit.selection();
+    if let Some(old) = state.pins.get(&selection.table) {
+        return if old.permit.selection() == selection {
+            Reply::Pinned { index: old.index }
+        } else {
+            Reply::Rejected(Rejection::Pending)
+        };
+    }
+    if !complete_prefix(&state.accepted, selection, index) {
+        return Reply::Rejected(Rejection::Prefix);
+    }
+    if state.pins.len() as u64 >= limits.max_tables.min(limits.max_active_plans) {
+        return Reply::Rejected(Rejection::Capacity);
+    }
+    state
+        .pins
+        .insert(selection.table, Arc::new(Pin { index, permit }));
+    Reply::Pinned { index }
+}
+pub(super) fn validate_state(
+    state: &State,
+    layout: &Layout,
+    group: u64,
+    cut: u64,
+    limits: &Limits,
+) -> Result<()> {
+    ensure!(
+        state.pins.len() as u64 <= limits.max_tables.min(limits.max_active_plans),
+        "source pin capacity"
+    );
+    for (table, pin) in &state.pins {
+        pin.permit.verify(layout, limits, group)?;
+        ensure!(
+            *table == pin.permit.selection.table
+                && pin.index > 0
+                && pin.index <= cut
+                && complete_prefix(&state.accepted, pin.permit.selection(), pin.index),
+            "invalid checkpoint source pin or prefix"
+        );
+    }
+    Ok(())
+}
+pub(super) fn encode_pin(pin: &Pin) -> Result<Vec<u8>> {
+    Ok(codec(MAX_PIN_BYTES).serialize(pin)?)
+}
+pub(super) fn decode_pin(bytes: &[u8]) -> Result<Pin> {
+    ensure!(bytes.len() <= MAX_PIN_BYTES, "oversized source pin frame");
+    let pin = codec(MAX_PIN_BYTES).deserialize(bytes)?;
+    ensure!(encode_pin(&pin)? == bytes, "noncanonical source pin frame");
+    Ok(pin)
+}
diff --git a/crates/lance-context-wal/src/replication/source/tests.rs b/crates/lance-context-wal/src/replication/source/tests.rs
new file mode 100644
index 0000000..95a671f
--- /dev/null
+++ b/crates/lance-context-wal/src/replication/source/tests.rs
@@ -0,0 +1,470 @@
+use super::*;
+use crate::ingest_contract::WritePolicy;
+use publication::{Acquire, Authority, Destination, Plan, PlanRequest, SourceRange};
+use query::{Outcome as ReadOutcome, Query, Value};
+
+struct Cluster {
+    nodes: Vec<Node>,
+    root: tempfile::TempDir,
+    layout: Arc<Layout>,
+    limits: Limits,
+    signer: Arc<CatalogSigner>,
+    table: Uuid,
+    transaction: Uuid,
+}
+impl Cluster {
+    fn new() -> Result<Self> {
+        Self::with_plan_limit(128)
+    }
+    fn with_plan_limit(max_active_plans: u64) -> Result<Self> {
+        let root = tempfile::tempdir()?;
+        let layout = Arc::new(Layout {
+            cluster: Uuid::from_u128(91),
+            nodes: (1..=6)
+                .map(|i| (i, format!("wal-{i}.example.invalid:7800")))
+                .collect(),
+            groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![4, 5, 6])]),
+            buckets: vec![10, 20],
+        });
+        let key = Ed25519KeyPair::generate_pkcs8(&ring::rand::SystemRandom::new())
+            .map_err(|_| anyhow::anyhow!("test key generation"))?;
+        let signer = Arc::new(CatalogSigner::from_pkcs8(key.as_ref())?);
+        let limits = Limits {
+            catalog_key: Some(signer.public_key()),
+            max_active_plans,
+            ..Default::default()
+        };
+        let mut c = Self {
+            nodes: vec![],
+            root,
+            layout,
+            limits,
+            signer,
+            table: Uuid::from_u128(92),
+            transaction: Uuid::from_u128(93),
+        };
+        c.reopen()?;
+        Ok(c)
+    }
+    fn reopen(&mut self) -> Result<()> {
+        self.nodes.clear();
+        for id in 1..=6 {
+            let mut node = Node::open(
+                &self.root.path().join(id.to_string()),
+                self.layout.clone(),
+                id,
+                self.limits.clone(),
+            )?;
+            if id <= 3 {
+                node.set_catalog_signer(self.signer.clone())?;
+            }
+            self.nodes.push(node);
+        }
+        self.nodes[0].campaign(10)?;
+        self.nodes[3].campaign(20)?;
+        self.pump()?;
+        Ok(())
+    }
+    fn pump(&mut self) -> Result<Events> {
+        let mut all = Events::default();
+        for _ in 0..200 {
+            let mut packets = vec![];
+            for node in &mut self.nodes {
+                let events = node.poll()?;
+                all.completions.extend(events.completions);
+                all.reads.extend(events.reads);
+                packets.extend(events.messages);
+            }
+            if packets.is_empty() {
+                return Ok(all);
+            }
+            for packet in packets {
+                self.nodes[packet.message.to as usize - 1].step(packet)?;
+            }
+        }
+        anyhow::bail!("pin test pump did not settle")
+    }
+    fn append(&mut self, table: Uuid, sequence: u64) -> Result<()> {
+        self.nodes[3].propose(
+            20,
+            sequence,
+            Batch::new(
+                table,
+                1,
+                Uuid::from_u128(94),
+                sequence,
+                Bytes::from(vec![sequence as u8; 4096]),
+            ),
+        )?;
+        assert!(self
+            .pump()?
+            .completions
+            .iter()
+            .any(|c| matches!(c.outcome, Outcome::Committed(Applied::Accepted { .. }))));
+        Ok(())
+    }
+    fn acquire(&mut self, expected_epoch: u64) -> Result<Authority> {
+        self.nodes[0].acquire_publisher(
+            100,
+            Acquire {
+                table: self.table,
+                destination: Destination {
+                    namespace: Uuid::from_u128(95),
+                    dataset: "tables/synthetic".into(),
+                },
+                ingest: IngestContract {
+                    tenant: "test".into(),
+                    qualified_table: "synthetic".into(),
+                    schema: [7; 32],
+                    policy: WritePolicy::ImmutableUniqueIds,
+                    max_record_bytes: 1024 * 1024,
+                    max_rows: 1024,
+                },
+                owner: Uuid::from_u128(100 + u128::from(expected_epoch)),
+                request: Uuid::from_u128(200 + u128::from(expected_epoch)),
+                expected_epoch,
+            },
+        )?;
+        let events = self.pump()?;
+        events
+            .completions
+            .into_iter()
+            .find_map(|c| match c.outcome {
+                Outcome::Committed(Applied::Publisher(publication::Reply::Acquired(a))) => Some(a),
+                _ => None,
+            })
+            .context("missing publisher completion")
+    }
+    fn plan(&mut self, mutate: impl FnOnce(&mut SourceRange)) -> Result<u64> {
+        for seq in 1..=12 {
+            self.append(Uuid::from_u128(999), seq)?;
+        }
+        self.append(self.table, 1)?;
+        self.append(self.table, 2)?;
+        self.nodes[3].query(
+            20,
+            101,
+            Query::Accepted {
+                after: 0,
+                limit: 128,
+            },
+        )?;
+        let reads = self.pump()?.reads;
+        let ReadOutcome::Observed {
+            value: Value::Accepted(page),
+            ..
+        } = &reads[0].outcome
+        else {
+            anyhow::bail!("source query failed");
+        };
+        let mut range = SourceRange {
+            after: 0,
+            through: page.through,
+            records: page
+                .entries
+                .iter()
+                .filter(|(_, m)| m.stream == self.table)
+                .cloned()
+                .collect(),
+        };
+        mutate(&mut range);
+        let authority = self.acquire(0)?;
+        self.nodes[0].plan_publication(
+            102,
+            PlanRequest {
+                authority,
+                plan: Plan {
+                    transaction: self.transaction,
+                    read_version: 8,
+                    sources: BTreeMap::from([(20, range)]),
+                    next_group: 10,
+                },
+            },
+        )?;
+        self.pump()?
+            .completions
+            .into_iter()
+            .find_map(|c| match c.outcome {
+                Outcome::Committed(Applied::Publisher(publication::Reply::Planned { index })) => {
+                    Some(index)
+                }
+                _ => None,
+            })
+            .context("missing plan completion")
+    }
+    fn permit(&mut self, index: u64) -> Result<Permit> {
+        self.nodes[0].query(
+            10,
+            103,
+            Query::Pin {
+                table: self.table,
+                transaction: self.transaction,
+                plan_index: index,
+                source_group: 20,
+            },
+        )?;
+        let events = self.pump()?;
+        let permit = events
+            .reads
+            .into_iter()
+            .find_map(|read| match read.outcome {
+                ReadOutcome::Observed {
+                    value: Value::Pin(Some(permit)),
+                    ..
+                } => Some(*permit),
+                _ => None,
+            })
+            .context("missing catalog attestation")?;
+        // Exercise the serialized boundary: the receiving source has no private key.
+        Ok(bincode::deserialize(&bincode::serialize(&permit)?)?)
+    }
+    fn pin(&mut self, permit: Permit) -> Result<Reply> {
+        self.nodes[3].pin_source(104, permit)?;
+        self.pump()?
+            .completions
+            .into_iter()
+            .find_map(|c| match c.outcome {
+                Outcome::Committed(Applied::Source(reply)) => Some(reply),
+                _ => None,
+            })
+            .context("missing source completion")
+    }
+}
+
+#[test]
+fn disjoint_groups_pin_their_full_prefix_and_recover_original_index() -> Result<()> {
+    let mut c = Cluster::new()?;
+    let plan = c.plan(|_| {})?;
+    let permit = c.permit(plan)?;
+    assert!(
+        permit.selection.range.through > permit.cut.applied,
+        "test did not separate group indices"
+    );
+    assert!(c.nodes[3].signer.is_none());
+    assert!(c.nodes[3].set_catalog_signer(c.signer.clone()).is_err());
+    let result = c.pin(permit.clone())?;
+    let Reply::Pinned { index } = result else {
+        anyhow::bail!("pin rejected");
+    };
+    assert!(index > permit.selection.range.through);
+    c.acquire(1)?;
+    let refreshed = c.permit(plan)?;
+    assert_ne!(permit.cut, refreshed.cut);
+    assert_eq!(c.pin(refreshed)?, result);
+    c.append(c.table, 3)?; // Suffix stays outside the immutable pin.
+    for n in 0..6 {
+        let group = if n < 3 { 10 } else { 20 };
+        let image = c.nodes[n]
+            .prepare_checkpoint(group, Uuid::from_u128(300 + n as u128))?
+            .run()?;
+        c.nodes[n].install_checkpoint(image)?;
+    }
+    c.reopen()?;
+    let refreshed = c.permit(plan)?;
+    assert_eq!(c.pin(refreshed)?, result);
+    for node in &c.nodes[3..] {
+        let state = &node.groups[&20].state;
+        let pin = &state.pins[&c.table];
+        assert_eq!(pin.index, index);
+        for (i, meta) in &pin.permit.selection.range.records {
+            let row = node.read_accepted(20, *i)?.unwrap();
+            assert_eq!(row.meta, *meta);
+            assert_eq!(row.body, Bytes::from(vec![meta.sequence as u8; 4096]));
+        }
+        let captured = state.clone();
+        assert!(Arc::ptr_eq(&captured.pins[&c.table], pin));
+    }
+    Ok(())
+}
+
+#[test]
+fn signed_catalog_selection_cannot_omit_or_change_the_source_prefix() -> Result<()> {
+    for mutation in 0..4 {
+        let mut c = Cluster::new()?;
+        let plan = c.plan(|range| match mutation {
+            0 => {
+                range.records.remove(0);
+            }
+            1 => {
+                range.records[0].1.digest[0] ^= 1;
+            }
+            2 => {
+                range.through += 10_000;
+            }
+            _ => {
+                range.records[0].0 -= 1;
+            }
+        })?;
+        let permit = c.permit(plan)?;
+        assert_eq!(c.pin(permit)?, Reply::Rejected(Rejection::Prefix));
+        assert!(c.nodes[3].groups[&20].state.pins.is_empty());
+    }
+    Ok(())
+}
+
+#[test]
+fn altered_attestations_and_foreign_keys_never_enqueue_a_pin() -> Result<()> {
+    let mut c = Cluster::new()?;
+    let plan = c.plan(|_| {})?;
+    let permit = c.permit(plan)?;
+    let before = c.nodes[3].groups[&20].raw.raft.raft_log.last_index();
+    let mut altered = permit.clone();
+    altered.selection.range.records.remove(0);
+    assert!(c.nodes[3].pin_source(104, altered).is_err());
+    for mutate in [
+        (|p: &mut Permit| p.cut.cluster = Uuid::from_u128(900)) as fn(&mut Permit),
+        |p| p.selection.plan_index += 1,
+        |p| p.signature.0[0] ^= 1,
+        |p| p.selection.ingest.schema[0] ^= 1,
+    ] {
+        let mut altered = permit.clone();
+        mutate(&mut altered);
+        assert!(c.nodes[3].pin_source(104, altered).is_err());
+    }
+    let mut foreign = c.limits.clone();
+    foreign.catalog_key = Some([42; 32]);
+    assert!(permit.verify(&c.layout, &foreign, 20).is_err());
+    assert_eq!(
+        c.nodes[3].groups[&20].raw.raft.raft_log.last_index(),
+        before
+    );
+    assert_eq!(c.nodes[3].status(20)?.pending, 0);
+    assert!(matches!(c.pin(permit)?, Reply::Pinned { .. }));
+    Ok(())
+}
+
+#[test]
+fn an_isolated_catalog_cannot_issue_a_new_attestation() -> Result<()> {
+    let mut c = Cluster::new()?;
+    let plan = c.plan(|_| {})?;
+    c.nodes[0].query(
+        10,
+        105,
+        Query::Pin {
+            table: c.table,
+            transaction: c.transaction,
+            plan_index: plan,
+            source_group: 20,
+        },
+    )?;
+    let mut unknown = false;
+    for _ in 0..40 {
+        c.nodes[0].tick()?;
+        for read in c.nodes[0].poll()?.reads {
+            assert!(matches!(read.outcome, ReadOutcome::Unknown));
+            unknown = true;
+        }
+    }
+    assert!(unknown);
+    Ok(())
+}
+
+#[test]
+fn catalog_can_also_be_a_source_and_retry_a_pin_at_capacity() -> Result<()> {
+    let mut c = Cluster::with_plan_limit(1)?;
+    let authority = c.acquire(0)?;
+    let row = Batch::new(
+        c.table,
+        0,
+        Uuid::from_u128(401),
+        1,
+        Bytes::from(vec![17; 8192]),
+    );
+    c.nodes[0].propose(10, 400, row.clone())?;
+    let index = c
+        .pump()?
+        .completions
+        .into_iter()
+        .find_map(|completion| match completion.outcome {
+            Outcome::Committed(Applied::Accepted { index }) => Some(index),
+            _ => None,
+        })
+        .context("missing catalog source append")?;
+    c.nodes[0].plan_publication(
+        401,
+        PlanRequest {
+            authority,
+            plan: Plan {
+                transaction: c.transaction,
+                read_version: 3,
+                next_group: 20,
+                sources: BTreeMap::from([(
+                    10,
+                    SourceRange {
+                        after: 0,
+                        through: index,
+                        records: vec![(index, row.meta.clone())],
+                    },
+                )]),
+            },
+        },
+    )?;
+    let plan_index = c
+        .pump()?
+        .completions
+        .into_iter()
+        .find_map(|completion| match completion.outcome {
+            Outcome::Committed(Applied::Publisher(publication::Reply::Planned { index })) => {
+                Some(index)
+            }
+            _ => None,
+        })
+        .context("missing catalog-source plan")?;
+    let mut original = None;
+    for attempt in 0..2 {
+        c.nodes[0].query(
+            10,
+            402,
+            Query::Pin {
+                table: c.table,
+                transaction: c.transaction,
+                plan_index,
+                source_group: 10,
+            },
+        )?;
+        let permit = c
+            .pump()?
+            .reads
+            .into_iter()
+            .find_map(|read| match read.outcome {
+                ReadOutcome::Observed {
+                    value: Value::Pin(Some(p)),
+                    ..
+                } => Some(*p),
+                _ => None,
+            })
+            .context("missing same-group attestation")?;
+        c.nodes[0].pin_source(403, permit)?;
+        let pinned = c
+            .pump()?
+            .completions
+            .into_iter()
+            .find_map(|completion| match completion.outcome {
+                Outcome::Committed(Applied::Source(Reply::Pinned { index })) => Some(index),
+                _ => None,
+            })
+            .context("missing same-group pin")?;
+        if attempt == 0 {
+            original = Some(pinned);
+        } else {
+            assert_eq!(Some(pinned), original);
+        }
+        assert_eq!(c.nodes[0].groups[&10].state.pins.len(), 1);
+    }
+    let state = &c.nodes[0].groups[&10].state;
+    state.validate_checkpoint(&c.layout, 10, &c.limits, state.applied)?;
+    let mut corrupt = state.clone();
+    Arc::make_mut(corrupt.pins.get_mut(&c.table).unwrap())
+        .permit
+        .signature
+        .1[0] ^= 1;
+    assert!(corrupt
+        .validate_checkpoint(&c.layout, 10, &c.limits, corrupt.applied)
+        .is_err());
+    let mut wrong_cut = state.clone();
+    Arc::make_mut(wrong_cut.pins.get_mut(&c.table).unwrap()).index = index;
+    assert!(wrong_cut
+        .validate_checkpoint(&c.layout, 10, &c.limits, wrong_cut.applied)
+        .is_err());
+    Ok(())
+}
diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs
index 8d9d9c8..427e3a7 100644
--- a/crates/lance-context-wal/src/replication/state.rs
+++ b/crates/lance-context-wal/src/replication/state.rs
@@ -25,6 +25,10 @@ pub(super) fn validate_meta(
     body: &[u8],
     digest: bool,
 ) -> Result<()> {
+    if meta.kind == RecordKind::Source {
+        meta.validate_payload(body, source::MAX_BYTES)?;
+        return source::validate_record(layout, group, limits, meta, body);
+    }
     if meta.kind == RecordKind::MultiPublication {
         meta.validate_payload(body, publication::MAX_COMMAND_BYTES)?;
         return publication::validate_record(layout, group, limits, meta, body);
@@ -79,6 +83,7 @@ pub(super) fn validate_entry(
 
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub(super) struct State {
+    pub pins: BTreeMap<Uuid, Arc<source::Pin>>,
     pub catalog: publication::Catalog,
     pub applied: u64,
     pub accepted: BTreeMap<u64, BatchMeta>,
@@ -152,6 +157,7 @@ impl State {
             "checkpoint producer tails disagree"
         );
         self.catalog.validate(layout, group, cut, limits)?;
+        source::validate_state(self, layout, group, cut, limits)?;
         Ok(())
     }
 
@@ -165,6 +171,14 @@ impl State {
             return Ok(None);
         }
         let meta = decode_meta(&entry.context)?;
+        if meta.kind == RecordKind::Source {
+            return Ok(Some(Applied::Source(source::apply(
+                self,
+                source::decode(&entry.data)?,
+                entry.index,
+                limits,
+            ))));
+        }
         if meta.kind == RecordKind::MultiPublication {
             return Ok(Some(Applied::Publisher(self.catalog.apply(
                 publication::decode(&entry.data)?,
diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs
index 705d121..f9017e5 100644
--- a/crates/lance-context-wal/tests/replication_process.rs
+++ b/crates/lance-context-wal/tests/replication_process.rs
@@ -13,6 +13,7 @@ use lance_context_wal::{
             Reply as PublisherReply, SourceRange, Table,
         },
         query::{Completion as ReadCompletion, Outcome as ReadOutcome, Query, Value},
+        source::{CatalogSigner, Permit, Reply as SourceReply},
         Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, PreparedCheckpoint,
         SnapshotReceiver, Status,
     },
@@ -61,6 +62,7 @@ impl WirePacket {
 }
 #[derive(Serialize, Deserialize)]
 enum Request {
+    Pin(u64, Permit),
     Query(u64, u64, Query),
     Plan(u64, PlanRequest),
     Acquire(u64, Acquire),
@@ -127,6 +129,7 @@ fn part(payload: bool) -> Part {
 fn run(worker: &mut Worker, request: Request) -> Result<Response> {
     let node = &mut worker.node;
     match request {
+        Request::Pin(token, permit) => node.pin_source(token, permit)?,
         Request::Query(group, token, query) => node.query(group, token, query)?,
         Request::Plan(token, plan) => node.plan_publication(token, plan)?,
         Request::Acquire(token, acquire) => node.acquire_publisher(token, acquire)?,
@@ -234,14 +237,26 @@ fn replication_peer_fixture() -> Result<()> {
     let root = PathBuf::from(root);
     let id = std::env::var("CONTEXT_REPLICATION_NODE")?.parse()?;
     let listener = TcpListener::bind("127.0.0.1:0")?;
-    let node = Node::open(
+    let limits_path = root.join("limits.json");
+    let limits = if limits_path.exists() {
+        serde_json::from_slice(&std::fs::read(limits_path)?)?
+    } else {
+        Limits::default()
+    };
+    let mut node = Node::open(
         &root.join(format!("data-{id}")),
         Arc::new(serde_json::from_slice(&std::fs::read(
             root.join("layout.json"),
         )?)?),
         id,
-        Limits::default(),
+        limits,
     )?;
+    let key_path = root.join(format!("catalog-key-{id}.pkcs8"));
+    if key_path.exists() {
+        node.set_catalog_signer(Arc::new(CatalogSigner::from_pkcs8(&std::fs::read(
+            key_path,
+        )?)?))?;
+    }
     let mut worker = Worker {
         node,
         prepared: None,
@@ -338,7 +353,6 @@ struct Cluster {
 }
 impl Cluster {
     fn new() -> Result<Self> {
-        let root = tempfile::tempdir()?;
         let layout = Layout {
             cluster: Uuid::from_u128(50),
             nodes: (1..=3)
@@ -347,6 +361,23 @@ impl Cluster {
             groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]),
             buckets: vec![10, 20],
         };
+        Self::with_layout(layout, None)
+    }
+    fn with_layout(layout: Layout, catalog_key: Option<&[u8]>) -> Result<Self> {
+        let root = tempfile::tempdir()?;
+        if let Some(key) = catalog_key {
+            let signer = CatalogSigner::from_pkcs8(key)?;
+            std::fs::write(
+                root.path().join("limits.json"),
+                serde_json::to_vec(&Limits {
+                    catalog_key: Some(signer.public_key()),
+                    ..Default::default()
+                })?,
+            )?;
+            for id in &layout.groups[layout.groups.first_key_value().unwrap().0] {
+                std::fs::write(root.path().join(format!("catalog-key-{id}.pkcs8")), key)?;
+            }
+        }
         std::fs::write(
             root.path().join("layout.json"),
             serde_json::to_vec(&layout)?,
@@ -361,8 +392,8 @@ impl Cluster {
             transfers: 0,
             interrupt_next_snapshot: false,
         };
-        for id in 1..=3 {
-            this.start(id)?;
+        for id in layout.nodes.keys() {
+            this.start(*id)?;
         }
         Ok(this)
     }
@@ -383,6 +414,7 @@ impl Cluster {
             Request::Query(group, token, _) => format!("query group {group}, token {token}"),
             Request::Acquire(token, _) => format!("acquire token {token}"),
             Request::Plan(token, _) => format!("plan token {token}"),
+            Request::Pin(token, _) => format!("pin token {token}"),
             _ => "replication control".to_string(),
         };
         let Response::Events(packets, completions, reads) = self
@@ -1174,3 +1206,236 @@ fn quorum_queries_capture_committed_tails_without_appends_and_reject_isolated_re
     }
     Ok(())
 }
+
+#[test]
+fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> Result<()> {
+    use lance_context_wal::ingest_contract::{IngestContract, WritePolicy};
+    let key = ring::signature::Ed25519KeyPair::generate_pkcs8(&ring::rand::SystemRandom::new())
+        .map_err(|_| anyhow::anyhow!("test key generation failed"))?;
+    let layout = Layout {
+        cluster: Uuid::from_u128(810),
+        nodes: (1..=6)
+            .map(|id| (id, format!("wal-{id}.example.invalid:7800")))
+            .collect(),
+        groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![4, 5, 6])]),
+        buckets: vec![10, 20],
+    };
+    let mut c = Cluster::with_layout(layout, Some(key.as_ref()))?;
+    let catalog = c.elect(1, 10, &[1, 2, 3])?;
+    let source = c.elect(4, 20, &[4, 5, 6])?;
+    assert_eq!(source, 4);
+    drop(c.peers.remove(&6));
+    let table = Uuid::from_u128(811);
+    let rows: Vec<_> = (1..=3)
+        .map(|sequence| {
+            Batch::new(
+                table,
+                1,
+                Uuid::from_u128(812),
+                sequence,
+                Bytes::from(vec![sequence as u8; 64 * 1024]),
+            )
+        })
+        .collect();
+    for (n, row) in rows.iter().enumerate() {
+        c.command(source, Request::Propose(20, 1000 + n as u64, row.clone()))?;
+        c.pump(&[4, 5])?;
+        assert!(matches!(
+            c.outcome(1000 + n as u64),
+            Some(Outcome::Committed(Applied::Accepted { .. }))
+        ));
+    }
+    c.command(
+        source,
+        Request::Query(
+            20,
+            1010,
+            Query::Accepted {
+                after: 0,
+                limit: 128,
+            },
+        ),
+    )?;
+    c.pump(&[4, 5])?;
+    let Some(ReadOutcome::Observed {
+        value: Value::Accepted(page),
+        ..
+    }) = c.read_outcome(1010)
+    else {
+        anyhow::bail!("source quorum page missing");
+    };
+    let mut acquire = Acquire {
+        table,
+        destination: Destination {
+            namespace: Uuid::from_u128(813),
+            dataset: "tables/pinned".into(),
+        },
+        ingest: IngestContract {
+            tenant: "test".into(),
+            qualified_table: "pinned".into(),
+            schema: [3; 32],
+            policy: WritePolicy::ImmutableUniqueIds,
+            max_record_bytes: 1024 * 1024,
+            max_rows: 1024,
+        },
+        owner: Uuid::from_u128(814),
+        request: Uuid::from_u128(815),
+        expected_epoch: 0,
+    };
+    c.command(catalog, Request::Acquire(1020, acquire.clone()))?;
+    c.pump(&[1, 2, 3])?;
+    let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(authority)))) =
+        c.outcome(1020)
+    else {
+        anyhow::bail!("catalog authority missing");
+    };
+    let transaction = Uuid::from_u128(816);
+    c.command(
+        catalog,
+        Request::Plan(
+            1021,
+            PlanRequest {
+                authority,
+                plan: Plan {
+                    transaction,
+                    read_version: 7,
+                    next_group: 10,
+                    sources: BTreeMap::from([(
+                        20,
+                        SourceRange {
+                            after: 0,
+                            through: page.through,
+                            records: page.entries,
+                        },
+                    )]),
+                },
+            },
+        ),
+    )?;
+    c.pump(&[1, 2, 3])?;
+    let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Planned { index: plan_index }))) =
+        c.outcome(1021)
+    else {
+        anyhow::bail!("catalog plan missing");
+    };
+    let query = Query::Pin {
+        table,
+        transaction,
+        plan_index,
+        source_group: 20,
+    };
+    c.command(catalog, Request::Query(10, 1022, query.clone()))?;
+    assert!(
+        c.read_outcome(1022).is_none(),
+        "attestation preceded quorum confirmation"
+    );
+    c.pump(&[1, 2, 3])?;
+    let Some(ReadOutcome::Observed {
+        value: Value::Pin(Some(permit)),
+        ..
+    }) = c.read_outcome(1022)
+    else {
+        anyhow::bail!("catalog attestation missing");
+    };
+    // The attestation crosses two real TCP connections and a process boundary.
+    // Source replicas have only the policy's public verification key.
+    c.command(source, Request::Pin(1030, *permit))?;
+    assert!(
+        c.outcome(1030).is_none(),
+        "pin preceded durable source quorum"
+    );
+    c.pump(&[4, 5])?;
+    let Some(Outcome::Committed(Applied::Source(SourceReply::Pinned { index }))) = c.outcome(1030)
+    else {
+        anyhow::bail!("source pin missing");
+    };
+    c.peers
+        .get_mut(&4)
+        .unwrap()
+        .call(Request::PrepareCheckpoint(20, Uuid::from_u128(817)))?;
+    c.command(4, Request::InstallCheckpoint)?;
+    c.pump(&[4, 5])?;
+    c.start(6)?;
+    for _ in 0..40 {
+        c.command(4, Request::Tick)?;
+        c.pump(&[4, 5, 6])?;
+        if c.transfers > 0 {
+            break;
+        }
+    }
+    assert!(
+        c.transfers > 0,
+        "source pin never crossed a snapshot transfer"
+    );
+    drop(c.peers.remove(&4));
+    drop(c.peers.remove(&1));
+    let source = c.elect(6, 20, &[5, 6])?;
+    let catalog = c.elect(2, 10, &[2, 3])?;
+    acquire.owner = Uuid::from_u128(818);
+    acquire.request = Uuid::from_u128(819);
+    acquire.expected_epoch = 1;
+    c.command(catalog, Request::Acquire(1040, acquire))?;
+    c.pump(&[2, 3])?;
+    assert!(matches!(
+        c.outcome(1040),
+        Some(Outcome::Committed(Applied::Publisher(
+            PublisherReply::Acquired(_)
+        )))
+    ));
+    c.command(catalog, Request::Query(10, 1041, query.clone()))?;
+    c.pump(&[2, 3])?;
+    let Some(ReadOutcome::Observed {
+        value: Value::Pin(Some(permit)),
+        ..
+    }) = c.read_outcome(1041)
+    else {
+        anyhow::bail!("takeover attestation missing");
+    };
+    c.command(source, Request::Pin(1042, *permit))?;
+    c.pump(&[5, 6])?;
+    assert_eq!(
+        c.outcome(1042),
+        Some(Outcome::Committed(Applied::Source(SourceReply::Pinned {
+            index
+        })))
+    );
+    c.queue.clear();
+    c.peers.clear();
+    for id in 1..=6 {
+        c.start(id)?;
+    }
+    let catalog = c.elect(2, 10, &[1, 2, 3])?;
+    let source = c.elect(6, 20, &[4, 5, 6])?;
+    c.command(catalog, Request::Query(10, 1050, query))?;
+    c.pump(&[1, 2, 3])?;
+    let Some(ReadOutcome::Observed {
+        value: Value::Pin(Some(permit)),
+        ..
+    }) = c.read_outcome(1050)
+    else {
+        anyhow::bail!("restart attestation missing");
+    };
+    c.command(source, Request::Pin(1051, *permit))?;
+    c.pump(&[4, 5, 6])?;
+    assert_eq!(
+        c.outcome(1051),
+        Some(Outcome::Committed(Applied::Source(SourceReply::Pinned {
+            index
+        })))
+    );
+    c.command(source, Request::Query(20, 1052, Query::SourcePin { table }))?;
+    c.pump(&[4, 5, 6])?;
+    let Some(ReadOutcome::Observed {
+        value: Value::SourcePin(Some(pin)),
+        ..
+    }) = c.read_outcome(1052)
+    else {
+        anyhow::bail!("restart source pin view missing");
+    };
+    assert_eq!(pin.index, index);
+    assert_eq!(pin.permit.selection().plan_index, plan_index);
+    for id in 4..=6 {
+        verify_records(c.peers.get_mut(&id).unwrap(), 20, &rows)?;
+    }
+    Ok(())
+}
diff --git a/docs/design/wal-publication-plans.md b/docs/design/wal-publication-plans.md
index 3a8b37e..d8f1822 100644
--- a/docs/design/wal-publication-plans.md
+++ b/docs/design/wal-publication-plans.md
@@ -36,14 +36,14 @@ active plans. Large metadata frames allocate only their bounded actual length on
 the heap. Decoders reject noncanonical maps instead of silently accepting duplicate
 keys normalized by the serializer.
 
-This changes the optional application identity and checkpoint format to v3.
-Existing v1/v2 directories and images are rejected; there is no in-place migration
+Source pins subsequently advance the optional application identity and checkpoint
+format to v4. Existing v1/v2/v3 directories and images are rejected; there is no in-place migration
 or production rollout in this increment. Catalog-only and mixed data/catalog
 snapshots preserve plans, with semantic validation after checksum verification.
 
 A stored plan is not a source pin, proof of complete source selection or permission
-to commit a Lance manifest. Source groups must still validate their full selected
-prefixes through the subsequent pin protocol. Descriptor CAS, exact-target Lance
+to commit a Lance manifest. Source groups validate their full selected prefixes
+through the [source pin protocol](wal-source-pins.md). Descriptor CAS, exact-target Lance
 outcome recovery, resolution and cleanup commands remain subsequent work; this
 increment intentionally retains pending plans and advances no published cursor.
 No source payload can be retired on the strength of this plan alone.
diff --git a/docs/design/wal-publisher-authority.md b/docs/design/wal-publisher-authority.md
index 2f34f5e..107ab5c 100644
--- a/docs/design/wal-publisher-authority.md
+++ b/docs/design/wal-publisher-authority.md
@@ -36,14 +36,16 @@ scope, epochs, grant positions, destination uniqueness and contracts, then repla
 the committed suffix. Metadata contains bounded, ordered table frames alongside
 the existing data records and producer tails. Payload packs still contain only
 opaque accepted data. The current replication identity and application image
-format are v3, including [immutable publication plans](wal-publication-plans.md).
-Earlier v1/v2 application directories are rejected; this optional component does
+format are v4, including [immutable publication plans](wal-publication-plans.md)
+and [source pins](wal-source-pins.md).
+Earlier v1/v2/v3 application directories are rejected; this optional component does
 not provide an in-place migration or change the production ingestion path.
 
 `Node::publisher` exposes only a local applied view. It is not a quorum read,
 lease, authorization check or proof of Lance visibility. A granted epoch alone
 does not stop an old owner from writing outside this protocol. Durable publication
-plans check the current authority at apply. Subsequent work must verify source dependencies, commit a
+plans check the current authority at apply. Source pins verify the immutable
+selection against the complete source prefix. Subsequent work must commit a
 pinned Lance target, resolve its immutable outcome and authorize retirement from
 that outcome. Table closing/rebinding, table-bound data admission and integration
 with the actual context catalog remain separate requirements.
diff --git a/docs/design/wal-source-pins.md b/docs/design/wal-source-pins.md
new file mode 100644
index 0000000..7c00e15
--- /dev/null
+++ b/docs/design/wal-source-pins.md
@@ -0,0 +1,52 @@
+# Catalog attestations and durable source pins
+
+Refs #378.
+
+A catalog plan identifies source records but cannot prove that a source still has
+its complete selected prefix. `Node::pin_source` now verifies a catalog attestation
+and commits the exact source dependency through that source group's quorum.
+The source rejects omitted, changed or future records. Independent catalog and
+source indices remain separate; only a source's own `through` position must
+precede its pin entry.
+
+A `Query::Pin` request names an exact table, transaction, catalog plan index and
+source group. Only safe ReadIndex completion at the catalog leader can issue its
+Ed25519 attestation. The signed metadata includes cluster/layout/policy scope,
+the confirmed catalog cut, ingestion contract and source selection. A caller
+cannot turn a serialized local view into a valid attestation. The signature is
+on publication metadata, not on individual data payloads. Replication continues
+to treat Arrow bodies as opaque bytes.
+
+The optional immutable `Limits::catalog_key` contains the public verification
+key. Trusted catalog replicas install a matching PKCS#8 private key with
+`set_catalog_signer`; source-only replicas need no signing key. Private bytes
+are never included in the application identity, snapshots, queries or commands.
+Key provisioning and rotation are deployment responsibilities. Changing the
+public key changes the durable policy and requires an explicit migration.
+This is an attestation by a trusted catalog service after quorum confirmation,
+not Byzantine consensus, transport authentication or authorization for callers.
+Existing peer and producer transports still require their own authentication.
+
+A refreshed observation or publisher takeover retains the immutable selection.
+Its retry traverses source consensus and returns the original pin index. Shared
+`Arc` ownership avoids copying a pin's selected metadata into captured views.
+Pins survive local checkpoints, received snapshots and retained-storage restart;
+checkpoint validation verifies both signatures and complete source prefixes.
+Command and image metadata lengths and the number of active pins are bounded.
+
+The application/checkpoint format advances to v4 and rejects older identities
+and images. There is no in-place migration in this optional component.
+
+This increment retains pins and payloads. It does not enable release, advance
+source publication cursors or permit Lance commits or deletion. All initial
+source prefixes begin at zero. Outcome resolution must subsequently add release
+fences, including for sources that never pinned, before catalog plans may be
+removed or their slots reused. The cleanup and recovery boundary
+requires separate coordination for manifests and staged files; a signed pin is
+not a publisher lease or proof of Lance visibility.
+
+Validation exercises disjoint catalog/source replica sets, safe query completion,
+changed and omitted records, signature tampering, independent numeric indices,
+source snapshot transfer, publisher takeover and full process restart. The six
+TCP processes use real persistent Raft logs; source replicas hold only the public
+key and recover the original pin index and exact source bytes.