fix(capture): validate checkpoint cut against captured committed log

crates/capture-authority/src/application.rs

@@ -39,7 +39,7 @@ use crate::{
     LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
     next_index, number,
 };
-use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition, TableHandle};
 use std::collections::BTreeMap;
 
 pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
@@ -89,6 +89,40 @@ fn decode_operation(b: &[u8]) -> Result<CreatedOperation, StoreError> {
         result,
     })
 }
+fn decode_provenance(b: &[u8]) -> Result<CreatedOperationProvenance, StoreError> {
+    if b.len() > MAX_OBJECT_KEY_LENGTH + 512 || b.get(..5) != Some(b"COPR\x01") {
+        return Err(StoreError::CorruptRecord);
+    }
+    let encoded_id = b.get(5..29).ok_or(StoreError::CorruptRecord)?;
+    if encoded_id == [0; 24] {
+        return Err(StoreError::CorruptRecord);
+    }
+    Ok(CreatedOperationProvenance {
+        first_decision_log_id: crate::decode_id(encoded_id)?,
+        operation: decode_operation(b.get(29..).ok_or(StoreError::CorruptRecord)?)?,
+    })
+}
+
+pub(crate) fn validate_provenance_encoding(bytes: &[u8]) -> Result<(), StoreError> {
+    decode_provenance(bytes).map(|_| ())
+}
+
+fn decode_stored_operation(b: &[u8], schema: u64) -> Result<CreatedOperation, StoreError> {
+    if matches!(schema, 5 | 6) {
+        Ok(decode_provenance(b)?.operation)
+    } else {
+        decode_operation(b)
+    }
+}
+
+fn encode_provenance(first: LogId, operation: &[u8]) -> Result<Vec<u8>, StoreError> {
+    let mut bytes = b"COPR\x01".to_vec();
+    bytes.extend(encode_id(first));
+    bytes.extend(operation);
+    decode_provenance(&bytes)?;
+    Ok(bytes)
+}
+
 fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     let mut d = Decoder::new(b);
     let r = CreatedApplyResult {
@@ -98,7 +132,7 @@ fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     d.finish()?;
     Ok(r)
 }
-fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
+pub(crate) fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
     let b = meta.get("capture_binding")?.ok_or(StoreError::BindingNotPinned)?;
     let mut d = Decoder::new(b.value());
     let binding = d.binding()?;
@@ -176,11 +210,12 @@ fn apply_created(
     command: &DecideCreated,
     canonical: &[u8],
     binding: &CaptureBinding,
+    schema: u64,
 ) -> Result<CreatedResult, StoreError> {
     let mut operations = tx.open_table(OPERATIONS)?;
     let existing = operations
         .get(command.operation_id.as_slice())?
-        .map(|v| decode_operation(v.value()))
+        .map(|v| decode_stored_operation(v.value(), schema))
         .transpose()?;
     let result = if let Some(existing) = existing {
         if existing.canonical_command == canonical {
@@ -233,6 +268,9 @@ fn apply_created(
             .to_vec();
         op.extend(canonical);
         op.extend(result_bytes(&outcome));
+        if matches!(schema, 5 | 6) {
+            op = encode_provenance(id, &op)?;
+        }
         operations.insert(command.operation_id.as_slice(), op.as_slice())?;
         outcome
     };
@@ -298,11 +336,16 @@ fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated,
 }
 
 pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    let meta = tx.open_table(META)?;
-    if number(required(meta.get("schema")?)?.value())? != 4 {
+    native_application_metadata(&tx.open_table(META)?)
+}
+
+fn native_application_metadata(
+    meta: &impl ReadableTable<&'static str, &'static [u8]>,
+) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
+    if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
         return Err(StoreError::UnsupportedSchema);
     }
-    let last = read_applied(&meta)?;
+    let last = read_applied(meta)?;
     let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
         .map_err(|_| StoreError::CorruptRecord)?;
     if membership
@@ -315,8 +358,7 @@ pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<L
 }
 
 impl Store {
-    /// Explicitly opt an empty, bound store into native entry persistence.
-    pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
+    pub fn migrate_legacy_created_to_native(&mut self) -> Result<(), StoreError> {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
@@ -324,8 +366,66 @@ impl Store {
             if !matches!(schema, 3 | 4) {
                 return Err(StoreError::UnsupportedSchema);
             }
-            read_binding(&meta)?;
+            let mut logs = tx.open_table(LOGS)?;
+            let heads = tx.open_table(HEADS)?;
+            let versions = tx.open_table(VERSIONS)?;
+            let events = tx.open_table(EVENTS)?;
+            let operations = tx.open_table(OPERATIONS)?;
+            let mut results = tx.open_table(APPLY_RESULTS)?;
+            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?;
             if schema == 4 {
+                return Ok(());
+            }
+            if meta.get("native_membership")?.is_some() {
+                return Err(StoreError::CorruptRecord);
+            }
+            // Validate the entire retained suffix before replacing any record.
+            let mut native_logs = Vec::new();
+            for row in logs.iter()? {
+                let (index, bytes) = row?;
+                let entry = decode_log(bytes.value())?;
+                let command = DecideCreated::decode(&entry.payload)?;
+                if entry.id.index != index.value() || command.encode()? != entry.payload {
+                    return Err(StoreError::CorruptRecord);
+                }
+                let native = crate::CaptureRaftEntry {
+                    log_id: entry.id.into(),
+                    payload: openraft::EntryPayload::Normal(crate::CaptureCommand::try_created(command)?),
+                };
+                native_logs.push((index.value(), encode_native_entry(&native)?));
+            }
+            let mut native_results = Vec::new();
+            for row in results.iter()? {
+                let (index, bytes) = row?;
+                let response = crate::CaptureResponse::created(decode_apply(bytes.value())?)?;
+                native_results.push((index.value(), encode_response(&response)?));
+            }
+            for (index, bytes) in native_logs {
+                logs.insert(index, bytes.as_slice())?;
+            }
+            for (index, bytes) in native_results {
+                results.insert(index, bytes.as_slice())?;
+            }
+            let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
+            meta.insert("native_membership", membership.encode()?.as_slice())?;
+            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
+            // Replay native decisions inside the same durable writer boundary.
+            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?;
+        }
+        self.commit(tx)
+    }
+
+    /// Explicitly opt an empty, bound store into native entry persistence.
+    pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 3..=6) {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            read_binding(&meta)?;
+            if matches!(schema, 4 | 6) {
                 let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
                     .map_err(|_| StoreError::CorruptRecord)?;
                 let last = read_applied(&meta)?;
@@ -354,7 +454,7 @@ impl Store {
             }
             let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
             meta.insert("native_membership", membership.encode()?.as_slice())?;
-            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
+            meta.insert("schema", (if schema == 5 { 6u64 } else { 4u64 }).to_be_bytes().as_slice())?;
         }
         self.commit(tx)
     }
@@ -362,7 +462,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 4 {
+            if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
                 return Err(StoreError::UnsupportedSchema);
             }
             if entries.is_empty() {
@@ -391,7 +491,7 @@ impl Store {
     }
     pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         tx.open_table(LOGS)?
@@ -408,14 +508,14 @@ impl Store {
     /// Read typed native entries within the requested bounds.
     pub fn read_native_range<R: std::ops::RangeBounds<u64>>(&self, range: R) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         read_native_range_in(&tx, range)
     }
     pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         if start > end {
@@ -429,7 +529,8 @@ impl Store {
         let response;
         {
             let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 4 {
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 4 | 6) {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -457,7 +558,7 @@ impl Store {
                 openraft::EntryPayload::Blank => crate::CaptureResponse::blank(id)?,
                 openraft::EntryPayload::Normal(command) => {
                     let canonical = command.created().encode()?;
-                    let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
+                    let result = apply_created(&tx, id, command.created(), &canonical, &binding, schema)?;
                     crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
                 }
                 openraft::EntryPayload::Membership(membership) => {
@@ -477,7 +578,7 @@ impl Store {
     }
     pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
         let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         let results = tx.open_table(APPLY_RESULTS)?;
@@ -495,7 +596,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+            if matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
                 if read_binding(&meta)? == binding {
                     return Ok(());
                 }
@@ -524,7 +625,8 @@ impl Store {
         let result;
         {
             let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 3 {
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 3 | 5) {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -547,7 +649,7 @@ impl Store {
                 return Err(StoreError::InvalidApplySequence);
             }
             let command = DecideCreated::decode(&entry.payload)?;
-            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
+            result = apply_created(&tx, id, &command, &entry.payload, &binding, schema)?;
             let mut applied = encode_id(id);
             applied.extend(result_bytes(&result));
             results.insert(id.index, applied.as_slice())?;
@@ -561,7 +663,7 @@ impl Store {
     pub fn capture_binding(&self) -> Result<Option<CaptureBinding>, StoreError> {
         let tx = self.database.begin_read()?;
         let meta = tx.open_table(META)?;
-        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
             return Ok(None);
         }
         Ok(Some(read_binding(&meta)?))
@@ -569,7 +671,7 @@ impl Store {
     pub fn last_applied(&self) -> Result<Option<LogId>, StoreError> {
         let tx = self.database.begin_read()?;
         let meta = tx.open_table(META)?;
-        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
             return Ok(None);
         }
         read_applied(&meta)
@@ -598,15 +700,87 @@ impl Store {
         let t = tx.open_table(EVENTS)?;
         t.get(operation.as_slice())?.map(|v| decode_event(v.value())).transpose()
     }
+    /// Read the persisted first decision, without fabricating legacy provenance.
+    pub fn created_operation_provenance(&self, operation: [u8; 16]) -> Result<Option<CreatedOperationProvenance>, StoreError> {
+        let tx = self.database.begin_read()?;
+        let schema = number(required(tx.open_table(META)?.get("schema")?)?.value())?;
+        if !matches!(schema, 5 | 6) {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        tx.open_table(OPERATIONS)?
+            .get(operation.as_slice())?
+            .map(|value| decode_provenance(value.value()))
+            .transpose()
+    }
+
+    /// Explicitly promote a fully verified legacy or native Created history.
+    pub fn promote_created_provenance(&mut self) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 3..=6) {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            let tables = tx
+                .list_tables()?
+                .map(|table| table.name().to_owned())
+                .collect::<std::collections::BTreeSet<_>>();
+            if [
+                "meta",
+                "vote",
+                "logs",
+                "heads",
+                "versions",
+                "events",
+                "operations",
+                "apply_results",
+            ]
+            .iter()
+            .any(|name| !tables.contains(*name))
+            {
+                return Err(StoreError::CorruptRecord);
+            }
+            crate::validate_identity(crate::decode_identity(required(meta.get("identity")?)?.value())?)?;
+            for row in tx.open_table(crate::VOTE)?.iter()? {
+                let (key, value) = row?;
+                if key.value() != 0 {
+                    return Err(StoreError::CorruptRecord);
+                }
+                crate::decode_vote(value.value())?;
+            }
+            let logs = tx.open_table(LOGS)?;
+            let heads = tx.open_table(HEADS)?;
+            let versions = tx.open_table(VERSIONS)?;
+            let events = tx.open_table(EVENTS)?;
+            let mut operations = tx.open_table(OPERATIONS)?;
+            let results = tx.open_table(APPLY_RESULTS)?;
+            let first = validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, true)?;
+            if matches!(schema, 5 | 6) {
+                return Ok(());
+            }
+            // Keep one encoded row at a time; the replay map contains only first LogIds.
+            for (operation, id) in first {
+                let original = required(operations.get(operation.as_slice())?)?.value().to_vec();
+                operations.insert(operation.as_slice(), encode_provenance(id, &original)?.as_slice())?;
+            }
+            meta.insert("schema", (if schema == 3 { 5u64 } else { 6u64 }).to_be_bytes().as_slice())?;
+        }
+        self.commit(tx)
+    }
+
     pub fn created_operation(&self, operation: [u8; 16]) -> Result<Option<CreatedOperation>, StoreError> {
         let tx = self.database.begin_read()?;
+        let schema = number(required(tx.open_table(META)?.get("schema")?)?.value())?;
         let t = tx.open_table(OPERATIONS)?;
-        t.get(operation.as_slice())?.map(|v| decode_operation(v.value())).transpose()
+        t.get(operation.as_slice())?
+            .map(|v| decode_stored_operation(v.value(), schema))
+            .transpose()
     }
     pub fn created_apply_result(&self, index: u64) -> Result<Option<CreatedApplyResult>, StoreError> {
         let tx = self.database.begin_read()?;
         let t = tx.open_table(APPLY_RESULTS)?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
+        if matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             t.get(index)?
                 .map(|v| {
                     decode_response(v.value()).map(|r| {
@@ -626,26 +800,72 @@ impl Store {
 
 pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
     let meta = tx.open_table(META)?;
-    let schema = number(required(meta.get("schema")?)?.value())?;
-    let native = schema == 4;
-    if !matches!(schema, 3 | 4) {
+    if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
         return Ok(());
     }
-    let binding = read_binding(&meta).map_err(|error| match error {
+    validate_application_tables(
+        &tx.open_table(META)?,
+        &tx.open_table(LOGS)?,
+        &tx.open_table(HEADS)?,
+        &tx.open_table(VERSIONS)?,
+        &tx.open_table(EVENTS)?,
+        &tx.open_table(OPERATIONS)?,
+        &tx.open_table(APPLY_RESULTS)?,
+        false,
+    )
+    .map(|_| ())
+}
+
+fn validate_application_tables(
+    meta: &impl ReadableTable<&'static str, &'static [u8]>,
+    logs: &impl ReadableTable<u64, &'static [u8]>,
+    heads: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    versions: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    events: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    operations: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    results: &impl ReadableTable<u64, &'static [u8]>,
+    promoting_provenance: bool,
+) -> Result<BTreeMap<[u8; 16], LogId>, StoreError> {
+    let schema = number(required(meta.get("schema")?)?.value())?;
+    let native = matches!(schema, 4 | 6);
+    let full_replay = native || schema == 5 || promoting_provenance;
+    if !matches!(schema, 3..=6) {
+        return Ok(BTreeMap::new());
+    }
+    let binding = read_binding(meta).map_err(|error| match error {
         StoreError::BindingNotPinned => StoreError::CorruptRecord,
         other => other,
     })?;
-    let last = read_applied(&meta)?;
-    let logs = tx.open_table(LOGS)?;
-    let committed = validate_committed(&meta, &logs)?;
-    let heads = tx.open_table(HEADS)?;
-    let versions = tx.open_table(VERSIONS)?;
-    let events = tx.open_table(EVENTS)?;
-    let operations = tx.open_table(OPERATIONS)?;
-    let results = tx.open_table(APPLY_RESULTS)?;
+    let last = read_applied(meta)?;
+    let committed = validate_committed(meta, logs)?;
     if last.is_some_and(|last| committed.is_none_or(|c| last.index > c.index)) {
         return Err(StoreError::CorruptRecord);
     }
+    if !native && meta.get("native_membership")?.is_some() {
+        return Err(StoreError::CorruptRecord);
+    }
+    let mut log_next = Some(0);
+    let mut tail = None;
+    for row in logs.iter()? {
+        let (index, bytes) = row?;
+        let entry = decode_log(bytes.value())?;
+        if log_next != Some(index.value()) || entry.id.index != index.value() {
+            return Err(StoreError::CorruptRecord);
+        }
+        if native {
+            decode_native_entry(entry.clone())?;
+        } else if promoting_provenance || schema == 5 {
+            let command = DecideCreated::decode(&entry.payload)?;
+            if command.encode()? != entry.payload {
+                return Err(StoreError::CorruptRecord);
+            }
+        }
+        log_next = index.value().checked_add(1);
+        tail = Some(entry.id);
+    }
+    if meta.get("tail")?.map(|value| crate::decode_id(value.value())).transpose()? != tail {
+        return Err(StoreError::CorruptRecord);
+    }
     let mut next = Some(0);
     let mut observed = None;
     let mut first_operations = BTreeMap::new();
@@ -688,14 +908,14 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         }
         let c = entry_command(&entry, native)?;
         let canonical = c.encode()?;
-        let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
+        let op = decode_stored_operation(required(operations.get(c.operation_id.as_slice())?)?.value(), schema)?;
         let canonical_matches = op.canonical_command == canonical;
         let expected = if canonical_matches {
             op.result
         } else {
             CreatedResult::OperationConflict
         };
-        if native {
+        if full_replay {
             let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
                 if *previous == canonical {
                     result.clone()
@@ -721,6 +941,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
                         decision_log_id: r.log_id,
                     }
                 };
+                first_operations.insert(c.operation_id, r.log_id);
                 replay_operations.insert(c.operation_id, (canonical.clone(), result.clone()));
                 result
             };
@@ -731,7 +952,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         if expected != r.result {
             return Err(StoreError::CorruptRecord);
         }
-        if canonical_matches {
+        if !full_replay && canonical_matches {
             first_operations.entry(c.operation_id).or_insert(r.log_id);
         }
         observed = Some(r.log_id);
@@ -740,8 +961,8 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
     if observed != last {
         return Err(StoreError::CorruptRecord);
     }
-    if native {
-        if native_application_state(tx)?.1 != expected_membership
+    if full_replay {
+        if (native && native_application_metadata(meta)?.1 != expected_membership)
             || heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
             || versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
         {
@@ -755,12 +976,15 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
     }
     for row in operations.iter()? {
         let (k, v) = row?;
-        let op = decode_operation(v.value())?;
+        let op = decode_stored_operation(v.value(), schema)?;
         let c = DecideCreated::decode(&op.canonical_command)?;
         if k.value() != c.operation_id {
             return Err(StoreError::CorruptRecord);
         }
         let first = required(first_operations.get(&c.operation_id))?;
+        if matches!(schema, 5 | 6) && decode_provenance(v.value())?.first_decision_log_id != *first {
+            return Err(StoreError::CorruptRecord);
+        }
         if let CreatedResult::Created {
             version_id,
             event_id,
@@ -795,7 +1019,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         if e.binding != binding {
             return Err(StoreError::CorruptRecord);
         }
-        let op = decode_operation(required(operations.get(k.value())?)?.value())?;
+        let op = decode_stored_operation(required(operations.get(k.value())?)?.value(), schema)?;
         let c = DecideCreated::decode(&op.canonical_command)?;
         if c.object != e.object
             || c.prepared != e.prepared
@@ -858,5 +1082,5 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         d.finish()?;
         decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
     }
-    Ok(())
+    Ok(first_operations)
 }

crates/capture-authority/src/committed.rs

@@ -8,13 +8,13 @@ pub(crate) fn validate_committed(
     let schema = number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?;
     let marker = meta.get("committed")?;
     match (schema, marker) {
-        (1 | 3 | 4, None) => {
-            if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
+        (1 | 3 | 4 | 5 | 6, None) => {
+            if matches!(schema, 3..=6) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
                 return Err(StoreError::CorruptRecord);
             }
             Ok(None)
         }
-        (2 | 3 | 4, Some(marker)) => {
+        (2 | 3 | 4 | 5 | 6, Some(marker)) => {
             let id = decode_id(marker.value())?;
             let record = logs.get(id.index)?.ok_or(StoreError::CorruptRecord)?;
             if decode_log(record.value())?.id != id {
@@ -54,7 +54,7 @@ impl Store {
                 }
             }
             meta.insert("committed", encode_id(id).as_slice())?;
-            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3 | 4) {
+            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3..=6) {
                 meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
             }
         }

crates/capture-authority/src/created.rs

@@ -60,6 +60,11 @@ pub struct CreatedOperation {
     pub result: CreatedResult,
 }
 #[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedOperationProvenance {
+    pub operation: CreatedOperation,
+    pub first_decision_log_id: LogId,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct CreatedApplyResult {
     pub log_id: LogId,
     pub result: CreatedResult,

crates/capture-authority/src/native_snapshot.rs

@@ -0,0 +1,154 @@
+//! Owned consistent native application snapshot view.
+use crate::{Store, StoreError, application, snapshot};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata};
+use std::{
+    io::Write,
+    sync::{Arc, Mutex},
+};
+
+#[cfg(test)]
+std::thread_local! {
+    static AFTER_CAPTURE: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
+}
+
+#[cfg(test)]
+pub(crate) struct CaptureHook;
+#[cfg(test)]
+impl Drop for CaptureHook {
+    fn drop(&mut self) {
+        AFTER_CAPTURE.with(|hook| {
+            hook.borrow_mut().take();
+        });
+    }
+}
+#[cfg(test)]
+pub(crate) fn after_capture(hook: impl FnOnce() + 'static) -> CaptureHook {
+    AFTER_CAPTURE.with(|slot| {
+        assert!(slot.borrow().is_none(), "one capture hook per test thread");
+        *slot.borrow_mut() = Some(Box::new(hook));
+    });
+    CaptureHook
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum NativeSnapshotError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error(transparent)]
+    Snapshot(#[from] snapshot::SnapshotError),
+    #[error("native snapshot Store lock poisoned")]
+    Poisoned,
+    #[error("native snapshot worker failed")]
+    Worker(#[source] tokio::task::JoinError),
+}
+
+pub struct NativeSnapshotView {
+    // Drop the transaction before the owner so the Database stays open throughout the view.
+    transaction: ReadTransaction,
+    header: snapshot::Header,
+    counts: [u64; 4],
+    schema: u64,
+    _owner: Arc<Mutex<Store>>,
+}
+impl NativeSnapshotView {
+    pub(crate) fn acquire(owner: Arc<Mutex<Store>>) -> Result<Self, NativeSnapshotError> {
+        let transaction = {
+            let store = owner.lock().map_err(|_| NativeSnapshotError::Poisoned)?;
+            if !store.writable {
+                return Err(StoreError::WriteDisabled.into());
+            }
+            store.database.begin_read().map_err(StoreError::from)?
+        };
+        #[cfg(test)]
+        if let Some(hook) = AFTER_CAPTURE.with(|slot| slot.borrow_mut().take()) {
+            hook();
+        }
+        let (applied, membership) = application::native_application_state(&transaction)?;
+        let (header, schema) = {
+            let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
+            let logs = transaction.open_table(crate::LOGS).map_err(StoreError::from)?;
+            let committed = crate::committed::validate_committed(&meta, &logs)?;
+            if let Some(applied) = applied {
+                let committed = committed.ok_or(StoreError::CorruptRecord)?;
+                if applied.index > committed.index || (applied.index == committed.index && applied != committed) {
+                    return Err(StoreError::CorruptRecord.into());
+                }
+            }
+            let identity = meta
+                .get("identity")
+                .map_err(StoreError::from)?
+                .ok_or(StoreError::CorruptRecord)?;
+            let schema = crate::number(
+                meta.get("schema")
+                    .map_err(StoreError::from)?
+                    .ok_or(StoreError::CorruptRecord)?
+                    .value(),
+            )?;
+            (
+                snapshot::Header {
+                    cluster_id: crate::decode_identity(identity.value())?.cluster_id,
+                    applied,
+                    membership,
+                    binding: application::read_binding(&meta)?,
+                },
+                schema,
+            )
+        };
+        let mut counts = [0; 4];
+        for (index, definition) in [
+            application::HEADS,
+            application::VERSIONS,
+            application::EVENTS,
+            application::OPERATIONS,
+        ]
+        .into_iter()
+        .enumerate()
+        {
+            counts[index] = transaction
+                .open_table(definition)
+                .map_err(StoreError::from)?
+                .len()
+                .map_err(StoreError::from)?;
+        }
+        Ok(Self {
+            transaction,
+            header,
+            counts,
+            schema,
+            _owner: owner,
+        })
+    }
+
+    pub fn stream_capability_artifact<W: Write>(
+        self,
+        writer: W,
+        limits: snapshot::Limits,
+    ) -> Result<snapshot::CapabilityArtifact<W>, NativeSnapshotError> {
+        if self.schema != 6 {
+            return Err(StoreError::UnsupportedSchema.into());
+        }
+        let output = snapshot::CapabilityOutput::new(writer, limits.total_bytes);
+        Ok(self.stream(output, limits)?.finish())
+    }
+
+    pub fn stream<W: Write>(self, writer: W, limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
+        let mut encoder = snapshot::Encoder::new(writer, &self.header, self.counts, limits)?;
+        for (definition, section) in [
+            (application::HEADS, snapshot::Section::Heads),
+            (application::VERSIONS, snapshot::Section::Versions),
+            (application::EVENTS, snapshot::Section::Events),
+            (application::OPERATIONS, snapshot::Section::Operations),
+        ] {
+            let table = self.transaction.open_table(definition).map_err(StoreError::from)?;
+            for row in table.iter().map_err(StoreError::from)? {
+                let (key, value) = row.map_err(StoreError::from)?;
+                encoder.row(snapshot::Row {
+                    section,
+                    key: key.value(),
+                    value: value.value(),
+                })?;
+            }
+        }
+        Ok(encoder.finish()?)
+    }
+}

crates/capture-authority/src/snapshot.rs

@@ -38,6 +38,161 @@ pub enum SnapshotError {
     #[error("snapshot IO failed")]
     Io(#[from] std::io::Error),
 }
+/// Fixed capability declaration. Consumers validate these untrusted bytes before decoding.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CapabilityManifest {
+    pub bytes: [u8; 56],
+}
+impl CapabilityManifest {
+    pub fn encode(&self) -> [u8; 56] {
+        self.bytes
+    }
+
+    pub fn decode(bytes: &[u8]) -> Result<Self, SnapshotError> {
+        format(bytes.len() == 56, "capability length")?;
+        format(&bytes[..6] == b"CCAP\x01\x01", "capability version/framing")?;
+        format(bytes[6..14] == 6u64.to_be_bytes(), "capability schema")?;
+        format(bytes[14..16] == [1, 1], "operation codec")?;
+        let mut encoded = [0; 56];
+        encoded.copy_from_slice(bytes);
+        Ok(Self { bytes: encoded })
+    }
+
+    fn payload_length(&self) -> u64 {
+        let mut bytes = [0; 8];
+        bytes.copy_from_slice(&self.bytes[16..24]);
+        u64::from_be_bytes(bytes)
+    }
+}
+
+/// Manifest and payload form one explicit input. This does not grant installation authority.
+pub struct CapabilityArtifact<T> {
+    pub manifest: CapabilityManifest,
+    pub payload: T,
+}
+
+pub(crate) struct CapabilityOutput<W> {
+    io: W,
+    hash: Sha256,
+    used: u64,
+    max: u64,
+}
+impl<W: Write> CapabilityOutput<W> {
+    pub(crate) fn new(io: W, max: u64) -> Self {
+        Self {
+            io,
+            hash: Sha256::new(),
+            used: 0,
+            max,
+        }
+    }
+
+    pub(crate) fn finish(self) -> CapabilityArtifact<W> {
+        let mut bytes = [0; 56];
+        bytes[..6].copy_from_slice(b"CCAP\x01\x01");
+        bytes[6..14].copy_from_slice(&6u64.to_be_bytes());
+        bytes[14..16].copy_from_slice(&[1, 1]);
+        bytes[16..24].copy_from_slice(&self.used.to_be_bytes());
+        bytes[24..].copy_from_slice(&self.hash.finalize());
+        CapabilityArtifact {
+            manifest: CapabilityManifest { bytes },
+            payload: self.io,
+        }
+    }
+}
+impl<W: Write> Write for CapabilityOutput<W> {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        let requested = u64::try_from(bytes.len()).map_err(std::io::Error::other)?;
+        if self.used.checked_add(requested).is_none_or(|n| n > self.max) {
+            return Err(std::io::Error::other("capability payload limit"));
+        }
+        let written = self.io.write(bytes)?;
+        let part = bytes
+            .get(..written)
+            .ok_or_else(|| std::io::Error::other("invalid writer count"))?;
+        let count = u64::try_from(written).map_err(std::io::Error::other)?;
+        self.used = self
+            .used
+            .checked_add(count)
+            .ok_or_else(|| std::io::Error::other("capability length overflow"))?;
+        self.hash.update(part);
+        Ok(written)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.io.flush()
+    }
+}
+
+struct CapabilityInput<R> {
+    io: R,
+    hash: Sha256,
+    used: u64,
+    max: u64,
+}
+impl<R: Read> Read for CapabilityInput<R> {
+    fn read(&mut self, bytes: &mut [u8]) -> std::io::Result<usize> {
+        // At the boundary, permit only the existing decoder's one-byte EOF probe.
+        let remaining = self
+            .max
+            .checked_sub(self.used)
+            .ok_or_else(|| std::io::Error::other("capability length overflow"))?;
+        let requested = u64::try_from(bytes.len()).map_err(std::io::Error::other)?;
+        let allowed = requested.min(if remaining == 0 { 1 } else { remaining });
+        let allowed = usize::try_from(allowed).map_err(std::io::Error::other)?;
+        let read = self.io.read(&mut bytes[..allowed])?;
+        let part = bytes
+            .get(..read)
+            .ok_or_else(|| std::io::Error::other("invalid reader count"))?;
+        let count = u64::try_from(read).map_err(std::io::Error::other)?;
+        let used = self
+            .used
+            .checked_add(count)
+            .ok_or_else(|| std::io::Error::other("capability length overflow"))?;
+        if used > self.max {
+            return Err(std::io::Error::other("capability payload limit"));
+        }
+        self.used = used;
+        self.hash.update(part);
+        Ok(read)
+    }
+}
+
+struct CapabilitySink<'a, S>(&'a mut S);
+impl<S: StagingSink> StagingSink for CapabilitySink<'_, S> {
+    fn row(&mut self, section: Section, key: &[u8], value: &[u8]) -> std::io::Result<()> {
+        if section == Section::Operations {
+            crate::application::validate_provenance_encoding(value).map_err(std::io::Error::other)?;
+        }
+        self.0.row(section, key, value)
+    }
+}
+
+/// Validates framing, declared capability and operation encoding into isolated staging.
+/// Failure may leave partial staging rows, which the caller must discard.
+/// Success does not prove cross-table consistency, historical truth or source authority.
+pub fn decode_capability_artifact(
+    artifact: CapabilityArtifact<impl Read>,
+    sink: &mut impl StagingSink,
+    limits: Limits,
+) -> Result<Header, SnapshotError> {
+    let manifest = CapabilityManifest::decode(&artifact.manifest.bytes)?;
+    let length = manifest.payload_length();
+    limit(length <= limits.total_bytes, "capability payload")?;
+    let mut reader = CapabilityInput {
+        io: artifact.payload,
+        hash: Sha256::new(),
+        used: 0,
+        max: length,
+    };
+    let header = decode(&mut reader, &mut CapabilitySink(sink), limits)?;
+    format(reader.used == length, "capability payload length")?;
+    if reader.hash.finalize().as_slice() != &manifest.bytes[24..] {
+        return Err(SnapshotError::Digest);
+    }
+    Ok(header)
+}
+
 /// Row values remain opaque. Operation values will contain the canonical decision
 /// LogId followed by the existing canonical operation bytes, validated by callers.
 pub struct Row<'a> {
@@ -167,6 +322,130 @@ fn row_lengths(k: u64, v: u64, limits: Limits) -> Result<u64, SnapshotError> {
     limit(v <= limits.value_bytes, "value bytes")?;
     k.checked_add(v).ok_or(SnapshotError::Limit("record bytes"))
 }
+/// Incremental encoder for isolated snapshot output. Completion verifies framing,
+/// not persistence or installation. A row error permanently fences the writer.
+pub struct Encoder<W: Write> {
+    out: Output<W>,
+    counts: [u64; 4],
+    section: usize,
+    written: u64,
+    previous: Option<Vec<u8>>,
+    limits: Limits,
+    failed: bool,
+}
+impl<W: Write> Encoder<W> {
+    pub fn new(writer: W, header: &Header, counts: [u64; 4], limits: Limits) -> Result<Self, SnapshotError> {
+        header_valid(header)?;
+        let count = counts
+            .iter()
+            .try_fold(0u64, |n, c| n.checked_add(*c).ok_or(SnapshotError::Limit("records")))?;
+        limit(count <= limits.records, "records")?;
+        format(header.applied.is_some() || count == 0, "rows without applied cut")?;
+        membership_size(header, limits)?;
+        let membership = header.membership.encode().map_err(|_| SnapshotError::Format("membership"))?;
+        let membership_len = u64::try_from(membership.len()).map_err(|_| SnapshotError::Limit("membership"))?;
+        limit(membership_len <= limits.membership_bytes, "membership")?;
+        let mut out = Output {
+            io: writer,
+            hash: Sha256::new(),
+            used: 0,
+            max: limits.total_bytes,
+        };
+        out.bytes(MAGIC)?;
+        out.bytes(&header.cluster_id)?;
+        match header.applied {
+            None => out.bytes(&[0])?,
+            Some(id) => {
+                out.bytes(&[1])?;
+                for n in [id.term, id.leader_node, id.index] {
+                    out.number(n)?;
+                }
+            }
+        }
+        out.number(membership_len)?;
+        out.bytes(&membership)?;
+        out.number(header.binding.revision)?;
+        out.bytes(&header.binding.binding_id)?;
+        out.bytes(&header.binding.target_id)?;
+        out.bytes(&[1])?;
+        out.number(counts[0])?;
+        let mut encoder = Self {
+            out,
+            counts,
+            section: 0,
+            written: 0,
+            previous: None,
+            limits,
+            failed: false,
+        };
+        encoder.advance_sections()?;
+        Ok(encoder)
+    }
+
+    fn advance_sections(&mut self) -> Result<(), SnapshotError> {
+        while self.section < SECTIONS.len() && self.written == self.counts[self.section] {
+            self.section += 1;
+            self.written = 0;
+            self.previous = None;
+            if self.section < SECTIONS.len() {
+                self.out
+                    .bytes(&[u8::try_from(self.section + 1).map_err(|_| SnapshotError::Format("section"))?])?;
+                self.out.number(self.counts[self.section])?;
+            }
+        }
+        Ok(())
+    }
+
+    pub fn row(&mut self, row: Row<'_>) -> Result<(), SnapshotError> {
+        format(!self.failed, "encoder failed")?;
+        let result = self.write_row(row);
+        if result.is_err() {
+            self.failed = true;
+        }
+        result
+    }
+
+    fn write_row(&mut self, row: Row<'_>) -> Result<(), SnapshotError> {
+        format(self.section < SECTIONS.len(), "record count")?;
+        format(row.section == SECTIONS[self.section], "section order")?;
+        let k = u64::try_from(row.key.len()).map_err(|_| SnapshotError::Limit("key bytes"))?;
+        let v = u64::try_from(row.value.len()).map_err(|_| SnapshotError::Limit("value bytes"))?;
+        let length = row_lengths(k, v, self.limits)?
+            .checked_add(16)
+            .ok_or(SnapshotError::Limit("record bytes"))?;
+        // Reserve the still-unwritten section framing as well as the footer.
+        let framing = u64::try_from(SECTIONS.len() - self.section - 1)
+            .map_err(|_| SnapshotError::Limit("total bytes"))?
+            .checked_mul(9)
+            .ok_or(SnapshotError::Limit("total bytes"))?;
+        limit(
+            self.out
+                .used
+                .checked_add(length)
+                .and_then(|n| n.checked_add(framing))
+                .and_then(|n| n.checked_add(32))
+                .is_some_and(|n| n <= self.out.max),
+            "total bytes",
+        )?;
+        format(self.previous.as_ref().is_none_or(|p| p.as_slice() < row.key), "key order")?;
+        self.out.number(k)?;
+        self.out.number(v)?;
+        self.out.bytes(row.key)?;
+        self.out.bytes(row.value)?;
+        self.previous = Some(row.key.to_vec());
+        self.written += 1;
+        self.advance_sections()
+    }
+
+    pub fn finish(self) -> Result<W, SnapshotError> {
+        format(!self.failed, "encoder failed")?;
+        format(self.section == SECTIONS.len(), "record count")?;
+        let mut out = self.out;
+        out.io.write_all(&out.hash.finalize())?;
+        Ok(out.io)
+    }
+}
+
 pub fn encode<'a>(
     writer: impl Write,
     header: &Header,
@@ -174,68 +453,11 @@ pub fn encode<'a>(
     rows: impl IntoIterator<Item = Row<'a>>,
     limits: Limits,
 ) -> Result<(), SnapshotError> {
-    header_valid(header)?;
-    let count = counts
-        .iter()
-        .try_fold(0u64, |n, c| n.checked_add(*c).ok_or(SnapshotError::Limit("records")))?;
-    limit(count <= limits.records, "records")?;
-    format(header.applied.is_some() || count == 0, "rows without applied cut")?;
-    membership_size(header, limits)?;
-    let membership = header.membership.encode().map_err(|_| SnapshotError::Format("membership"))?;
-    let membership_len = u64::try_from(membership.len()).map_err(|_| SnapshotError::Limit("membership"))?;
-    limit(membership_len <= limits.membership_bytes, "membership")?;
-    let mut out = Output {
-        io: writer,
-        hash: Sha256::new(),
-        used: 0,
-        max: limits.total_bytes,
-    };
-    out.bytes(MAGIC)?;
-    out.bytes(&header.cluster_id)?;
-    match header.applied {
-        None => out.bytes(&[0])?,
-        Some(id) => {
-            out.bytes(&[1])?;
-            for n in [id.term, id.leader_node, id.index] {
-                out.number(n)?;
-            }
-        }
+    let mut encoder = Encoder::new(writer, header, counts, limits)?;
+    for row in rows {
+        encoder.row(row)?;
     }
-    out.number(membership_len)?;
-    out.bytes(&membership)?;
-    out.number(header.binding.revision)?;
-    out.bytes(&header.binding.binding_id)?;
-    out.bytes(&header.binding.target_id)?;
-    let mut rows = rows.into_iter();
-    for (index, section) in SECTIONS.into_iter().enumerate() {
-        out.bytes(&[u8::try_from(index + 1).map_err(|_| SnapshotError::Format("section"))?])?;
-        out.number(counts[index])?;
-        let mut previous: Option<Vec<u8>> = None;
-        for _ in 0..counts[index] {
-            let row = rows.next().ok_or(SnapshotError::Format("record count"))?;
-            format(row.section == section, "section order")?;
-            let k = u64::try_from(row.key.len()).map_err(|_| SnapshotError::Limit("key bytes"))?;
-            let v = u64::try_from(row.value.len()).map_err(|_| SnapshotError::Limit("value bytes"))?;
-            let length = row_lengths(k, v, limits)?
-                .checked_add(16)
-                .ok_or(SnapshotError::Limit("record bytes"))?;
-            limit(
-                out.used
-                    .checked_add(length)
-                    .and_then(|n| n.checked_add(32))
-                    .is_some_and(|n| n <= out.max),
-                "total bytes",
-            )?;
-            format(previous.as_ref().is_none_or(|p| p.as_slice() < row.key), "key order")?;
-            out.number(k)?;
-            out.number(v)?;
-            out.bytes(row.key)?;
-            out.bytes(row.value)?;
-            previous = Some(row.key.to_vec());
-        }
-    }
-    format(rows.next().is_none(), "record count")?;
-    out.io.write_all(&out.hash.finalize())?;
+    encoder.finish()?;
     Ok(())
 }
 pub fn decode(reader: impl Read, sink: &mut impl StagingSink, limits: Limits) -> Result<Header, SnapshotError> {
@@ -403,6 +625,257 @@ mod tests {
         assert!(result.is_ok(), "U22_CODEC_ROUNDTRIP: {result:?}");
         b
     }
+    #[test]
+    fn u23_redb_guard_rows() -> Result<(), Box<dyn std::error::Error>> {
+        use redb::{ReadableDatabase, ReadableTable, TableDefinition};
+        const TABLE: TableDefinition<&str, &str> = TableDefinition::new("u23_rows");
+        let dir = tempfile::tempdir()?;
+        let db = redb::Database::create(dir.path().join("rows.redb"))?;
+        let fixture = [("a", "head"), ("b", "next")];
+        let write = db.begin_write()?;
+        {
+            let mut table = write.open_table(TABLE)?;
+            for (key, value) in fixture {
+                table.insert(key, value)?;
+            }
+        }
+        write.commit()?;
+        let read = db.begin_read()?;
+        let table = read.open_table(TABLE)?;
+        let mut bytes = Vec::new();
+        let result = Encoder::new(&mut bytes, &header(), [2, 0, 0, 0], limits());
+        assert!(result.is_ok(), "U23_STREAM_GUARD: constructor failed");
+        let mut encoder = result?;
+        for row in table.iter()? {
+            let (key, value) = row?;
+            encoder.row(Row {
+                section: Section::Heads,
+                key: key.value().as_bytes(),
+                value: value.value().as_bytes(),
+            })?;
+        }
+        encoder.finish()?;
+        let mut expected = Vec::new();
+        encode(
+            &mut expected,
+            &header(),
+            [2, 0, 0, 0],
+            fixture.into_iter().map(|(key, value)| Row {
+                section: Section::Heads,
+                key: key.as_bytes(),
+                value: value.as_bytes(),
+            }),
+            limits(),
+        )?;
+        assert_eq!(bytes, expected, "U23_STREAM_GUARD: wire changed");
+        let mut sink = Sink::default();
+        assert_eq!(decode(bytes.as_slice(), &mut sink, limits())?, header());
+        assert_eq!(sink.0.len(), fixture.len());
+        for ((section, key, value), (expected_key, expected_value)) in sink.0.iter().zip(fixture) {
+            assert_eq!(*section, Section::Heads);
+            assert_eq!(key, expected_key.as_bytes());
+            assert_eq!(value, expected_value.as_bytes());
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn u23_failure_fence() {
+        use std::{cell::Cell, rc::Rc};
+        struct Recovering {
+            fail: Rc<Cell<bool>>,
+            writes: Rc<Cell<usize>>,
+        }
+        impl Write for Recovering {
+            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+                self.writes.set(self.writes.get() + 1);
+                if self.fail.replace(false) {
+                    Err(io::Error::other("once"))
+                } else {
+                    Ok(bytes.len())
+                }
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let fail = Rc::new(Cell::new(false));
+        let writes = Rc::new(Cell::new(0));
+        let mut encoder = Encoder::new(
+            Recovering {
+                fail: fail.clone(),
+                writes: writes.clone(),
+            },
+            &header(),
+            [1, 0, 0, 0],
+            limits(),
+        )
+        .expect("U23_FAILURE_FENCE: construct");
+        fail.set(true);
+        assert!(matches!(
+            encoder.row(Row {
+                section: Section::Heads,
+                key: b"a",
+                value: b"v"
+            }),
+            Err(SnapshotError::Io(_))
+        ));
+        let after_failure = writes.get();
+        assert!(
+            encoder
+                .row(Row {
+                    section: Section::Heads,
+                    key: b"a",
+                    value: b"v"
+                })
+                .is_err(),
+            "U23_FAILURE_FENCE: recovered after IO error"
+        );
+        assert!(encoder.finish().is_err(), "U23_FAILURE_FENCE: finish after IO error");
+        assert_eq!(writes.get(), after_failure, "U23_FAILURE_FENCE: wrote after error");
+        for invalid in [
+            Row {
+                section: Section::Versions,
+                key: b"a",
+                value: b"v",
+            },
+            Row {
+                section: Section::Heads,
+                key: &[0; 33],
+                value: b"v",
+            },
+        ] {
+            let mut output = Vec::new();
+            let mut encoder = Encoder::new(&mut output, &header(), [1, 0, 0, 0], limits()).expect("construct validation encoder");
+            assert!(encoder.row(invalid).is_err());
+            assert!(
+                encoder
+                    .row(Row {
+                        section: Section::Heads,
+                        key: b"a",
+                        value: b"v"
+                    })
+                    .is_err(),
+                "U23_FAILURE_FENCE: validation retry"
+            );
+            assert!(encoder.finish().is_err(), "U23_FAILURE_FENCE: validation finish");
+        }
+        for key in [b"b".as_slice(), b"a".as_slice()] {
+            let mut output = Vec::new();
+            let mut encoder = Encoder::new(&mut output, &header(), [2, 0, 0, 0], limits()).expect("construct ordered encoder");
+            encoder
+                .row(Row {
+                    section: Section::Heads,
+                    key: b"b",
+                    value: b"v",
+                })
+                .expect("first row");
+            assert!(
+                encoder
+                    .row(Row {
+                        section: Section::Heads,
+                        key,
+                        value: b"v"
+                    })
+                    .is_err()
+            );
+            assert!(encoder.finish().is_err(), "U23_FAILURE_FENCE: key order finish");
+        }
+    }
+
+    #[test]
+    fn u23_finish_counts() {
+        for counts in [[1, 0, 0, 0], [0, 1, 0, 0], [0, 0, 0, 1]] {
+            let mut output = Vec::new();
+            let encoder = Encoder::new(&mut output, &header(), counts, limits()).expect("construct incomplete encoder");
+            assert!(encoder.finish().is_err(), "U23_FINISH_COUNT: incomplete accepted");
+            assert!(
+                decode(output.as_slice(), &mut Sink::default(), limits()).is_err(),
+                "U23_FINISH_COUNT: incomplete footer"
+            );
+        }
+        for counts in [[0, 0, 0, 0], [0, 0, 0, 1]] {
+            let mut output = Vec::new();
+            let mut encoder = Encoder::new(&mut output, &header(), counts, limits()).expect("construct empty-section encoder");
+            if counts[3] == 1 {
+                encoder
+                    .row(Row {
+                        section: Section::Operations,
+                        key: b"o",
+                        value: b"v",
+                    })
+                    .expect("skip empty sections");
+            }
+            encoder.finish().expect("U23_FINISH_COUNT: exact count");
+            decode(output.as_slice(), &mut Sink::default(), limits()).expect("valid finished snapshot");
+        }
+        let mut output = Vec::new();
+        let mut encoder = Encoder::new(&mut output, &header(), [1, 0, 0, 0], limits()).expect("construct extra-row encoder");
+        encoder
+            .row(Row {
+                section: Section::Heads,
+                key: b"a",
+                value: b"v",
+            })
+            .expect("declared row");
+        assert!(
+            encoder
+                .row(Row {
+                    section: Section::Heads,
+                    key: b"b",
+                    value: b"v"
+                })
+                .is_err(),
+            "U23_FINISH_COUNT: extra row"
+        );
+        assert!(encoder.finish().is_err(), "U23_FINISH_COUNT: extra row finish");
+    }
+
+    #[test]
+    fn u23_row_limits() {
+        use std::{cell::Cell, rc::Rc};
+        struct Counting(Rc<Cell<usize>>);
+        impl Write for Counting {
+            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+                self.0.set(self.0.get() + bytes.len());
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let empty = {
+            let mut bytes = Vec::new();
+            encode(&mut bytes, &header(), [0; 4], [], limits()).expect("empty framing length");
+            u64::try_from(bytes.len()).expect("length")
+        };
+        for (bounded, key, value, label) in [
+            (limits(), &[0; 33][..], b"v".as_slice(), "key bytes"),
+            (limits(), b"a".as_slice(), &[0; 65][..], "value bytes"),
+            (
+                Limits {
+                    total_bytes: empty + 17,
+                    ..limits()
+                },
+                b"a".as_slice(),
+                b"v".as_slice(),
+                "total bytes",
+            ),
+        ] {
+            let written = Rc::new(Cell::new(0));
+            let mut encoder =
+                Encoder::new(Counting(written.clone()), &header(), [1, 0, 0, 0], bounded).expect("construct bounded encoder");
+            let before = written.get();
+            assert!(
+                matches!(encoder.row(Row { section: Section::Heads, key, value }), Err(SnapshotError::Limit(name)) if name == label),
+                "U23_ROW_LIMIT: {label}"
+            );
+            assert_eq!(written.get(), before, "U23_ROW_LIMIT: wrote rejected row");
+            assert!(encoder.finish().is_err(), "U23_ROW_LIMIT: failed finish");
+            assert_eq!(written.get(), before, "U23_ROW_LIMIT: wrote footer");
+        }
+    }
+
     // Independent wire fixture: magic, version, hash algorithm, cluster, optional
     // applied ID, length-prefixed complete membership, binding, four sections.
     fn raw(h: &Header, entries: &[(u8, Vec<(Vec<u8>, Vec<u8>)>)]) -> Vec<u8> {

crates/ecstore/src/api/mod.rs

@@ -372,6 +372,15 @@ pub mod config {
             with_server_config_write_lock,
         };
 
+        /// Observe persisted config bytes and their reader-associated metadata within a byte limit.
+        pub async fn read_config_observation(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            max_bytes: usize,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_observation(api, file, max_bytes).await
+        }
+
         /// Read config bytes and metadata without treating an empty payload as absent.
         pub async fn read_config_preserve_empty_with_metadata_opts(
             api: std::sync::Arc<crate::store::ECStore>,

crates/ecstore/src/config/com.rs

@@ -449,6 +449,22 @@ where
     Ok(data)
 }
 
+pub(crate) async fn read_config_observation<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    let opts = ObjectOptions {
+        suppress_read_repair: true,
+        no_lock: false,
+        skip_verify_bitrot: false,
+        ..Default::default()
+    };
+    crate::object_api::without_get_object_body_cache_hook(read_config_limited_preserve_empty_with_metadata_opts(
+        api, file, &opts, max_bytes,
+    ))
+    .await
+}
+
 pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
 where
     S: EcstoreObjectIO,
@@ -5425,6 +5441,166 @@ mod tests {
         assert_eq!(store.write_calls.load(Ordering::SeqCst), 0);
     }
 
+    struct ObservationStream {
+        inner: Cursor<Vec<u8>>,
+        consumed: Arc<AtomicUsize>,
+        fail: bool,
+    }
+
+    impl AsyncRead for ObservationStream {
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            if self.fail {
+                return Poll::Ready(Err(std::io::Error::other("observation stream failed")));
+            }
+            let before = buf.filled().len();
+            let result = Pin::new(&mut self.inner).poll_read(cx, buf);
+            self.consumed.fetch_add(buf.filled().len() - before, Ordering::SeqCst);
+            result
+        }
+    }
+
+    struct ObservationStore {
+        body: Vec<u8>,
+        declared_size: i64,
+        error: Option<Error>,
+        stream_error: bool,
+        consumed: Arc<AtomicUsize>,
+        options: std::sync::Mutex<Option<(bool, bool, bool)>>,
+        writes: AtomicUsize,
+    }
+
+    impl Debug for ObservationStore {
+        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+            f.debug_struct("ObservationStore").finish()
+        }
+    }
+
+    impl ObservationStore {
+        fn new(body: Vec<u8>, declared_size: i64) -> Self {
+            Self {
+                body,
+                declared_size,
+                error: None,
+                stream_error: false,
+                consumed: Arc::new(AtomicUsize::new(0)),
+                options: std::sync::Mutex::new(None),
+                writes: AtomicUsize::new(0),
+            }
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::object::ObjectIO for ObservationStore {
+        type Error = Error;
+        type RangeSpec = HTTPRangeSpec;
+        type HeaderMap = HeaderMap;
+        type ObjectOptions = ObjectOptions;
+        type ObjectInfo = ObjectInfo;
+        type GetObjectReader = GetObjectReader;
+        type PutObjectReader = PutObjReader;
+
+        async fn get_object_reader(
+            &self,
+            bucket: &str,
+            object: &str,
+            range: Option<HTTPRangeSpec>,
+            _headers: HeaderMap,
+            opts: &ObjectOptions,
+        ) -> Result<GetObjectReader> {
+            assert_eq!(bucket, RUSTFS_META_BUCKET);
+            assert_eq!(object, "config/observation.bin");
+            assert!(range.is_none());
+            *self.options.lock().expect("observation options") =
+                Some((opts.suppress_read_repair, opts.no_lock, opts.skip_verify_bitrot));
+            if let Some(error) = &self.error {
+                return Err(error.clone());
+            }
+            Ok(GetObjectReader {
+                stream: Box::new(ObservationStream {
+                    inner: Cursor::new(self.body.clone()),
+                    consumed: self.consumed.clone(),
+                    fail: self.stream_error,
+                }),
+                object_info: ObjectInfo {
+                    size: self.declared_size,
+                    etag: Some("observation-etag".to_string()),
+                    ..Default::default()
+                },
+                buffered_body: None,
+                body_source: Default::default(),
+            })
+        }
+
+        async fn put_object(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _data: &mut PutObjReader,
+            _opts: &ObjectOptions,
+        ) -> Result<ObjectInfo> {
+            self.writes.fetch_add(1, Ordering::SeqCst);
+            Err(Error::other("observation must not write"))
+        }
+    }
+
+    #[tokio::test]
+    async fn observation_preserves_options_and_errors() {
+        for body in [Vec::new(), b"raw config".to_vec()] {
+            let store = Arc::new(ObservationStore::new(body.clone(), i64::try_from(body.len()).expect("fixture size")));
+            let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
+            assert!(result.is_ok(), "observation must preserve empty and nonempty objects: {result:?}");
+            let (observed, info) = result.expect("successful observation");
+            assert_eq!(observed, body);
+            assert_eq!(info.etag.as_deref(), Some("observation-etag"));
+            assert_eq!(
+                *store.options.lock().expect("options"),
+                Some((true, false, false)),
+                "I05_REPAIR_SUPPRESSED"
+            );
+            assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        }
+        for (source, expected) in [
+            (Error::FileNotFound, Error::ConfigNotFound),
+            (
+                Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), "config/observation.bin".to_string()),
+                Error::ConfigNotFound,
+            ),
+            (Error::ErasureReadQuorum, Error::ErasureReadQuorum),
+        ] {
+            let mut fixture = ObservationStore::new(Vec::new(), 0);
+            fixture.error = Some(source);
+            let store = Arc::new(fixture);
+            let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
+            assert_eq!(result.expect_err("read error must propagate"), expected);
+            assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        }
+        let mut fixture = ObservationStore::new(Vec::new(), 0);
+        fixture.stream_error = true;
+        let store = Arc::new(fixture);
+        let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
+        let error = result.expect_err("stream error must propagate");
+        assert!(error.to_string().contains("observation stream failed"));
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test]
+    async fn observation_enforces_byte_limit() {
+        let store = Arc::new(ObservationStore::new(vec![7; 64], 0));
+        let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
+        assert_eq!(result.is_err(), true, "I05_BOUND_ENFORCED");
+        assert_eq!(
+            result.expect_err("bounded observation error"),
+            Error::CorruptedFormat,
+            "I05_BOUND_ENFORCED"
+        );
+        assert_eq!(store.consumed.load(Ordering::SeqCst), 9, "I05_BOUND_ENFORCED: drain exactly limit + 1");
+        let store = Arc::new(ObservationStore::new(vec![7; 8], 8));
+        let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
+        assert!(result.is_ok(), "exact byte limit must succeed: {result:?}");
+        assert_eq!(result.expect("exact limit").0, vec![7; 8]);
+        assert_eq!(store.consumed.load(Ordering::SeqCst), 8);
+    }
+
     #[tokio::test]
     async fn read_config_preserve_empty_distinguishes_empty_object() {
         let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));

crates/ecstore/src/set_disk/mod.rs

@@ -1095,6 +1095,217 @@ mod prepared_get_object_metadata_tests {
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
+    const OBSERVATION_HOOK_BODY_LEN: usize = 1024 * 1024 + 1;
+
+    struct ObservationBodyHook {
+        object: &'static str,
+        calls: Arc<AtomicU64>,
+        body: bytes::Bytes,
+    }
+
+    #[async_trait::async_trait]
+    impl crate::object_api::GetObjectBodyCacheHook for ObservationBodyHook {
+        async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option<bytes::Bytes> {
+            if bucket == crate::disk::RUSTFS_META_BUCKET && object == self.object {
+                assert_eq!(info.size, i64::try_from(self.body.len()).expect("hook fixture size"));
+                self.calls.fetch_add(1, Ordering::SeqCst);
+                Some(self.body.clone())
+            } else {
+                None
+            }
+        }
+    }
+
+    struct ObservationHookGuard;
+
+    impl Drop for ObservationHookGuard {
+        fn drop(&mut self) {
+            crate::object_api::unregister_get_object_body_cache_hook();
+        }
+    }
+
+    async fn ordinary_observation_get(set: &SetDisks, object: &str) -> Vec<u8> {
+        let mut reader = set
+            .get_object_reader(crate::disk::RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("ordinary GET positive control");
+        let mut bytes = Vec::new();
+        reader.stream.read_to_end(&mut bytes).await.expect("ordinary GET body");
+        bytes
+    }
+
+    fn observation_register_hook(object: &'static str, body: Vec<u8>) -> (ObservationHookGuard, Arc<AtomicU64>) {
+        let calls = Arc::new(AtomicU64::new(0));
+        crate::object_api::register_get_object_body_cache_hook(Arc::new(ObservationBodyHook {
+            object,
+            calls: calls.clone(),
+            body: bytes::Bytes::from(body),
+        }));
+        (ObservationHookGuard, calls)
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_real_backend_boundaries() {
+        let object = "config/observation-real.bin";
+        let payload = b"persisted config";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(payload.to_vec());
+        let written = set
+            .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let result = crate::config::com::read_config_observation(set.clone(), object, payload.len()).await;
+        assert_eq!(result.is_ok(), true, "I05_OBSERVATION_SUCCESS: {result:?}");
+        let (bytes, info) = result.expect("real observation success");
+        assert_eq!(bytes, payload);
+        assert_eq!(info.size, written.size);
+        assert_eq!(info.etag, written.etag);
+        assert_eq!(info.data_dir, written.data_dir);
+        assert_eq!(info.name, object);
+        let over = crate::config::com::read_config_observation(set.clone(), object, payload.len() - 1).await;
+        assert_eq!(over.is_err(), true, "I05_BOUND_ENFORCED");
+        assert_eq!(over.expect_err("over limit must fail"), Error::CorruptedFormat, "I05_BOUND_ENFORCED");
+        let absent = crate::config::com::read_config_observation(set.clone(), "config/observation-absent.bin", 32).await;
+        assert_eq!(absent.expect_err("absent config"), Error::ConfigNotFound);
+        let empty_object = "config/observation-empty.bin";
+        let mut empty = PutObjReader::from_vec(Vec::new());
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, empty_object, &mut empty, &ObjectOptions::default())
+            .await
+            .expect("persist empty config");
+        let empty = crate::config::com::read_config_observation(set, empty_object, 0).await;
+        assert!(empty.is_ok(), "existing empty config must succeed: {empty:?}");
+        let (bytes, info) = empty.expect("empty observation");
+        assert!(bytes.is_empty());
+        assert_eq!(info.size, 0);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_bypasses_registered_body_hook() {
+        let object = "config/observation-hook.bin";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
+        let result = crate::config::com::read_config_observation(set, object, OBSERVATION_HOOK_BODY_LEN).await;
+        assert_eq!(result.is_ok(), true, "I05_HOOK_BYPASS: {result:?}");
+        assert_eq!(
+            result.expect("observed persisted body").0,
+            vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
+            "I05_HOOK_BYPASS"
+        );
+        assert_eq!(calls.load(Ordering::SeqCst), 0, "I05_HOOK_BYPASS");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_hook_scope_restores_after_error() {
+        let object = "config/observation-hook-error.bin";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        let result = crate::config::com::read_config_observation(set.clone(), object, OBSERVATION_HOOK_BODY_LEN - 1).await;
+        assert_eq!(result.expect_err("bounded observation error"), Error::CorruptedFormat);
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
+    struct PendingObservationStore {
+        set: Arc<SetDisks>,
+        ready: std::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
+    }
+
+    impl std::fmt::Debug for PendingObservationStore {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            f.debug_struct("PendingObservationStore").finish()
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::object::ObjectIO for PendingObservationStore {
+        type Error = Error;
+        type RangeSpec = HTTPRangeSpec;
+        type HeaderMap = HeaderMap;
+        type ObjectOptions = ObjectOptions;
+        type ObjectInfo = ObjectInfo;
+        type GetObjectReader = GetObjectReader;
+        type PutObjectReader = PutObjReader;
+
+        async fn get_object_reader(
+            &self,
+            bucket: &str,
+            object: &str,
+            range: Option<HTTPRangeSpec>,
+            headers: HeaderMap,
+            opts: &ObjectOptions,
+        ) -> Result<GetObjectReader> {
+            let mut reader = self.set.get_object_reader(bucket, object, range, headers, opts).await?;
+            let mut body = Vec::new();
+            reader.stream.read_to_end(&mut body).await.expect("pending producer real GET");
+            assert_eq!(
+                body,
+                vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
+                "I05_HOOK_BYPASS: producer must enter its own scope"
+            );
+            self.ready
+                .lock()
+                .expect("pending signal")
+                .take()
+                .expect("one GET")
+                .send(())
+                .expect("producer entry signal");
+            std::future::pending::<Result<GetObjectReader>>().await
+        }
+
+        async fn put_object(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _data: &mut PutObjReader,
+            _opts: &ObjectOptions,
+        ) -> Result<ObjectInfo> {
+            assert!(false, "observation must not PUT");
+            Err(Error::other("unexpected observation PUT"))
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_hook_scope_restores_after_cancel() {
+        let object = "config/observation-hook-cancel.bin";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
+        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
+        let adapter = Arc::new(PendingObservationStore {
+            set: set.clone(),
+            ready: std::sync::Mutex::new(Some(ready_tx)),
+        });
+        let mut scope = Box::pin(crate::config::com::read_config_observation(adapter, object, OBSERVATION_HOOK_BODY_LEN));
+        tokio::select! {
+            result = ready_rx => { result.expect("producer reached pending GET"); }
+            result = &mut scope => { assert!(false, "producer must remain pending until cancelled: {result:?}"); }
+        }
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        drop(scope);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
     const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
 
     fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {

docs/architecture/README.md

@@ -32,6 +32,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 
 | Document | Use this when |
 |---|---|
+| [capture-checkpoint-contract.md](capture-checkpoint-contract.md) | changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup |
 | [local-descriptor-cache.md](local-descriptor-cache.md) | changing descriptor cache keys/open modes or invalidation after local mutations |
 | [erasure-coding.md](erasure-coding.md) | changing anything under `crates/ecstore/src/erasure/`, `crates/filemeta/`, `crates/ecstore/src/set_disk/`, storage-class or layout code, or any decode, quorum, or heal boundary (normative spec) |
 | [placement-repair-invariants.md](placement-repair-invariants.md) | changing anything that resolves an object to a pool, set, or disk, or that admits scanner or heal work |

docs/architecture/capture-checkpoint-contract.md

@@ -0,0 +1,211 @@
+**Use this when:** changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup.
+
+# Capture checkpoint contract
+
+This document defines the requirements for a recoverable Capture checkpoint.
+It is a design contract, not evidence that production capture, installation,
+purge, quorum integration, or S3 behavior implements these requirements.
+
+## Existing evidence and the missing boundary
+
+In `crates/capture-authority/src/application.rs`, `apply_created` inserts one
+canonical command and result on the first decision for an operation ID. An
+identical command returns that stored result; a different command returns
+`OperationConflict` without replacing the canonical record. The first result
+remains authoritative even when later state would produce a different result.
+
+`CreatedOperation` in `crates/capture-authority/src/created.rs` contains command
+bytes and a result. Only `CreatedResult::Created` carries `decision_log_id`.
+Negative results carry no first-decision ID. Heads, versions, events, and
+operations therefore cannot reconstruct the source of every first decision.
+For example, a stored `HeadMismatch` records an observed head, but does not
+prove that the head existed at the original decision cut. Later retries do
+not supply the missing historical evidence.
+
+`validate_application` in `crates/capture-authority/src/application.rs` derives
+first operation IDs from retained log/application records and, for native
+schema 4, replays decisions from the beginning using in-memory maps and sets.
+That intact-prefix validator is evidence for existing behavior; it is not a
+checkpoint-baseline validator or a bounded recovery algorithm.
+
+The `Row` comment and `StagingSink` in
+`crates/capture-authority/src/snapshot.rs` reserve operation provenance for
+caller validation and require isolated staging. Rows remain opaque to the
+framing decoder. A framing digest detects byte corruption; it does not establish
+semantic consistency, historical truth, or authority to install a checkpoint.
+
+## Canonical operation provenance
+
+Checkpoint-capable storage must use an explicitly versioned operation envelope
+containing the full first-decision `LogId` and the existing canonical operation
+bytes, including their stored result. Preserve those bytes and result; do not
+recompute a negative result against checkpoint-time heads or binding state.
+The envelope version is a storage capability boundary, distinct from the
+snapshot framing version. The existing CSNP v1 framing and header must not be
+silently extended. Header capability metadata requires a declared framing
+revision or an authenticated, versioned manifest whose consistency with the
+captured header is validated. Distinct storage and framing versions do not
+permit silently changing the v1 format.
+
+The first insertion must atomically record the envelope, application response,
+applied position, and any successful head/version/event changes in the same
+write transaction. Its first-decision ID must identify that insertion's log
+entry. An identical-command retry returns the original result and leaves both
+bytes and first-decision ID unchanged. A conflicting-command retry records its
+own application response as `OperationConflict`; it cannot overwrite the
+canonical envelope or claim to be the first decision.
+
+```mermaid
+flowchart TD
+    A[Committed command and operation ID] --> B{Canonical operation exists?}
+    B -->|No| C[Evaluate once at this log ID]
+    C --> D[Atomically persist first ID, original bytes and result]
+    B -->|Yes| E{Same canonical command bytes?}
+    E -->|Yes| F[Return original result; preserve first ID]
+    E -->|No| G[Return OperationConflict; preserve canonical record]
+```
+
+## Compatibility and promotion
+
+The existing schema 3 to schema 4 migration remains unchanged. Adding provenance
+requires a separately declared storage capability with explicit readers, writers,
+upgrade, and downgrade rules. A schema 4 label alone must not imply provenance
+support. Unknown envelope versions and stores missing required provenance must
+be rejected by checkpoint installation and purge authorization.
+
+An old schema 4 store may backfill first IDs only from its complete, intact,
+validated prefix: replay and validate the original command/results, identify the
+first canonical insertion for each operation, then atomically persist the
+capability and envelopes. A retry's ID or a successful event's ID cannot stand
+in for missing negative provenance. If the prefix has already been removed or
+cannot be validated, refuse promotion and cleanup; do not invent IDs. Readers
+without the new capability must refuse the upgraded format rather than silently
+interpret it as the old operation encoding. Retaining all history forever is
+not the checkpoint design: validated promotion supplies a recoverable baseline
+that permits bounded suffix retention.
+
+## Capture cut and resource ownership
+
+Capture must read the header, heads, versions, events, and operations through
+one owned read transaction. The view must retain the owning store/database
+handle for the entire transaction, cursor, and output lifetime; an owned
+transaction type alone does not prove database lifetime safety. Keep the store
+`Arc` alive so dropping another handle cannot invalidate the view. These are
+ownership requirements, not a claim about unverified library behavior.
+
+The header must describe that same committed, applied cut with its full `LogId`,
+full effective membership including its decision position and node metadata,
+cluster identity, binding identity/revision, and supported storage/envelope
+capabilities. Membership and binding must come from the captured transaction,
+not from a later live read. Never capture beyond the durable committed position.
+A four-table view without this cut and metadata is not independently installable.
+
+Emit rows incrementally in deterministic section/key order. Bound row size,
+section and total bytes, counts, temporary disk consumption, and concurrent
+capture admission. Use checked arithmetic. Do not collect the whole snapshot
+in RAM. Cancellation, disk exhaustion, or limit failure must leave the live
+store unchanged and release task-owned staging resources.
+
+## Isolated validation and trust
+
+Decode only into isolated disk staging. Check framing, limits, section order,
+unique keys, complete header metadata, capability versions, cluster/binding
+compatibility, canonical command/result encoding, envelope IDs at or before the
+cut, and all head/version/event/operation relationships. A successful operation
+must agree with its version, event, and first decision. Negative operations must
+have no success event or side effect attributable to that decision.
+
+These checks still cannot prove the historical predicate behind a negative
+result. Neither a digest nor a first ID proves that a `HeadMismatch`,
+`BindingMismatch`, or `VersionConflict` was justified when first decided.
+Remote installation therefore requires an explicitly authenticated and
+authorized Raft authority for the claimed cluster, cut, and membership, or a
+complete historical proof validated from a trusted baseline. Reject input that
+has neither. Authentication and authorization must precede publication and
+purge; a caller's assertion of trust is insufficient. This document does not
+claim that this remote Raft authentication boundary is implemented.
+
+Replay validation must use disk-backed staging/indexes and bounded buffers,
+including duplicate-operation and version/head tracking. Putting the input on
+disk while retaining replay maps proportional to history does not meet this
+requirement. The existing `validate_application` RAM replay cannot simply be
+reused to claim bounded checkpoint installation.
+
+## Atomic publication and recovery
+
+After isolated validation, publication must have one durable atomic point that
+selects the complete checkpoint generation, its application state, cut,
+membership, binding, capability, and recovery metadata. A pointer switch is
+acceptable only when all referenced bytes are already durable and recovery
+cannot observe a partial generation. Serialize publication with application so
+no committed suffix write is lost between validation and the switch.
+
+Preserve local node identity and persisted vote/hard state. Remote checkpoint
+metadata cannot overwrite either. Validate cluster identity, binding, full cut
+identity, and membership before replacing state. Reject rollback behind the
+local durable application/checkpoint floor; an equal index with a different
+term/leader identity is a conflict, not an equivalent cut. Retain valid local
+suffix and commit state consistently with the installed baseline, rather than
+resetting them from remote metadata.
+
+On reopen, select only a fully published generation and verify its recoverable
+baseline before replay. The suffix starts exactly at cut.index + 1, with checked
+index arithmetic; the complete cut ID must agree wherever the boundary is
+retained. Validate contiguous suffix indices, entry identities, committed and
+applied bounds, typed application responses, canonical retry behavior, and
+membership transitions relative to checkpoint state. Reject gaps, conflicting
+boundary IDs, stale membership, or a suffix belonging to another cluster or
+binding. Recovery must begin from checkpoint heads and canonical operations,
+not from empty state or the assumption that the retained log starts at zero.
+
+Purge is authorized only after the published checkpoint can be reopened and
+validated as a recovery baseline. Delete only entries/results covered by that
+exact cut, retaining everything needed for the continuous suffix and local
+Raft durability. Re-check the selected generation and cut immediately before
+cleanup. Persist a recoverable purge boundary so interrupted cleanup is
+idempotent. Never delete the only evidence before provenance promotion or
+checkpoint durability succeeds.
+
+```mermaid
+flowchart TD
+    A[One owned transaction and store lifetime] --> B[Stream captured cut to isolated disk staging]
+    B --> C[Validate framing, semantics, provenance and authorized source]
+    C --> D[Make complete generation durable]
+    D --> E[Single atomic publication point]
+    E --> F[Reopen baseline and validate continuous suffix]
+    F --> G[Authorize purge only through that exact cut]
+    G --> H[Idempotent cleanup; preserve recoverable baseline and suffix]
+    B -->|Failure| I[Discard staging; live generation remains usable]
+    C -->|Failure| I
+    D -->|Before publication crash| I
+```
+
+A crash before publication leaves the previous generation recoverable. A crash
+after publication must recover the new complete baseline, including when purge
+has not begun or is only partly complete. Orphan staging is never selected as
+live state. Old generations may be reclaimed only after no active view needs
+them and the selected generation remains recoverable.
+
+## Behavioral verification requirements
+
+The following are required behavior checks, not reports of tests run or gates
+passed. Existing symbols above anchor the semantics; future checkpoint paths
+must demonstrate the additional requirements directly.
+
+| Behavior | Required observation |
+|---|---|
+| First negative result | Original command/result and first full ID survive capture, install, and reopen; later state does not re-evaluate it. |
+| Identical and conflicting retries | `apply_created` semantics hold across the checkpoint boundary: original result/ID retained; conflict cannot overwrite the canonical record. |
+| Old schema 4 promotion | An intact prefix validates and supplies first IDs; missing/corrupt prefix refuses promotion and purge; schema 3 to 4 behavior stays unchanged. |
+| Snapshot cut under concurrent writes | Header, membership, binding, and all rows describe one transaction cut; later writes remain in the suffix. |
+| View lifetime | Dropping other store handles during row emission cannot invalidate the retained transaction or database. |
+| Framing versus authority | Valid digest with forged negative history is insufficient; unauthenticated or unauthorized remote input is rejected. |
+| Cross-record corruption | Altered command, first ID, event, version, head, membership, or binding fails isolated validation before publication. |
+| Recovery boundary | Exact cut plus continuous suffix recovers; gaps, conflicting full IDs, cross-cluster input, and rollback are rejected. |
+| Local Raft state | Installation preserves local identity, vote/hard state, and compatible durable suffix/commit information. |
+| Resource exhaustion | Oversize rows, arithmetic overflow, disk exhaustion, cancellation, and large histories fail safely within declared memory/disk bounds. |
+| Crash publication | Crashes before and after the atomic point select a complete old or new generation, never a mixture. |
+| Interrupted purge | Reopen succeeds before cleanup is authorized; each interruption resumes without deleting the recoverable baseline or required suffix. |
+
+Documentation path and whitespace guards only validate document hygiene. They
+cannot establish these production behaviors or the remote trust boundary.