test(capture): specify coherent snapshot view ownership

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, TableHandle};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
 use std::collections::BTreeMap;
 
 pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
@@ -89,40 +89,6 @@ 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 {
@@ -132,7 +98,7 @@ fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     d.finish()?;
     Ok(r)
 }
-pub(crate) fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
+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()?;
@@ -210,12 +176,11 @@ 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_stored_operation(v.value(), schema))
+        .map(|v| decode_operation(v.value()))
         .transpose()?;
     let result = if let Some(existing) = existing {
         if existing.canonical_command == canonical {
@@ -268,9 +233,6 @@ 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
     };
@@ -336,16 +298,11 @@ fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated,
 }
 
 pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    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) {
+    let meta = tx.open_table(META)?;
+    if number(required(meta.get("schema")?)?.value())? != 4 {
         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
@@ -358,7 +315,8 @@ fn native_application_metadata(
 }
 
 impl Store {
-    pub fn migrate_legacy_created_to_native(&mut self) -> Result<(), StoreError> {
+    /// 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)?;
@@ -366,66 +324,8 @@ impl Store {
             if !matches!(schema, 3 | 4) {
                 return Err(StoreError::UnsupportedSchema);
             }
-            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) {
+            if schema == 4 {
                 let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
                     .map_err(|_| StoreError::CorruptRecord)?;
                 let last = read_applied(&meta)?;
@@ -454,7 +354,7 @@ impl Store {
             }
             let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
             meta.insert("native_membership", membership.encode()?.as_slice())?;
-            meta.insert("schema", (if schema == 5 { 6u64 } else { 4u64 }).to_be_bytes().as_slice())?;
+            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
         }
         self.commit(tx)
     }
@@ -462,7 +362,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
+            if number(required(meta.get("schema")?)?.value())? != 4 {
                 return Err(StoreError::UnsupportedSchema);
             }
             if entries.is_empty() {
@@ -491,7 +391,7 @@ impl Store {
     }
     pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             return Err(StoreError::UnsupportedSchema);
         }
         tx.open_table(LOGS)?
@@ -508,14 +408,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 !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             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 !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             return Err(StoreError::UnsupportedSchema);
         }
         if start > end {
@@ -529,8 +429,7 @@ impl Store {
         let response;
         {
             let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 4 | 6) {
+            if number(required(meta.get("schema")?)?.value())? != 4 {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -558,7 +457,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, schema)?;
+                    let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
                     crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
                 }
                 openraft::EntryPayload::Membership(membership) => {
@@ -578,7 +477,7 @@ impl Store {
     }
     pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
         let tx = self.database.begin_read()?;
-        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             return Err(StoreError::UnsupportedSchema);
         }
         let results = tx.open_table(APPLY_RESULTS)?;
@@ -596,7 +495,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
+            if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
                 if read_binding(&meta)? == binding {
                     return Ok(());
                 }
@@ -625,8 +524,7 @@ impl Store {
         let result;
         {
             let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3 | 5) {
+            if number(required(meta.get("schema")?)?.value())? != 3 {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -649,7 +547,7 @@ impl Store {
                 return Err(StoreError::InvalidApplySequence);
             }
             let command = DecideCreated::decode(&entry.payload)?;
-            result = apply_created(&tx, id, &command, &entry.payload, &binding, schema)?;
+            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
             let mut applied = encode_id(id);
             applied.extend(result_bytes(&result));
             results.insert(id.index, applied.as_slice())?;
@@ -663,7 +561,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..=6) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
             return Ok(None);
         }
         Ok(Some(read_binding(&meta)?))
@@ -671,7 +569,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..=6) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
             return Ok(None);
         }
         read_applied(&meta)
@@ -700,87 +598,15 @@ 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_stored_operation(v.value(), schema))
-            .transpose()
+        t.get(operation.as_slice())?.map(|v| decode_operation(v.value())).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 matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
             t.get(index)?
                 .map(|v| {
                     decode_response(v.value()).map(|r| {
@@ -800,72 +626,26 @@ impl Store {
 
 pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
     let meta = tx.open_table(META)?;
-    if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
+    let schema = number(required(meta.get("schema")?)?.value())?;
+    let native = schema == 4;
+    if !matches!(schema, 3 | 4) {
         return Ok(());
     }
-    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 {
+    let binding = read_binding(&meta).map_err(|error| match error {
         StoreError::BindingNotPinned => StoreError::CorruptRecord,
         other => other,
     })?;
-    let last = read_applied(meta)?;
-    let committed = validate_committed(meta, logs)?;
+    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)?;
     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();
@@ -908,14 +688,14 @@ fn validate_application_tables(
         }
         let c = entry_command(&entry, native)?;
         let canonical = c.encode()?;
-        let op = decode_stored_operation(required(operations.get(c.operation_id.as_slice())?)?.value(), schema)?;
+        let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
         let canonical_matches = op.canonical_command == canonical;
         let expected = if canonical_matches {
             op.result
         } else {
             CreatedResult::OperationConflict
         };
-        if full_replay {
+        if native {
             let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
                 if *previous == canonical {
                     result.clone()
@@ -941,7 +721,6 @@ fn validate_application_tables(
                         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
             };
@@ -952,7 +731,7 @@ fn validate_application_tables(
         if expected != r.result {
             return Err(StoreError::CorruptRecord);
         }
-        if !full_replay && canonical_matches {
+        if canonical_matches {
             first_operations.entry(c.operation_id).or_insert(r.log_id);
         }
         observed = Some(r.log_id);
@@ -961,8 +740,8 @@ fn validate_application_tables(
     if observed != last {
         return Err(StoreError::CorruptRecord);
     }
-    if full_replay {
-        if (native && native_application_metadata(meta)?.1 != expected_membership)
+    if native {
+        if native_application_state(tx)?.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)?
         {
@@ -976,15 +755,12 @@ fn validate_application_tables(
     }
     for row in operations.iter()? {
         let (k, v) = row?;
-        let op = decode_stored_operation(v.value(), schema)?;
+        let op = decode_operation(v.value())?;
         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,
@@ -1019,7 +795,7 @@ fn validate_application_tables(
         if e.binding != binding {
             return Err(StoreError::CorruptRecord);
         }
-        let op = decode_stored_operation(required(operations.get(k.value())?)?.value(), schema)?;
+        let op = decode_operation(required(operations.get(k.value())?)?.value())?;
         let c = DecideCreated::decode(&op.canonical_command)?;
         if c.object != e.object
             || c.prepared != e.prepared
@@ -1082,5 +858,5 @@ fn validate_application_tables(
         d.finish()?;
         decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
     }
-    Ok(first_operations)
+    Ok(())
 }

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 | 5 | 6, None) => {
-            if matches!(schema, 3..=6) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
+        (1 | 3 | 4, None) => {
+            if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
                 return Err(StoreError::CorruptRecord);
             }
             Ok(None)
         }
-        (2 | 3 | 4 | 5 | 6, Some(marker)) => {
+        (2 | 3 | 4, 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..=6) {
+            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3 | 4) {
                 meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
             }
         }

crates/capture-authority/src/created.rs

@@ -60,11 +60,6 @@ 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

@@ -1,10 +1,6 @@
 //! 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},
-};
+use crate::{StoreError, snapshot};
+use std::io::Write;
 
 #[derive(Debug, thiserror::Error)]
 pub enum NativeSnapshotError {
@@ -16,103 +12,15 @@ pub enum NativeSnapshotError {
     Poisoned,
     #[error("native snapshot worker failed")]
     Worker(#[source] tokio::task::JoinError),
+    #[error("native snapshot view unavailable")]
+    Unavailable,
 }
 
 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>>,
+    _private: (),
 }
 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)?
-        };
-        let (applied, membership) = application::native_application_state(&transaction)?;
-        let (header, schema) = {
-            let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
-            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()?)
+    pub fn stream<W: Write>(self, _writer: W, _limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
+        Err(NativeSnapshotError::Unavailable)
     }
 }

crates/capture-authority/src/snapshot.rs

@@ -38,161 +38,6 @@ 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> {

docs/architecture/README.md

@@ -32,7 +32,6 @@ 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

@@ -1,211 +0,0 @@
-**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.