diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md
index 5674918..85d32b2 100644
--- a/crates/lance-context-wal/README.md
+++ b/crates/lance-context-wal/README.md
@@ -176,6 +176,11 @@ publication prerequisite, not a production Lance writer or a manifest lease.
 catalog, producer-tail and accepted-page views after a safe read-index barrier.
 They share pending capacity with proposals and do not fetch bulk payloads.
 
+[Source closure proofs](../../docs/design/wal-source-closure-proofs.md) attest that
+an exact catalog-authorized closure committed at its source. Optional per-group
+keys keep catalog private keys off source-only replicas; catalog checkpoint
+advancement and physical reclamation remain separate steps.
+
 [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](../../docs/design/wal-source-pins.md)
diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs
index 1458216..e2858e0 100644
--- a/crates/lance-context-wal/src/replication.rs
+++ b/crates/lance-context-wal/src/replication.rs
@@ -46,6 +46,11 @@ 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]>,
+    /// Optional per-group keys for quorum-confirmed source closure proofs.
+    /// A nonempty policy must cover every group with distinct keys, separate
+    /// from the catalog key. Installing it is an explicit policy migration.
+    #[serde(default)]
+    pub source_keys: BTreeMap<u64, [u8; 32]>,
     pub max_tables: u64,
     pub max_active_plans: u64,
     pub max_plan_source_bytes: u64,
@@ -61,6 +66,7 @@ impl Default for Limits {
     fn default() -> Self {
         Self {
             catalog_key: None,
+            source_keys: BTreeMap::new(),
             max_tables: 1024,
             max_active_plans: 128,
             max_plan_source_bytes: 64 * 1024 * 1024,
@@ -79,6 +85,17 @@ impl Limits {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
                 && self.catalog_key.is_none_or(|key| key != [0; 32])
+                && self.source_keys.len() <= 64
+                && (self.source_keys.is_empty() || self.catalog_key.is_some())
+                && self.source_keys.iter().all(|(group, key)| {
+                    *group > 0 && *key != [0; 32] && Some(*key) != self.catalog_key
+                })
+                && self
+                    .source_keys
+                    .values()
+                    .collect::<std::collections::BTreeSet<_>>()
+                    .len()
+                    == self.source_keys.len()
                 && (1..=256).contains(&self.max_active_plans)
                 && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes)
                 && (1..=publication::MAX_DESCRIPTOR_BYTES as u64)
@@ -92,8 +109,26 @@ impl Limits {
                 && self.max_uncommitted_bytes <= 256 * 1024 * 1024,
             "invalid replication limits"
         );
+        if self.source_keys.is_empty() {
+            // Bincode structs and tuples encode their fields in order. Keep the
+            // exact old policy bytes when source attestations are not enabled.
+            return Ok(bincode::serialize(&(
+                "lance-context/replicated-append/v9",
+                self.catalog_key,
+                self.max_tables,
+                self.max_active_plans,
+                self.max_plan_source_bytes,
+                self.max_descriptor_bytes,
+                self.max_manifest_records,
+                self.max_record_bytes,
+                self.max_records,
+                self.max_producers,
+                self.max_pending,
+                self.max_uncommitted_bytes,
+            ))?);
+        }
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v9",
+            "lance-context/replicated-append/v10",
             self,
         ))?)
     }
@@ -185,6 +220,7 @@ struct Group {
 /// drop it, reconcile outstanding requests and reopen the original directories.
 pub struct Node {
     signer: Option<Arc<source::CatalogSigner>>,
+    source_signers: BTreeMap<u64, Arc<source::SourceSigner>>,
     layout: Arc<Layout>,
     layout_hash: [u8; 32],
     policy: [u8; 32],
@@ -208,6 +244,10 @@ impl Node {
                 } else {
                     None
                 },
+                source_signers: self
+                    .source_signers
+                    .get(&id)
+                    .map_or_else(BTreeMap::new, |s| BTreeMap::from([(id, s.clone())])),
                 layout: self.layout.clone(),
                 layout_hash: self.layout_hash,
                 policy: self.policy,
@@ -223,6 +263,10 @@ impl Node {
     pub fn open(directory: &Path, layout: Arc<Layout>, id: u64, limits: Limits) -> Result<Self> {
         let layout_hash = layout.fingerprint()?;
         let identity = limits.identity()?;
+        ensure!(
+            limits.source_keys.is_empty() || limits.source_keys.keys().eq(layout.groups.keys()),
+            "source signing policy must cover exactly the configured groups"
+        );
         let policy = Sha256::digest(&identity).into();
         let log = PhysicalLog::open(directory, &layout, id)?;
         let images = images::Owner::new(directory)?;
@@ -284,6 +328,7 @@ impl Node {
         }
         Ok(Self {
             signer: None,
+            source_signers: BTreeMap::new(),
             layout,
             layout_hash,
             policy,
@@ -318,6 +363,23 @@ impl Node {
         Ok(())
     }
 
+    /// Provision this source group's key without exposing the catalog's private
+    /// key to source-only replicas. The persisted policy pins the public key.
+    pub fn set_source_signer(
+        &mut self,
+        group: u64,
+        signer: Arc<source::SourceSigner>,
+    ) -> Result<()> {
+        self.healthy()?;
+        ensure!(
+            self.groups.contains_key(&group)
+                && self.limits.source_keys.get(&group) == Some(&signer.public_key()),
+            "source signer does not match hosted scope and policy"
+        );
+        self.source_signers.insert(group, signer);
+        Ok(())
+    }
+
     pub fn tick(&mut self) -> Result<()> {
         self.healthy()?;
         for group in self.groups.values_mut() {
@@ -369,12 +431,21 @@ impl Node {
             "request token already pending"
         );
         query.validate(&self.layout, group)?;
-        if matches!(query, query::Query::Pin { .. }) {
+        if matches!(
+            query,
+            query::Query::Pin { .. } | query::Query::ClosePermit { .. }
+        ) {
             ensure!(
                 self.signer.is_some(),
                 "catalog dependency signer unavailable"
             );
         }
+        if matches!(query, query::Query::ClosedProof { .. }) {
+            ensure!(
+                self.source_signers.contains_key(&group),
+                "source 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.
@@ -668,7 +739,10 @@ impl Node {
                         cut,
                         &self.layout,
                         &self.limits,
-                        self.signer.as_deref(),
+                        source::QuerySigners {
+                            catalog: self.signer.as_deref(),
+                            source: self.source_signers.get(id).map(Arc::as_ref),
+                        },
                     )
                 },
             )?);
diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs
index bdc24f3..51fa3f5 100644
--- a/crates/lance-context-wal/src/replication/query.rs
+++ b/crates/lance-context-wal/src/replication/query.rs
@@ -42,6 +42,9 @@ pub enum Query {
     SourcePin {
         table: Uuid,
     },
+    ClosedProof {
+        table: Uuid,
+    },
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -55,6 +58,7 @@ pub enum Value {
     CursorAhead { applied: u64 },
     Pin(Option<Box<source::Permit>>),
     SourcePin(Option<Arc<source::Pin>>),
+    ClosedProof(Option<Box<source::ClosedProof>>),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -114,7 +118,9 @@ impl Query {
                     && group == publication::control_group(layout),
                 "invalid catalog dependency query"
             ),
-            Self::SourceClosure { table } | Self::SourcePin { table } => {
+            Self::SourceClosure { table }
+            | Self::SourcePin { table }
+            | Self::ClosedProof { table } => {
                 ensure!(!table.is_nil(), "nil source table")
             }
             Self::Publisher { table } => ensure!(
@@ -143,9 +149,25 @@ impl Query {
         cut: &Cut,
         layout: &Layout,
         limits: &Limits,
-        signer: Option<&source::CatalogSigner>,
+        signers: source::QuerySigners<'_>,
     ) -> Result<Value> {
         Ok(match self {
+            Self::ClosedProof { table } => Value::ClosedProof(
+                state
+                    .closed
+                    .get(table)
+                    .map(|closed| {
+                        source::ClosedProof::issue(
+                            signers.source.context("source signer unavailable")?,
+                            cut.clone(),
+                            layout,
+                            limits,
+                            closed.clone(),
+                        )
+                        .map(Box::new)
+                    })
+                    .transpose()?,
+            ),
             Self::SourceClosure { table } => Value::SourceClosure(state.closed.get(table).cloned()),
             Self::ClosePermit {
                 table,
@@ -165,7 +187,9 @@ impl Query {
                     selected
                         .map(|table| {
                             source::ClosurePermit::issue(
-                                signer.context("catalog closure signer unavailable")?,
+                                signers
+                                    .catalog
+                                    .context("catalog closure signer unavailable")?,
                                 cut.clone(),
                                 layout,
                                 limits,
@@ -203,7 +227,9 @@ impl Query {
                     selected
                         .map(|table| {
                             source::Permit::issue(
-                                signer.context("catalog dependency signer unavailable")?,
+                                signers
+                                    .catalog
+                                    .context("catalog dependency signer unavailable")?,
                                 cut.clone(),
                                 layout,
                                 limits,
diff --git a/crates/lance-context-wal/src/replication/query/tests.rs b/crates/lance-context-wal/src/replication/query/tests.rs
index 0464bce..efda4bf 100644
--- a/crates/lance-context-wal/src/replication/query/tests.rs
+++ b/crates/lance-context-wal/src/replication/query/tests.rs
@@ -39,7 +39,10 @@ fn evaluate(query: &Query, cut: &Cut, state: &State) -> Result<Value> {
         cut,
         &super::super::tests::layout(),
         &Limits::default(),
-        None,
+        source::QuerySigners {
+            catalog: None,
+            source: None,
+        },
     )
 }
 
diff --git a/crates/lance-context-wal/src/replication/source.rs b/crates/lance-context-wal/src/replication/source.rs
index cb54e4d..63bef9e 100644
--- a/crates/lance-context-wal/src/replication/source.rs
+++ b/crates/lance-context-wal/src/replication/source.rs
@@ -7,8 +7,15 @@ use bincode::Options;
 use bytes::Bytes;
 use ring::signature::{Ed25519KeyPair, KeyPair, UnparsedPublicKey, ED25519};
 mod closure;
+mod proof;
 pub(super) use closure::{decode_closed, encode_closed, MAX_CLOSED_BYTES};
 pub use closure::{Closed, Closure, ClosurePermit};
+pub use proof::{ClosedProof, SourceSigner};
+
+pub(super) struct QuerySigners<'a> {
+    pub catalog: Option<&'a CatalogSigner>,
+    pub source: Option<&'a SourceSigner>,
+}
 
 #[cfg(test)]
 mod tests;
diff --git a/crates/lance-context-wal/src/replication/source/proof.rs b/crates/lance-context-wal/src/replication/source/proof.rs
new file mode 100644
index 0000000..321e1a1
--- /dev/null
+++ b/crates/lance-context-wal/src/replication/source/proof.rs
@@ -0,0 +1,107 @@
+//! A source quorum's durable closure, distinct from a catalog permit to close.
+use super::*;
+
+const DOMAIN: &str = "lance-context/source-closed-proof/v1";
+const MAX_PROOF_BYTES: usize = 4096;
+
+/// Private signing key shared only by the replicas of one source group.
+/// This has no public signing operation and does not contain the catalog key.
+pub struct SourceSigner(Ed25519KeyPair);
+impl SourceSigner {
+    pub fn from_pkcs8(bytes: &[u8]) -> Result<Self> {
+        Ok(Self(Ed25519KeyPair::from_pkcs8(bytes).map_err(|_| {
+            anyhow::anyhow!("invalid source signing key")
+        })?))
+    }
+
+    pub fn public_key(&self) -> [u8; 32] {
+        self.0
+            .public_key()
+            .as_ref()
+            .try_into()
+            .expect("Ed25519 key")
+    }
+}
+
+/// Issued only from a successful source ReadIndex completion. It attests the
+/// exact persisted closure and its original source index. It is not a lease,
+/// publisher authority or permission to delete Lance objects or log files.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ClosedProof {
+    policy: [u8; 32],
+    cut: query::Cut,
+    closed: Arc<Closed>,
+    signature: ([u8; 32], [u8; 32]),
+}
+impl ClosedProof {
+    pub fn source_cut(&self) -> &query::Cut {
+        &self.cut
+    }
+
+    pub fn closed(&self) -> &Closed {
+        &self.closed
+    }
+
+    fn signed_bytes(&self) -> Result<Vec<u8>> {
+        Ok(codec(MAX_PROOF_BYTES).serialize(&(DOMAIN, self.policy, &self.cut, &self.closed))?)
+    }
+
+    pub(in crate::replication) fn issue(
+        signer: &SourceSigner,
+        cut: query::Cut,
+        layout: &Layout,
+        limits: &Limits,
+        closed: Arc<Closed>,
+    ) -> Result<Self> {
+        ensure!(
+            limits.source_keys.get(&cut.group) == Some(&signer.public_key()),
+            "source signing policy mismatch"
+        );
+        let mut proof = Self {
+            policy: Sha256::digest(limits.identity()?).into(),
+            cut,
+            closed,
+            signature: ([0; 32], [0; 32]),
+        };
+        let signature = signer.0.sign(&proof.signed_bytes()?);
+        proof.signature = (
+            signature.as_ref()[..32].try_into()?,
+            signature.as_ref()[32..].try_into()?,
+        );
+        proof.verify(layout, limits)?;
+        Ok(proof)
+    }
+
+    /// Authenticate the scoped statement. A consumer must additionally match
+    /// its exact expected table, plan and publication before advancing state.
+    pub fn verify(&self, layout: &Layout, limits: &Limits) -> Result<()> {
+        let key = limits
+            .source_keys
+            .get(&self.cut.group)
+            .context("source verification key unavailable")?;
+        ensure!(
+            self.policy == <[u8; 32]>::from(Sha256::digest(limits.identity()?)),
+            "foreign source proof 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 source closure proof"))?;
+        self.closed.permit.verify(layout, limits, self.cut.group)?;
+        ensure!(
+            self.cut.cluster == layout.cluster
+                && self.cut.layout == layout.fingerprint()?
+                && self.cut.term > 0
+                && self.cut.quorum_index > 0
+                && self.cut.quorum_index <= self.cut.applied
+                && self.cut.applied < u64::MAX
+                && self.closed.index <= self.cut.applied
+                && self.closed.index > self.closed.pin_index
+                && self.closed.pin_index > self.closed.permit.closure().through,
+            "invalid source closure proof scope or cut"
+        );
+        Ok(())
+    }
+}
diff --git a/crates/lance-context-wal/src/replication/source/tests.rs b/crates/lance-context-wal/src/replication/source/tests.rs
index 9483c71..6163d54 100644
--- a/crates/lance-context-wal/src/replication/source/tests.rs
+++ b/crates/lance-context-wal/src/replication/source/tests.rs
@@ -10,6 +10,7 @@ struct Cluster {
     layout: Arc<Layout>,
     limits: Limits,
     signer: Arc<CatalogSigner>,
+    source_signers: BTreeMap<u64, Arc<SourceSigner>>,
     table: Uuid,
     transaction: Uuid,
 }
@@ -21,6 +22,9 @@ impl Cluster {
         Self::with_policy(max_active_plans, 100_000)
     }
     fn with_policy(max_active_plans: u64, max_records: u64) -> Result<Self> {
+        Self::with_source_policy(max_active_plans, max_records, false)
+    }
+    fn with_source_policy(max_active_plans: u64, max_records: u64, proofs: bool) -> Result<Self> {
         let root = tempfile::tempdir()?;
         let layout = Arc::new(Layout {
             cluster: Uuid::from_u128(91),
@@ -33,8 +37,20 @@ impl Cluster {
         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 mut source_signers = BTreeMap::new();
+        if proofs {
+            for group in layout.groups.keys() {
+                let key = Ed25519KeyPair::generate_pkcs8(&ring::rand::SystemRandom::new())
+                    .map_err(|_| anyhow::anyhow!("test source key generation"))?;
+                source_signers.insert(*group, Arc::new(SourceSigner::from_pkcs8(key.as_ref())?));
+            }
+        }
         let limits = Limits {
             catalog_key: Some(signer.public_key()),
+            source_keys: source_signers
+                .iter()
+                .map(|(g, s)| (*g, s.public_key()))
+                .collect(),
             max_active_plans,
             max_records,
             ..Default::default()
@@ -45,6 +61,7 @@ impl Cluster {
             layout,
             limits,
             signer,
+            source_signers,
             table: Uuid::from_u128(92),
             transaction: Uuid::from_u128(93),
         };
@@ -63,6 +80,11 @@ impl Cluster {
             if id <= 3 {
                 node.set_catalog_signer(self.signer.clone())?;
             }
+            for (group, signer) in &self.source_signers {
+                if self.layout.groups[group].contains(&id) {
+                    node.set_source_signer(*group, signer.clone())?;
+                }
+            }
             self.nodes.push(node);
         }
         self.nodes[0].campaign(10)?;
diff --git a/crates/lance-context-wal/src/replication/source/tests/closed.rs b/crates/lance-context-wal/src/replication/source/tests/closed.rs
index ca24bc0..798ee1a 100644
--- a/crates/lance-context-wal/src/replication/source/tests/closed.rs
+++ b/crates/lance-context-wal/src/replication/source/tests/closed.rs
@@ -3,6 +3,7 @@ use publication::{
     BootstrapManifest, ConfirmPublication, Descriptors, ManifestObject, PrepareRequest,
     ReserveManifest,
 };
+mod proof;
 
 fn publisher(events: Events) -> Result<publication::Reply> {
     events
@@ -22,7 +23,10 @@ fn fixture_with_background(
     max_records: u64,
     background: u64,
 ) -> Result<(Cluster, Permit, Uuid, Uuid)> {
-    let mut c = Cluster::with_policy(128, max_records)?;
+    fixture_from(Cluster::with_policy(128, max_records)?, background)
+}
+
+fn fixture_from(mut c: Cluster, background: u64) -> Result<(Cluster, Permit, Uuid, Uuid)> {
     let authority = c.acquire(0)?;
     let incarnation = Uuid::from_u128(800);
     c.nodes[0].bootstrap_manifest(
diff --git a/crates/lance-context-wal/src/replication/source/tests/closed/proof.rs b/crates/lance-context-wal/src/replication/source/tests/closed/proof.rs
new file mode 100644
index 0000000..2f0adbf
--- /dev/null
+++ b/crates/lance-context-wal/src/replication/source/tests/closed/proof.rs
@@ -0,0 +1,247 @@
+use super::*;
+
+fn proofs_fixture() -> Result<(Cluster, Permit, Uuid, Uuid)> {
+    fixture_from(Cluster::with_source_policy(128, 100_000, true)?, 12)
+}
+
+fn observe(c: &mut Cluster, token: u64) -> Result<Option<Box<ClosedProof>>> {
+    c.nodes[3].query(20, token, Query::ClosedProof { table: c.table })?;
+    c.pump()?
+        .reads
+        .into_iter()
+        .find_map(|r| match r.outcome {
+            ReadOutcome::Observed {
+                value: Value::ClosedProof(proof),
+                ..
+            } => Some(proof),
+            _ => None,
+        })
+        .context("closure proof read missing")
+}
+
+#[test]
+fn source_attests_only_committed_closure_and_preserves_it_after_checkpoint_restart() -> Result<()> {
+    let (mut c, _, incarnation, predecessor) = proofs_fixture()?;
+    assert!(observe(&mut c, 900)?.is_none());
+    let permit = publish(&mut c, incarnation, predecessor)?;
+    assert!(
+        observe(&mut c, 901)?.is_none(),
+        "catalog authorization is not source closure"
+    );
+    let Reply::Closed { index } = close(&mut c, permit.clone())? else {
+        anyhow::bail!("closure missing")
+    };
+    let first = observe(&mut c, 902)?.context("closed proof missing")?;
+    first.verify(&c.layout, &c.limits)?;
+    assert_eq!(first.closed().index, index);
+    assert_eq!(first.closed().permit, permit);
+    assert!(c.nodes[3..].iter().all(|node| node.signer.is_none()));
+    let wire = bincode::serialize(&first)?;
+    let decoded: ClosedProof = bincode::deserialize(&wire)?;
+    decoded.verify(&c.layout, &c.limits)?;
+    for offset in [0, wire.len() / 2, wire.len() - 1] {
+        let mut damaged = wire.clone();
+        damaged[offset] ^= 1;
+        if let Ok(proof) = bincode::deserialize::<ClosedProof>(&damaged) {
+            assert!(proof.verify(&c.layout, &c.limits).is_err());
+        }
+    }
+    let mut foreign = (*c.layout).clone();
+    foreign.cluster = Uuid::new_v4();
+    assert!(first.verify(&foreign, &c.limits).is_err());
+    let mut policy = c.limits.clone();
+    policy.max_records += 1;
+    assert!(first.verify(&c.layout, &policy).is_err());
+    policy = c.limits.clone();
+    policy
+        .source_keys
+        .insert(20, c.source_signers[&10].public_key());
+    assert!(first.verify(&c.layout, &policy).is_err());
+    for node in &mut c.nodes {
+        let group = if node.id <= 3 { 10 } else { 20 };
+        let image = node.prepare_checkpoint(group, Uuid::new_v4())?.run()?;
+        node.install_checkpoint(image)?;
+    }
+    c.reopen()?;
+    let refreshed = observe(&mut c, 903)?.context("proof missing after restart")?;
+    refreshed.verify(&c.layout, &c.limits)?;
+    assert_eq!(refreshed.closed(), first.closed());
+    assert_ne!(refreshed.source_cut(), first.source_cut());
+    first.verify(&c.layout, &c.limits)?;
+    Ok(())
+}
+
+#[test]
+fn isolated_source_leader_cannot_refresh_a_proof_from_its_local_closure() -> Result<()> {
+    let (mut c, _, incarnation, predecessor) = proofs_fixture()?;
+    let permit = publish(&mut c, incarnation, predecessor)?;
+    close(&mut c, permit)?;
+    assert!(observe(&mut c, 904)?.is_some());
+    c.nodes[3].query(20, 905, Query::ClosedProof { table: c.table })?;
+    let mut unknown = false;
+    for _ in 0..40 {
+        c.nodes[3].tick()?;
+        for read in c.nodes[3].poll()?.reads {
+            assert!(matches!(read.outcome, ReadOutcome::Unknown));
+            unknown = true;
+        }
+    }
+    assert!(unknown);
+    Ok(())
+}
+
+#[test]
+fn missing_signers_reject_before_read_admission_and_keep_the_node_healthy() -> Result<()> {
+    let (mut c, _, _, _) = proofs_fixture()?;
+    c.nodes[3].source_signers.clear();
+    assert!(c.nodes[3]
+        .query(20, 906, Query::ClosedProof { table: c.table })
+        .is_err());
+    c.nodes[3].query(20, 906, Query::SourceClosure { table: c.table })?;
+    assert!(c.pump()?.reads.iter().any(|r| r.request == 906));
+    c.nodes[0].signer = None;
+    assert!(c.nodes[0]
+        .query(
+            10,
+            907,
+            Query::ClosePermit {
+                table: c.table,
+                transaction: c.transaction,
+                plan_index: 1,
+                source_group: 20,
+            }
+        )
+        .is_err());
+    c.nodes[0].query(10, 907, Query::Publisher { table: c.table })?;
+    assert!(c.pump()?.reads.iter().any(|r| r.request == 907));
+    Ok(())
+}
+
+#[test]
+fn source_signing_policy_rejects_missing_reused_and_unhosted_keys() -> Result<()> {
+    let c = Cluster::with_source_policy(128, 100_000, true)?;
+    for keys in [
+        BTreeMap::from([(10, c.source_signers[&10].public_key())]),
+        BTreeMap::from([
+            (10, c.source_signers[&10].public_key()),
+            (30, c.source_signers[&20].public_key()),
+        ]),
+        BTreeMap::from([
+            (10, c.source_signers[&10].public_key()),
+            (20, c.source_signers[&10].public_key()),
+        ]),
+        BTreeMap::from([
+            (10, c.signer.public_key()),
+            (20, c.source_signers[&20].public_key()),
+        ]),
+        BTreeMap::from([(10, [0; 32]), (20, c.source_signers[&20].public_key())]),
+    ] {
+        let root = tempfile::tempdir()?;
+        let mut policy = c.limits.clone();
+        policy.source_keys = keys;
+        assert!(Node::open(root.path(), c.layout.clone(), 1, policy).is_err());
+        assert_eq!(
+            std::fs::read_dir(root.path())?.count(),
+            0,
+            "bad policy opened storage"
+        );
+    }
+    let root = tempfile::tempdir()?;
+    let mut node = Node::open(root.path(), c.layout.clone(), 4, c.limits.clone())?;
+    assert!(node
+        .set_source_signer(10, c.source_signers[&10].clone())
+        .is_err());
+    assert!(node
+        .set_source_signer(20, c.source_signers[&10].clone())
+        .is_err());
+    node.set_source_signer(20, c.source_signers[&20].clone())?;
+    drop(node);
+    let mut policy = c.limits.clone();
+    policy.source_keys.clear();
+    let error = Node::open(root.path(), c.layout.clone(), 4, policy)
+        .err()
+        .context("policy change accepted")?;
+    assert!(format!("{error:#}").contains("policy"));
+    Ok(())
+}
+
+#[test]
+fn splitting_hosted_groups_keeps_only_their_own_source_key() -> Result<()> {
+    let c = Cluster::with_source_policy(128, 100_000, true)?;
+    let root = tempfile::tempdir()?;
+    let mut layout = (*c.layout).clone();
+    layout.nodes.retain(|id, _| *id <= 3);
+    layout.groups.insert(20, vec![1, 2, 3]);
+    let mut node = Node::open(root.path(), Arc::new(layout), 1, c.limits.clone())?;
+    node.set_catalog_signer(c.signer.clone())?;
+    for (group, signer) in &c.source_signers {
+        node.set_source_signer(*group, signer.clone())?;
+    }
+    let groups = node.into_groups();
+    assert_eq!(groups.len(), 2);
+    for node in groups {
+        assert_eq!(node.groups.len(), 1);
+        let group = *node.groups.keys().next().unwrap();
+        assert_eq!(node.source_signers.len(), 1);
+        assert_eq!(
+            node.source_signers[&group].public_key(),
+            c.limits.source_keys[&group]
+        );
+        assert_eq!(node.signer.is_some(), group == 10);
+    }
+    Ok(())
+}
+
+#[test]
+fn empty_source_keys_preserve_independently_encoded_v9_policy_bytes() -> Result<()> {
+    #[derive(Serialize)]
+    struct LegacyLimits {
+        catalog_key: Option<[u8; 32]>,
+        max_tables: u64,
+        max_active_plans: u64,
+        max_plan_source_bytes: u64,
+        max_descriptor_bytes: u64,
+        max_manifest_records: u64,
+        max_record_bytes: u64,
+        max_records: u64,
+        max_producers: u64,
+        max_pending: u64,
+        max_uncommitted_bytes: u64,
+    }
+    for limits in [
+        Limits::default(),
+        Limits {
+            catalog_key: Some([17; 32]),
+            max_tables: 7,
+            max_active_plans: 3,
+            max_plan_source_bytes: 12345,
+            max_descriptor_bytes: 4321,
+            max_manifest_records: 9,
+            max_record_bytes: 16384,
+            max_records: 123,
+            max_producers: 31,
+            max_pending: 43,
+            max_uncommitted_bytes: 2 * 1024 * 1024,
+            ..Default::default()
+        },
+    ] {
+        let old = LegacyLimits {
+            catalog_key: limits.catalog_key,
+            max_tables: limits.max_tables,
+            max_active_plans: limits.max_active_plans,
+            max_plan_source_bytes: limits.max_plan_source_bytes,
+            max_descriptor_bytes: limits.max_descriptor_bytes,
+            max_manifest_records: limits.max_manifest_records,
+            max_record_bytes: limits.max_record_bytes,
+            max_records: limits.max_records,
+            max_producers: limits.max_producers,
+            max_pending: limits.max_pending,
+            max_uncommitted_bytes: limits.max_uncommitted_bytes,
+        };
+        assert_eq!(
+            limits.identity()?,
+            bincode::serialize(&("lance-context/replicated-append/v9", old))?
+        );
+    }
+    Ok(())
+}
diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs
index df9915d..a2d774f 100644
--- a/crates/lance-context-wal/tests/replication_process.rs
+++ b/crates/lance-context-wal/tests/replication_process.rs
@@ -13,7 +13,7 @@ use lance_context_wal::{
             Rejection as PublisherRejection, Reply as PublisherReply, SourceRange, Table,
         },
         query::{Completion as ReadCompletion, Outcome as ReadOutcome, Query, Value},
-        source::{CatalogSigner, Permit, Reply as SourceReply},
+        source::{CatalogSigner, Permit, Reply as SourceReply, SourceSigner},
         Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, PreparedCheckpoint,
         SnapshotReceiver, Status,
     },
@@ -257,7 +257,7 @@ fn replication_peer_fixture() -> Result<()> {
     let id = std::env::var("CONTEXT_REPLICATION_NODE")?.parse()?;
     let listener = TcpListener::bind("127.0.0.1:0")?;
     let limits_path = root.join("limits.json");
-    let limits = if limits_path.exists() {
+    let limits: Limits = if limits_path.exists() {
         serde_json::from_slice(&std::fs::read(limits_path)?)?
     } else {
         Limits::default()
@@ -268,7 +268,7 @@ fn replication_peer_fixture() -> Result<()> {
             root.join("layout.json"),
         )?)?),
         id,
-        limits,
+        limits.clone(),
     )?;
     let key_path = root.join(format!("catalog-key-{id}.pkcs8"));
     if key_path.exists() {
@@ -276,6 +276,15 @@ fn replication_peer_fixture() -> Result<()> {
             key_path,
         )?)?))?;
     }
+    for group in limits.source_keys.keys() {
+        let key_path = root.join(format!("source-key-{group}-{id}.pkcs8"));
+        if key_path.exists() {
+            node.set_source_signer(
+                *group,
+                Arc::new(SourceSigner::from_pkcs8(&std::fs::read(key_path)?)?),
+            )?;
+        }
+    }
     let mut worker = Worker {
         node,
         prepared: None,
@@ -386,10 +395,26 @@ impl Cluster {
         let root = tempfile::tempdir()?;
         if let Some(key) = catalog_key {
             let signer = CatalogSigner::from_pkcs8(key)?;
+            let mut source_keys = BTreeMap::new();
+            for (group, members) in &layout.groups {
+                let source_key = ring::signature::Ed25519KeyPair::generate_pkcs8(
+                    &ring::rand::SystemRandom::new(),
+                )
+                .map_err(|_| anyhow::anyhow!("test source key generation failed"))?;
+                let source_signer = SourceSigner::from_pkcs8(source_key.as_ref())?;
+                source_keys.insert(*group, source_signer.public_key());
+                for id in members {
+                    std::fs::write(
+                        root.path().join(format!("source-key-{group}-{id}.pkcs8")),
+                        source_key.as_ref(),
+                    )?;
+                }
+            }
             std::fs::write(
                 root.path().join("limits.json"),
                 serde_json::to_vec(&Limits {
                     catalog_key: Some(signer.public_key()),
+                    source_keys,
                     ..Default::default()
                 })?,
             )?;
diff --git a/crates/lance-context-wal/tests/replication_process/source_closure.rs b/crates/lance-context-wal/tests/replication_process/source_closure.rs
index 2cbf827..8cb86a3 100644
--- a/crates/lance-context-wal/tests/replication_process/source_closure.rs
+++ b/crates/lance-context-wal/tests/replication_process/source_closure.rs
@@ -1,7 +1,35 @@
 //! Real quorum closure after a real Lance publication, with a lost reply and
 //! full process restart before attempting to replay the original pin request.
 use super::*;
-use lance_context_wal::replication::source::Rejection as SourceRejection;
+use lance_context_wal::replication::source::{ClosedProof, Rejection as SourceRejection};
+
+fn proof(
+    c: &mut Cluster,
+    source: u64,
+    token: u64,
+    table: Uuid,
+) -> Result<Option<Box<ClosedProof>>> {
+    c.command(
+        source,
+        Request::Query(20, token, Query::ClosedProof { table }),
+    )?;
+    c.pump(&[4, 5, 6])?;
+    let Some(ReadOutcome::Observed {
+        value: Value::ClosedProof(proof),
+        ..
+    }) = c.read_outcome(token)
+    else {
+        anyhow::bail!("source closure proof read missing")
+    };
+    let layout: Layout =
+        serde_json::from_slice(&std::fs::read(c.root.path().join("layout.json"))?)?;
+    let limits: Limits =
+        serde_json::from_slice(&std::fs::read(c.root.path().join("limits.json"))?)?;
+    if let Some(proof) = &proof {
+        proof.verify(&layout, &limits)?;
+    }
+    Ok(proof)
+}
 
 pub(super) fn close_and_restart(
     cluster: &Arc<Mutex<Cluster>>,
@@ -12,6 +40,14 @@ pub(super) fn close_and_restart(
     let mut c = cluster.lock().unwrap();
     let catalog = c.elect(1, 10, &[1, 2, 3])?;
     let source = c.elect(4, 20, &[4, 5, 6])?;
+    for id in 4..=6 {
+        assert!(!c
+            .root
+            .path()
+            .join(format!("catalog-key-{id}.pkcs8"))
+            .exists());
+    }
+    assert!(proof(&mut c, source, 600_009, table)?.is_none());
     c.command(
         catalog,
         Request::Query(
@@ -58,6 +94,9 @@ pub(super) fn close_and_restart(
     let before = c.completions.len();
     c.completions.retain(|c| c.request != 600_002);
     assert_eq!(before - c.completions.len(), 1);
+    let original_proof =
+        proof(&mut c, source, 600_010, table)?.context("committed closure proof missing")?;
+    assert_eq!(original_proof.closed().permit.closure(), permit.closure());
     let original_index = stale.selection().range.records.last().unwrap().0;
     let Response::Image(_, _, payload_bytes) = c
         .peers
@@ -79,6 +118,10 @@ pub(super) fn close_and_restart(
         c.start(id)?;
     }
     let source = c.elect(4, 20, &[4, 5, 6])?;
+    let recovered_proof =
+        proof(&mut c, source, 600_011, table)?.context("proof lost after all peers restarted")?;
+    assert_eq!(recovered_proof.closed(), original_proof.closed());
+    assert_ne!(recovered_proof.source_cut(), original_proof.source_cut());
     c.command(
         source,
         Request::Query(20, 600_003, Query::SourceClosure { table }),
diff --git a/docs/design/wal-source-closure-proofs.md b/docs/design/wal-source-closure-proofs.md
new file mode 100644
index 0000000..6a76478
--- /dev/null
+++ b/docs/design/wal-source-closure-proofs.md
@@ -0,0 +1,61 @@
+# Quorum-confirmed source closure proofs
+
+Tracking: [#378](https://github.com/lance-format/lance-context/issues/378).
+
+A catalog `ClosurePermit` authorizes closing an exact source pin after a durable
+publication outcome. It does not establish that the source committed that closure.
+`Query::ClosedProof { table }` returns the latest source closure after a safe
+ReadIndex barrier, signed with the source group's own Ed25519 key. Before closure
+it returns `None`. An isolated former leader cannot refresh a proof from local
+state. Missing signers are rejected before reserving read capacity.
+
+The proof binds the cluster, layout, immutable policy, source read cut, original
+closure index and pin index, and the complete catalog-signed closure permit.
+Verification checks the source signature, nested catalog signature and their
+scope. Source log positions are not compared with catalog log positions. A
+checkpoint and restart preserve the original closure; a later read can attest
+that same closure at a newer source cut.
+
+## Provisioning and compatibility
+
+Set `Limits::source_keys` to distinct public keys for every configured group,
+including groups that also host the catalog. Each key must differ from the catalog
+key. Provision the corresponding private key only to that group's replicas using
+`Node::set_source_signer`. Source-only nodes do not need the catalog private key.
+Splitting a node into group actors retains only the appropriate source signer in
+each actor. Private key holders are neither serializable nor printable.
+
+An empty `source_keys` map disables these queries and preserves the exact v9
+replication policy identity. Existing JSON configurations may omit the field.
+Enabling or changing source keys uses a different immutable policy identity;
+existing directories reject a mismatched policy. There is no online key rotation
+or existing-directory migration in this change. The application checkpoint state
+has not changed. This compatibility guarantee concerns persisted policy identity,
+not arbitrary external bincode serialization of the public `Limits` struct.
+
+These are attestations issued by trusted replicas after their consensus read
+barrier, not threshold signatures or Byzantine quorum certificates. Deployment
+must protect private keys and authenticate transport independently.
+
+## Consumer boundary
+
+Call `ClosedProof::verify` with the expected layout and policy, then match the
+expected table, transaction, plan, source range and publication outcome before
+using the statement. A previously issued valid proof remains a statement about
+that exact closure; it does not establish that no later publication exists.
+
+This increment does not advance catalog checkpoints, grant publisher authority,
+or authorize deletion of Lance objects or log files. Catalog advancement must
+separately require the exact closure from every selected source. Physical
+reclamation still requires the corresponding durable checkpoint and file-lifetime
+rules.
+
+## Validation
+
+Rust tests cover independent v9 policy-byte compatibility, key-policy rejection,
+group-actor key isolation, missing-signer admission, proof tampering, an isolated
+source leader, and checkpoint/restart preservation of the original closure.
+The actual TCP child-process and Lance publication tests also request and verify
+the source proof after publication/closure and after every replica restarts, while
+ensuring source-only replicas receive no catalog private key. These tests do not
+establish throughput or production key-management behavior.