feat(ecstore): observe bounded config under retained write owner

feat(ecstore): add bounded persisted config observations
feat(capture): persist first-decision provenance
2026-10-31 16:22:06 +00:00 · 2026-10-11 22:57:45 +08:00 · 2026-10-11 20:29:27 +08:00 · 2026-10-11 18:58:12 +08:00
8 changed files with 1572 additions and 48 deletions
@@ -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,36 @@ 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)?)?,
+    })
+}
+
+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 {
@@ -176,11 +206,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 +264,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
    };
@@ -304,7 +338,7 @@ pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<L
 fn native_application_metadata(
    meta: &impl ReadableTable<&'static str, &'static [u8]>,
 ) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    if number(required(meta.get("schema")?)?.value())? != 4 {
+    if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
        return Err(StoreError::UnsupportedSchema);
    }
    let last = read_applied(meta)?;
@@ -334,7 +368,7 @@ impl Store {
            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)?;
+            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?;
            if schema == 4 {
                return Ok(());
            }
@@ -372,7 +406,7 @@ impl Store {
            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)?;
+            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?;
        }
        self.commit(tx)
    }
@@ -383,11 +417,11 @@ impl Store {
        {
            let mut meta = tx.open_table(META)?;
            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3 | 4) {
+            if !matches!(schema, 3..=6) {
                return Err(StoreError::UnsupportedSchema);
            }
            read_binding(&meta)?;
-            if schema == 4 {
+            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)?;
@@ -416,7 +450,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)
    }
@@ -424,7 +458,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() {
@@ -453,7 +487,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)?
@@ -470,14 +504,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 {
@@ -491,7 +525,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)?;
@@ -519,7 +554,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) => {
@@ -539,7 +574,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)?;
@@ -557,7 +592,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(());
                }
@@ -586,7 +621,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)?;
@@ -609,7 +645,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())?;
@@ -623,7 +659,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)?))
@@ -631,7 +667,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)
@@ -660,15 +696,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| {
@@ -688,7 +796,7 @@ 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 | 4) {
+    if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
        return Ok(());
    }
    validate_application_tables(
@@ -699,7 +807,9 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
        &tx.open_table(EVENTS)?,
        &tx.open_table(OPERATIONS)?,
        &tx.open_table(APPLY_RESULTS)?,
+        false,
    )
+    .map(|_| ())
 }

 fn validate_application_tables(
@@ -710,11 +820,13 @@ fn validate_application_tables(
    events: &impl ReadableTable<&'static [u8], &'static [u8]>,
    operations: &impl ReadableTable<&'static [u8], &'static [u8]>,
    results: &impl ReadableTable<u64, &'static [u8]>,
-) -> Result<(), StoreError> {
+    promoting_provenance: bool,
+) -> Result<BTreeMap<[u8; 16], LogId>, StoreError> {
    let schema = number(required(meta.get("schema")?)?.value())?;
-    let native = schema == 4;
-    if !matches!(schema, 3 | 4) {
-        return Ok(());
+    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,
@@ -725,6 +837,31 @@ fn validate_application_tables(
    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();
@@ -767,14 +904,14 @@ fn validate_application_tables(
        }
        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()
@@ -800,6 +937,7 @@ 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
            };
@@ -810,7 +948,7 @@ fn validate_application_tables(
        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);
@@ -819,8 +957,8 @@ fn validate_application_tables(
    if observed != last {
        return Err(StoreError::CorruptRecord);
    }
-    if native {
-        if native_application_metadata(meta)?.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)?
        {
@@ -834,12 +972,15 @@ fn validate_application_tables(
    }
    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,
@@ -874,7 +1015,7 @@ fn validate_application_tables(
        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
@@ -937,5 +1078,5 @@ fn validate_application_tables(
        d.finish()?;
        decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
    }
-    Ok(())
+    Ok(first_operations)
 }
@@ -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())?;
            }
        }
@@ -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,
@@ -15,8 +15,8 @@ pub use native_snapshot::{NativeSnapshotError, NativeSnapshotView};
 mod raft_types;
 pub mod snapshot;
 pub use created::{
-    CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedResult, CreatedVersion, DecideCreated,
-    MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity,
+    CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedOperationProvenance, CreatedResult,
+    CreatedVersion, DecideCreated, MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity,
 };
 pub use node::{CaptureNode, CaptureNodeError};
 pub use raft_types::{
@@ -381,7 +381,7 @@ impl Store {
        let tx = store.database.begin_read()?;
        let meta = tx.open_table(META)?;
        let schema = meta.get("schema")?.ok_or(StoreError::CorruptRecord)?;
-        if !matches!(number(schema.value())?, 1..=4) {
+        if !matches!(number(schema.value())?, 1..=6) {
            return Err(StoreError::UnsupportedSchema);
        }
        let identity = meta.get("identity")?.ok_or(StoreError::CorruptRecord)?;
@@ -403,13 +403,13 @@ impl Store {
            let (key, value) = row?;
            let entry = decode_log(value.value())?;
            if next != Some(key.value()) || entry.id.index != key.value() {
-                return Err(if number(schema.value())? == 4 {
+                return Err(if matches!(number(schema.value())?, 4 | 6) {
                    StoreError::CorruptRecord
                } else {
                    StoreError::InvalidLogSequence
                });
            }
-            if number(schema.value())? == 4 {
+            if matches!(number(schema.value())?, 4 | 6) {
                application::decode_native_entry(entry.clone())?;
            }
            next = key.value().checked_add(1);
@@ -419,7 +419,7 @@ impl Store {
        application::validate_application(&tx)?;
        let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?;
        if tail != last {
-            return Err(if number(schema.value())? == 4 {
+            return Err(if matches!(number(schema.value())?, 4 | 6) {
                StoreError::CorruptRecord
            } else {
                StoreError::InvalidLogSequence
@@ -465,7 +465,10 @@ impl Store {
            return Err(StoreError::WriteDisabled);
        }
        let tx = self.database.begin_read()?;
-        if number(tx.open_table(META)?.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())? == 4 {
+        if matches!(
+            number(tx.open_table(META)?.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?,
+            4 | 6
+        ) {
            return Err(StoreError::UnsupportedSchema);
        }
        drop(tx);
@@ -1236,6 +1239,624 @@ mod tests {
        u19_assert_io(error);
    }

+    fn provenance_promote(store: &mut Store) {
+        let result = store.promote_created_provenance();
+        assert_eq!(result.as_ref().map(|_| ()).map_err(StoreError::code), Ok(()), "U26_PROVENANCE_ENABLED");
+    }
+
+    fn provenance_assert(store: &Store, command: &DecideCreated, first: LogId, result: &CreatedResult) {
+        let actual = store
+            .created_operation_provenance(command.operation_id)
+            .expect("provenance read")
+            .expect("permanent operation");
+        assert_eq!(actual.first_decision_log_id, first, "U26_FIRST_ID_STICKY");
+        assert_eq!(actual.operation.canonical_command, command.encode().expect("canonical bytes"));
+        assert_eq!(&actual.operation.result, result);
+        assert_eq!(
+            store.created_operation(command.operation_id).expect("compatible projection"),
+            Some(actual.operation)
+        );
+    }
+
+    fn provenance_commands() -> Vec<DecideCreated> {
+        let success = u17a_command();
+        let mut conflict = success.clone();
+        conflict.binding_revision = 99;
+        let mut binding = conflict.clone();
+        binding.operation_id = [11; 16];
+        binding.expected_head = Some([12; 16]);
+        let mut head = success.clone();
+        head.operation_id = [13; 16];
+        let mut version = success.clone();
+        version.operation_id = [14; 16];
+        version.expected_head = Some(success.prepared.version_id);
+        vec![
+            success.clone(),
+            success,
+            conflict,
+            binding.clone(),
+            head.clone(),
+            version.clone(),
+            binding,
+            head,
+            version,
+        ]
+    }
+
+    fn provenance_history(native: bool) -> (tempfile::TempDir, CrashBackend, Store, Vec<DecideCreated>) {
+        let (dir, backend, mut store) = u17a_bound();
+        store.save_vote(vote()).expect("nonempty provenance fixture vote");
+        let commands = provenance_commands();
+        if native {
+            store.enable_native_empty().expect("native fixture");
+            store
+                .append_native(
+                    &commands
+                        .iter()
+                        .enumerate()
+                        .map(|(i, c)| u17a_created(u64::try_from(i).expect("index"), c.clone()))
+                        .collect::<Vec<_>>(),
+                )
+                .expect("native prefix");
+        } else {
+            store
+                .append(
+                    &commands
+                        .iter()
+                        .enumerate()
+                        .map(|(i, c)| LogEntry {
+                            id: u17a_id(u64::try_from(i).expect("index")),
+                            payload: c.encode().expect("legacy canonical"),
+                        })
+                        .collect::<Vec<_>>(),
+                )
+                .expect("legacy prefix");
+        }
+        store.save_committed(u17a_id(8)).expect("complete commit");
+        for i in 0..9 {
+            if native {
+                store.apply_next_native(u17a_id(i)).expect("native apply");
+            } else {
+                store.apply_next_created(u17a_id(i)).expect("legacy apply");
+            }
+        }
+        (dir, backend, store, commands)
+    }
+
+    #[test]
+    fn provenance_first_decisions_and_retries_reopen() {
+        let (_dir, backend, mut store) = u17a_store();
+        provenance_promote(&mut store);
+        assert_eq!(committed_schema(&store), 6);
+        let commands = provenance_commands();
+        let entries = commands
+            .iter()
+            .enumerate()
+            .map(|(i, c)| u17a_created(u64::try_from(i).expect("index"), c.clone()))
+            .collect::<Vec<_>>();
+        store.append_native(&entries).expect("Created entries");
+        store.save_committed(u17a_id(8)).expect("committed Created");
+        for i in 0..9 {
+            store.apply_next_native(u17a_id(i)).expect("decision and retry");
+        }
+        let expected = [
+            (
+                0usize,
+                0u64,
+                CreatedResult::Created {
+                    version_id: [3; 16],
+                    event_id: [1; 16],
+                    decision_log_id: u17a_id(0),
+                },
+            ),
+            (3, 3, CreatedResult::BindingMismatch),
+            (
+                4,
+                4,
+                CreatedResult::HeadMismatch {
+                    actual_head: Some([3; 16]),
+                },
+            ),
+            (5, 5, CreatedResult::VersionConflict),
+        ];
+        for (index, first, result) in &expected {
+            provenance_assert(&store, &commands[*index], u17a_id(*first), result);
+        }
+        assert_eq!(
+            store
+                .native_apply_response(2)
+                .expect("conflict receipt")
+                .expect("receipt")
+                .created_result(),
+            Some(&CreatedResult::OperationConflict)
+        );
+        drop(store);
+        let store = Store::open(&backend.path, identity()).expect("reopen full provenance replay");
+        for (index, first, result) in &expected {
+            provenance_assert(&store, &commands[*index], u17a_id(*first), result);
+        }
+        assert_eq!(store.created_event(commands[3].operation_id).expect("negative event"), None);
+        assert_eq!(encode_id(u17a_id(3)).len(), 24);
+    }
+
+    #[test]
+    fn provenance_native_controls_preserve_state() {
+        let (_dir, backend, mut store) = u17a_store();
+        provenance_promote(&mut store);
+        let (member, membership) = u17a_membership(1);
+        let entries = vec![u17a_blank(0), member, u17a_created(2, u17a_command()), u17a_blank(3)];
+        let writer = NativeLogWriter::new(store);
+        let runtime = u19_runtime();
+        runtime.block_on(writer.append_native(entries.clone())).expect("async append");
+        let responses = runtime
+            .block_on(writer.apply_committed(entries))
+            .expect("async committed apply");
+        assert_eq!(responses[0], CaptureResponse::blank(u17a_id(0)).expect("blank"));
+        assert_eq!(responses[1], CaptureResponse::membership(u17a_id(1)).expect("membership"));
+        assert_eq!(
+            runtime.block_on(writer.applied_state()).expect("applied pair"),
+            (Some(u17a_id(3)), membership.clone())
+        );
+        {
+            let store = writer.store.lock().expect("store");
+            assert_eq!(
+                store
+                    .database
+                    .begin_read()
+                    .expect("read")
+                    .open_table(application::OPERATIONS)
+                    .expect("operations")
+                    .len()
+                    .expect("count"),
+                1
+            );
+            assert_eq!(store.created_apply_result(0).expect("blank projection"), None);
+            assert_eq!(store.created_apply_result(1).expect("membership projection"), None);
+        }
+        let view = runtime.block_on(writer.snapshot_view()).expect("schema6 native view");
+        drop(view);
+        backend.cut();
+        drop(writer);
+        let store = Store::open(&backend.path, identity()).expect("async ACK cut reopen");
+        assert_eq!(store.native_application_state().expect("recovered pair"), (Some(u17a_id(3)), membership));
+        for (i, response) in responses.iter().enumerate() {
+            assert_eq!(
+                store
+                    .native_apply_response(u64::try_from(i).expect("index"))
+                    .expect("receipt")
+                    .as_ref(),
+                Some(response),
+                "U26_DURABLE_ACK"
+            );
+        }
+        assert_eq!(
+            store
+                .created_operation_provenance([1; 16])
+                .expect("durable operation")
+                .expect("operation")
+                .first_decision_log_id,
+            u17a_id(2),
+            "U26_DURABLE_ACK"
+        );
+    }
+
+    #[test]
+    fn provenance_promotes_schema3_and_schema4_exact_history() {
+        for native in [false, true] {
+            let (_dir, backend, mut store, commands) = provenance_history(native);
+            let old = u17b_snapshot(&store);
+            assert_eq!(
+                store
+                    .created_operation_provenance([1; 16])
+                    .expect_err("old schema has no provenance")
+                    .code(),
+                "unsupported_schema"
+            );
+            provenance_promote(&mut store);
+            assert_eq!(committed_schema(&store), if native { 6 } else { 5 });
+            let new = u17b_snapshot(&store);
+            let metadata_without_schema = |rows: &Vec<(Vec<u8>, Vec<u8>)>| {
+                rows.iter()
+                    .filter(|(key, _)| key.as_slice() != b"schema")
+                    .cloned()
+                    .collect::<Vec<_>>()
+            };
+            assert_eq!(
+                metadata_without_schema(&old[0]),
+                metadata_without_schema(&new[0]),
+                "only schema metadata changes"
+            );
+            for i in 1..6 {
+                assert_eq!(old[i], new[i], "promotion preserves logs, responses and business state");
+            }
+            assert_eq!(old[7], new[7], "promotion preserves vote");
+            for (old_row, new_row) in old[6].iter().zip(&new[6]) {
+                assert_eq!(old_row.0, new_row.0);
+                assert_eq!(&new_row.1[..5], b"COPR\x01");
+                assert_eq!(&new_row.1[29..], old_row.1.as_slice(), "canonical old value must remain byte exact");
+            }
+            for (i, first) in [(0usize, 0u64), (3, 3), (4, 4), (5, 5)] {
+                let row = store
+                    .created_operation_provenance(commands[i].operation_id)
+                    .expect("promoted read")
+                    .expect("operation");
+                assert_eq!(row.first_decision_log_id, u17a_id(first), "U26_PROMOTION_FIRST_ID");
+            }
+            let once = u17b_snapshot(&store);
+            provenance_promote(&mut store);
+            assert_eq!(u17b_snapshot(&store), once);
+            drop(store);
+            let mut store = Store::open(&backend.path, identity()).expect("reopen promoted history");
+            let mut suffix = commands[0].clone();
+            suffix.operation_id = [20; 16];
+            suffix.prepared.version_id = [21; 16];
+            suffix.expected_head = Some([3; 16]);
+            if native {
+                store
+                    .append_native(&[u17a_created(9, suffix.clone())])
+                    .expect("native suffix");
+            } else {
+                store
+                    .append(&[LogEntry {
+                        id: u17a_id(9),
+                        payload: suffix.encode().expect("suffix"),
+                    }])
+                    .expect("legacy suffix");
+            }
+            store.save_committed(u17a_id(9)).expect("suffix committed");
+            if native {
+                store.apply_next_native(u17a_id(9)).expect("native suffix apply");
+            } else {
+                store.apply_next_created(u17a_id(9)).expect("legacy suffix apply");
+            }
+            assert_eq!(
+                store
+                    .created_operation_provenance(suffix.operation_id)
+                    .expect("suffix provenance")
+                    .expect("suffix operation")
+                    .first_decision_log_id,
+                u17a_id(9)
+            );
+        }
+    }
+
+    #[test]
+    fn provenance_rejects_incomplete_or_forged_prefix_atomically() {
+        use application::{APPLY_RESULTS, EVENTS, HEADS, OPERATIONS, VERSIONS};
+        for native in [false, true] {
+            for corruption in 0..9 {
+                let (_dir, _backend, mut store, commands) = provenance_history(native);
+                let tx = store.database.begin_write().expect("corruption transaction");
+                match corruption {
+                    0 => {
+                        tx.open_table(LOGS).expect("logs").remove(0).expect("missing prefix");
+                    }
+                    1 => {
+                        tx.open_table(APPLY_RESULTS)
+                            .expect("responses")
+                            .remove(3)
+                            .expect("missing negative response");
+                    }
+                    2 => {
+                        let mut logs = tx.open_table(LOGS).expect("logs");
+                        let raw = logs.get(3).expect("get").expect("row").value().to_vec();
+                        let mut entry = decode_log(&raw).expect("log");
+                        entry.id.term += 1;
+                        logs.insert(3, encode_log(&entry).expect("encode").as_slice())
+                            .expect("wrong full ID");
+                    }
+                    3 => {
+                        let mut operations = tx.open_table(OPERATIONS).expect("operations");
+                        let raw = operations
+                            .get(commands[3].operation_id.as_slice())
+                            .expect("get")
+                            .expect("row")
+                            .value()
+                            .to_vec();
+                        operations.insert([99; 16].as_slice(), raw.as_slice()).expect("orphan");
+                    }
+                    4 => {
+                        tx.open_table(EVENTS)
+                            .expect("events")
+                            .remove([1; 16].as_slice())
+                            .expect("missing event");
+                    }
+                    5 => {
+                        let mut meta = tx.open_table(META).expect("meta");
+                        meta.insert("last_applied", encode_id(u17a_id(7)).as_slice())
+                            .expect("wrong applied ID");
+                    }
+                    6 => {
+                        tx.open_table(HEADS)
+                            .expect("heads")
+                            .remove(crate::created::object_bytes(&commands[0].object).expect("object").as_slice())
+                            .expect("missing head");
+                    }
+                    7 => {
+                        let mut versions = tx.open_table(VERSIONS).expect("versions");
+                        let key = versions
+                            .iter()
+                            .expect("iter")
+                            .next()
+                            .expect("version")
+                            .expect("row")
+                            .0
+                            .value()
+                            .to_vec();
+                        versions.remove(key.as_slice()).expect("missing version");
+                    }
+                    _ => {
+                        let mut logs = tx.open_table(LOGS).expect("logs");
+                        let raw = logs.get(2).expect("get").expect("row").value().to_vec();
+                        let mut entry = decode_log(&raw).expect("log");
+                        entry.payload = if native {
+                            decode_log(logs.get(3).expect("source log").expect("source row").value())
+                                .expect("source entry")
+                                .payload
+                        } else {
+                            commands[3].encode().expect("canonical")
+                        };
+                        logs.insert(2, encode_log(&entry).expect("encode").as_slice())
+                            .expect("forged early conflict");
+                    }
+                }
+                tx.commit().expect("inject corruption");
+                let before = u17b_snapshot(&store);
+                assert_eq!(
+                    store
+                        .promote_created_provenance()
+                        .expect_err("corrupt promotion refused")
+                        .code(),
+                    "corrupt_record"
+                );
+                assert_eq!(u17b_snapshot(&store), before, "rejection must have zero writes");
+            }
+        }
+    }
+
+    #[test]
+    fn provenance_empty_first_commit_never_downgrades() {
+        for native in [false, true] {
+            let (_dir, backend, mut store) = u17a_bound();
+            provenance_promote(&mut store);
+            assert_eq!(committed_schema(&store), 5);
+            if native {
+                store.enable_native_empty().expect("empty5 explicit native6");
+                store.append_native(&[u17a_blank(0)]).expect("blank");
+            } else {
+                store
+                    .append(&[LogEntry {
+                        id: u17a_id(0),
+                        payload: u17a_command().encode().expect("command"),
+                    }])
+                    .expect("legacy");
+            }
+            store.save_committed(u17a_id(0)).expect("first committed");
+            assert_eq!(committed_schema(&store), if native { 6 } else { 5 }, "U26_SCHEMA_NEVER_DOWNGRADES");
+            if native {
+                store.apply_next_native(u17a_id(0)).expect("blank apply");
+            } else {
+                store.apply_next_created(u17a_id(0)).expect("legacy apply");
+                assert_eq!(
+                    store.enable_native_empty().expect_err("populated5 forbidden").code(),
+                    "unsupported_schema"
+                );
+                assert_eq!(
+                    store
+                        .migrate_legacy_created_to_native()
+                        .expect_err("U17B cannot reinterpret5")
+                        .code(),
+                    "unsupported_schema"
+                );
+            }
+            drop(store);
+            let store = Store::open(&backend.path, identity()).expect("reopen schema");
+            assert_eq!(committed_schema(&store), if native { 6 } else { 5 }, "U26_SCHEMA_NEVER_DOWNGRADES");
+            if !native {
+                let writer = NativeLogWriter::new(store);
+                assert!(
+                    u19_runtime().block_on(writer.snapshot_view()).is_err(),
+                    "schema5 cannot expose a native view"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn provenance_old_binary_and_unknown_versions_reject() {
+        for native in [false, true] {
+            for corruption in 0..10 {
+                let (_dir, backend, mut store, _) = provenance_history(native);
+                provenance_promote(&mut store);
+                let tx = store.database.begin_write().expect("corruption");
+                if corruption == 0 {
+                    tx.open_table(META)
+                        .expect("meta")
+                        .insert("schema", 7u64.to_be_bytes().as_slice())
+                        .expect("future schema");
+                } else {
+                    let mut operations = tx.open_table(application::OPERATIONS).expect("operations");
+                    let key = [11; 16];
+                    let mut raw = operations
+                        .get(key.as_slice())
+                        .expect("get")
+                        .expect("negative")
+                        .value()
+                        .to_vec();
+                    match corruption {
+                        1 => raw[4] = 2,
+                        2 => {
+                            raw.pop();
+                        }
+                        3 => raw.push(0),
+                        4 => raw[5..29].fill(0),
+                        5 => raw[5..29].copy_from_slice(&encode_id(u17a_id(6))),
+                        6 => {
+                            let mut id = u17a_id(3);
+                            id.leader_node += 1;
+                            raw[5..29].copy_from_slice(&encode_id(id));
+                        }
+                        7 => {
+                            let mut id = u17a_id(3);
+                            id.term += 1;
+                            raw[5..29].copy_from_slice(&encode_id(id));
+                        }
+                        8 => raw[29..37].copy_from_slice(&u64::MAX.to_be_bytes()),
+                        _ => raw = raw[29..].to_vec(),
+                    }
+                    operations.insert(key.as_slice(), raw.as_slice()).expect("bad envelope");
+                }
+                tx.commit().expect("persist corruption");
+                drop(store);
+                assert_eq!(
+                    Store::open(&backend.path, identity())
+                        .err()
+                        .expect("strict open rejects")
+                        .code(),
+                    if corruption == 0 {
+                        "unsupported_schema"
+                    } else {
+                        "corrupt_record"
+                    }
+                );
+            }
+        }
+        // The independent accepted old-binary probe is run by the root task.
+    }
+
+    #[test]
+    fn provenance_promotion_and_apply_powercut_atomic() {
+        for native in [false, true] {
+            for promotion in [false, true] {
+                let fixture = || {
+                    let (dir, backend, mut store, _) = provenance_history(native);
+                    if !promotion {
+                        provenance_promote(&mut store);
+                        let mut c = u17a_command();
+                        c.operation_id = [30; 16];
+                        c.binding_revision = 99;
+                        if native {
+                            store.append_native(&[u17a_created(9, c)]).expect("pending native");
+                        } else {
+                            store
+                                .append(&[LogEntry {
+                                    id: u17a_id(9),
+                                    payload: c.encode().expect("pending canonical"),
+                                }])
+                                .expect("pending legacy");
+                        }
+                        store.save_committed(u17a_id(9)).expect("pending commit");
+                    }
+                    (dir, backend, store)
+                };
+                let action = |store: &mut Store| -> Result<(), StoreError> {
+                    if promotion {
+                        store.promote_created_provenance()
+                    } else if native {
+                        store.apply_next_native(u17a_id(9)).map(|_| ())
+                    } else {
+                        store.apply_next_created(u17a_id(9)).map(|_| ())
+                    }
+                };
+                let (_dir, backend, mut store) = fixture();
+                let old = u17b_snapshot(&store);
+                backend.media.lock().expect("reset syncs").syncs = 0;
+                let result = action(&mut store);
+                assert_eq!(result.as_ref().map(|_| ()).map_err(StoreError::code), Ok(()), "U26_PROVENANCE_ENABLED");
+                let reached = backend.media.lock().expect("syncs").syncs;
+                assert_eq!(reached > 0, true, "U26_DURABLE_ACK");
+                let new = u17b_snapshot(&store);
+                backend.cut();
+                drop(store);
+                let database = Database::open(&backend.path).expect("ACK database recovery");
+                assert_eq!(u17b_snapshot_database(&database), new, "U26_DURABLE_ACK");
+                drop(database);
+                let (_dir, backend, mut store) = fixture();
+                backend.arm(1, false, false);
+                assert_eq!(
+                    action(&mut store).expect_err("uncertain provenance commit").code(),
+                    "commit_indeterminate"
+                );
+                let fenced = u17b_snapshot(&store);
+                let mutations: [fn(&mut Store) -> Result<(), StoreError>; 11] = [
+                    Store::promote_created_provenance,
+                    |store| store.append(&[]),
+                    |store| store.append_native(&[]),
+                    |store| store.save_vote(vote()),
+                    |store| store.save_committed(u17a_id(8)),
+                    |store| store.apply_next_created(u17a_id(0)).map(|_| ()),
+                    |store| store.apply_next_native(u17a_id(0)).map(|_| ()),
+                    |store| store.pin_capture_binding(u17a_binding()),
+                    Store::enable_native_empty,
+                    Store::migrate_legacy_created_to_native,
+                    |store| store.truncate_suffix(9),
+                ];
+                for mutation in mutations {
+                    assert_eq!(
+                        mutation(&mut store)
+                            .expect_err("uncertain handle must fence every mutation")
+                            .code(),
+                        "write_disabled",
+                        "U26_WRITE_FENCE"
+                    );
+                    assert_eq!(u17b_snapshot(&store), fenced, "fenced mutation must not change any table");
+                }
+                let writer = NativeLogWriter::new(store);
+                let result = u19_runtime().block_on(writer.apply_committed(vec![]));
+                assert!(
+                    matches!(result, Err(NativeWriteError::Store(StoreError::WriteDisabled))),
+                    "U26_WRITE_FENCE async apply"
+                );
+                assert_eq!(u17b_snapshot(&writer.store.lock().expect("fenced writer")), fenced);
+                drop(writer);
+                for sync in 1..=reached {
+                    for after in [false, true] {
+                        let (_dir, backend, mut store) = fixture();
+                        assert_eq!(u17b_snapshot(&store), old);
+                        backend.arm(sync, after, true);
+                        let result = action(&mut store);
+                        let actual_syncs = backend.media.lock().expect("reached cut").syncs;
+                        assert!(actual_syncs >= sync, "cut must actually reach target sync");
+                        backend.cut();
+                        drop(store);
+                        match Database::open(&backend.path) {
+                            Ok(database) => {
+                                let actual = u17b_snapshot_database(&database);
+                                assert!(actual == old || actual == new, "physical all-table atomicity");
+                                if result.is_ok() {
+                                    assert_eq!(actual, new, "U26_DURABLE_ACK");
+                                }
+                                let recovered = Store::from_database(database, identity())
+                                    .expect("logical corruption cannot hide mixed state");
+                                assert_eq!(u17b_snapshot(&recovered), actual);
+                            }
+                            Err(redb::DatabaseError::Storage(redb::StorageError::Corrupted(_))) => {
+                                assert!(result.is_err(), "U26_DURABLE_ACK");
+                            }
+                            Err(error) => panic!("unexpected physical recovery error: {error:?}"),
+                        }
+                    }
+                }
+            }
+            let (_dir, _backend, mut expected_store, _) = provenance_history(native);
+            provenance_promote(&mut expected_store);
+            let expected = u17b_snapshot(&expected_store);
+            let dir = tempfile::tempdir().expect("child provenance directory");
+            let path = dir.path().join("db");
+            let mut store = Store::initialize(&path, identity()).expect("child database");
+            store.pin_capture_binding(u17a_binding()).expect("child binding");
+            drop(store);
+            // Process termination after ACK is separate from the physical media cuts above.
+            let mut process = child(&path, if native { "provenance-native" } else { "provenance-legacy" });
+            ack(&mut process);
+            process.kill().expect("SIGKILL acknowledged child");
+            assert!(!process.wait().expect("reap killed child").success());
+            let reopened = Store::open(&path, identity()).expect("independent child ACK reopen");
+            assert_eq!(committed_schema(&reopened), if native { 6 } else { 5 });
+            assert_eq!(u17b_snapshot(&reopened), expected, "U26_DURABLE_ACK complete child state");
+        }
+    }
+
    fn u17b_snapshot(store: &Store) -> Vec<Vec<(Vec<u8>, Vec<u8>)>> {
        u17b_snapshot_database(&store.database)
    }
@@ -1278,6 +1899,17 @@ mod tests {
                    .collect(),
            );
        }
+        snapshot.push(
+            tx.open_table(VOTE)
+                .expect("snapshot vote table")
+                .iter()
+                .expect("vote iterator")
+                .map(|row| {
+                    let (key, value) = row.expect("vote row");
+                    (key.value().to_be_bytes().to_vec(), value.value().to_vec())
+                })
+                .collect(),
+        );
        snapshot
    }
    fn u17b_history() -> (tempfile::TempDir, CrashBackend, Store, Vec<DecideCreated>) {
@@ -3440,7 +4072,48 @@ mod tests {
                assert!(Store::open(path, identity()).is_err(), "U08_INVARIANT exclusive lock absent");
            } else {
                let mut store = Store::open(path, identity()).expect("child open");
-                if std::env::var("U08_CHILD_MODE").expect("mode") == "committed" {
+                let mode = std::env::var("U08_CHILD_MODE").expect("mode");
+                if matches!(mode.as_str(), "provenance-native" | "provenance-legacy") {
+                    store.save_vote(vote()).expect("child provenance vote ACK");
+                    let native = mode == "provenance-native";
+                    if native {
+                        store.enable_native_empty().expect("child native enable");
+                    }
+                    provenance_promote(&mut store);
+                    let commands = provenance_commands();
+                    if native {
+                        store
+                            .append_native(
+                                &commands
+                                    .iter()
+                                    .enumerate()
+                                    .map(|(i, command)| u17a_created(u64::try_from(i).expect("child index"), command.clone()))
+                                    .collect::<Vec<_>>(),
+                            )
+                            .expect("child native append");
+                    } else {
+                        store
+                            .append(
+                                &commands
+                                    .iter()
+                                    .enumerate()
+                                    .map(|(i, command)| LogEntry {
+                                        id: u17a_id(u64::try_from(i).expect("child index")),
+                                        payload: command.encode().expect("child canonical"),
+                                    })
+                                    .collect::<Vec<_>>(),
+                            )
+                            .expect("child legacy append");
+                    }
+                    store.save_committed(u17a_id(8)).expect("child committed ACK");
+                    for index in 0..9 {
+                        if native {
+                            store.apply_next_native(u17a_id(index)).expect("child native apply ACK");
+                        } else {
+                            store.apply_next_created(u17a_id(index)).expect("child legacy apply ACK");
+                        }
+                    }
+                } else if mode == "committed" {
                    store.save_committed(truncate_batch()[2].id).expect("child committed ACK");
                } else if std::env::var("U08_CHILD_MODE").expect("mode") == "truncate" {
                    store.truncate_suffix(2).expect("truncate ACK");
@@ -372,6 +372,25 @@ 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
+        }
+
+        /// Observe bounded config bytes while the caller retains the object's write owner.
+        pub async fn read_config_observation_no_lock(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            max_bytes: usize,
+            guard: &crate::object_api::WriteCommitGuard,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_observation_no_lock(api, file, max_bytes, guard).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>,
@@ -449,6 +449,64 @@ 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_observation_no_lock<S>(
+    api: Arc<S>,
+    file: &str,
+    max_bytes: usize,
+    guard: &crate::object_api::WriteCommitGuard,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    if !guard.matches(RUSTFS_META_BUCKET, file) {
+        return Err(Error::InvalidArgument(
+            RUSTFS_META_BUCKET.to_owned(),
+            file.to_owned(),
+            "locked observation requires a matching write owner".to_owned(),
+        ));
+    }
+    let lock_lost_error = || Error::NamespaceLockQuorumUnavailable {
+        mode: "locked_config_observation",
+        bucket: RUSTFS_META_BUCKET.to_owned(),
+        object: file.to_owned(),
+        required: 1,
+        achieved: 0,
+    };
+    if guard.is_lock_lost() {
+        return Err(lock_lost_error());
+    }
+    let opts = ObjectOptions {
+        suppress_read_repair: true,
+        no_lock: true,
+        skip_verify_bitrot: false,
+        ..Default::default()
+    };
+    let result = crate::object_api::without_get_object_body_cache_hook(read_config_limited_preserve_empty_with_metadata_opts(
+        api, file, &opts, max_bytes,
+    ))
+    .await;
+    if guard.is_lock_lost() {
+        return Err(lock_lost_error());
+    }
+    result
+}
+
 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 +5483,329 @@ mod tests {
        assert_eq!(store.write_calls.load(Ordering::SeqCst), 0);
    }

+    struct ObservationStream {
+        inner: Cursor<Vec<u8>>,
+        consumed: Arc<AtomicUsize>,
+        fail: bool,
+        eof_gate: Option<(Option<tokio::sync::oneshot::Sender<()>>, tokio::sync::oneshot::Receiver<()>)>,
+    }
+
+    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")));
+            }
+            if self.inner.position() == u64::try_from(self.inner.get_ref().len()).expect("fixture length")
+                && let Some((ready, release)) = self.eof_gate.as_mut()
+            {
+                if let Some(ready) = ready.take() {
+                    ready.send(()).expect("EOF observer remains active");
+                }
+                match std::future::Future::poll(Pin::new(release), cx) {
+                    Poll::Pending => return Poll::Pending,
+                    Poll::Ready(result) => result.expect("EOF gate released"),
+                }
+                self.eof_gate = None;
+            }
+            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,
+        gets: AtomicUsize,
+        eof_gate: std::sync::Mutex<Option<(Option<tokio::sync::oneshot::Sender<()>>, tokio::sync::oneshot::Receiver<()>)>>,
+        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,
+                gets: AtomicUsize::new(0),
+                eof_gate: std::sync::Mutex::new(None),
+                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");
+            self.gets.fetch_add(1, Ordering::SeqCst);
+            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,
+                    eof_gate: self.eof_gate.lock().expect("EOF gate").take(),
+                }),
+                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 locked_observation_preserves_options_and_typed_errors() {
+        let object = "config/observation.bin";
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let wrapper = rustfs_lock::NamespaceLockWrapper::new(
+            rustfs_lock::NamespaceLock::with_local_manager("locked-observation-options".to_owned(), manager.clone()),
+            rustfs_lock::ObjectKey::new(RUSTFS_META_BUCKET, object),
+            "locked-observation-options".to_owned(),
+        );
+        let guard = crate::object_api::WriteCommitGuard::acquire(&wrapper, std::time::Duration::from_secs(5))
+            .await
+            .expect("matching real owner");
+        let store = Arc::new(ObservationStore::new(b"raw".to_vec(), 3));
+        let result = super::read_config_observation_no_lock(store.clone(), object, 3, &guard).await;
+        assert_eq!(result.is_ok(), true, "I06L_READ_FLAGS: {result:?}");
+        assert_eq!(
+            *store.options.lock().expect("options") == Some((true, true, false)),
+            true,
+            "I06L_READ_FLAGS"
+        );
+        assert_eq!(result.expect("raw observation").0, b"raw");
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        for (bucket, key) in [(RUSTFS_META_BUCKET, "config/other.bin"), ("wrong-bucket", object)] {
+            let wrapper = rustfs_lock::NamespaceLockWrapper::new(
+                rustfs_lock::NamespaceLock::with_local_manager("locked-observation-target".to_owned(), manager.clone()),
+                rustfs_lock::ObjectKey::new(bucket, key),
+                "locked-observation-target".to_owned(),
+            );
+            let wrong = crate::object_api::WriteCommitGuard::acquire(&wrapper, std::time::Duration::from_secs(5))
+                .await
+                .expect("wrong-target real owner");
+            let store = Arc::new(ObservationStore::new(Vec::new(), 0));
+            let result = super::read_config_observation_no_lock(store.clone(), object, 8, &wrong).await;
+            assert_eq!(matches!(result, Err(Error::InvalidArgument(..))), true, "I06L_TARGET_REJECTED");
+            assert_eq!(store.gets.load(Ordering::SeqCst) == 0, true, "I06L_TARGET_REJECTED");
+        }
+        for error in [
+            Error::FileNotFound,
+            Error::ObjectNotFound(RUSTFS_META_BUCKET.to_owned(), object.to_owned()),
+            Error::ErasureReadQuorum,
+        ] {
+            let expected = if error == Error::FileNotFound || matches!(error, Error::ObjectNotFound(..)) {
+                Error::ConfigNotFound
+            } else {
+                error.clone()
+            };
+            let mut fixture = ObservationStore::new(Vec::new(), 0);
+            fixture.error = Some(error);
+            let store = Arc::new(fixture);
+            assert_eq!(
+                super::read_config_observation_no_lock(store.clone(), object, 8, &guard)
+                    .await
+                    .expect_err("typed error"),
+                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 err = super::read_config_observation_no_lock(store.clone(), object, 8, &guard)
+            .await
+            .expect_err("stream failure");
+        assert_eq!(err.to_string().contains("observation stream failed"), true, "I06L_TYPED_STREAM_ERROR");
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        let store = Arc::new(ObservationStore::new(vec![7; 64], 0));
+        let result = super::read_config_observation_no_lock(store.clone(), object, 8, &guard).await;
+        assert_eq!(matches!(result, Err(Error::CorruptedFormat)), true, "I06L_STREAM_BOUND");
+        assert_eq!(store.consumed.load(Ordering::SeqCst) == 9, true, "I06L_STREAM_BOUND");
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test]
+    async fn locked_observation_rejects_actual_owner_loss_before_and_after_read() {
+        let object = "config/observation.bin";
+        for after_read in [false, true] {
+            let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+            let client = Arc::new(RefreshControlledClient::new(manager));
+            let wrapper = rustfs_lock::NamespaceLockWrapper::new(
+                rustfs_lock::NamespaceLock::new("locked-observation-loss".to_owned(), client.clone()),
+                rustfs_lock::ObjectKey::new(RUSTFS_META_BUCKET, object),
+                "locked-observation-loss".to_owned(),
+            );
+            let guard = crate::object_api::WriteCommitGuard::acquire(&wrapper, std::time::Duration::from_secs(5))
+                .await
+                .expect("real distributed owner");
+            let signal = guard.lock_lost_signal().expect("acquired distributed owner signal");
+            let store = Arc::new(ObservationStore::new(b"raw".to_vec(), 3));
+            if after_read {
+                let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
+                let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+                *store.eof_gate.lock().expect("EOF gate") = Some((Some(ready_tx), release_rx));
+                let mut read = Box::pin(super::read_config_observation_no_lock(store.clone(), object, 3, &guard));
+                tokio::time::timeout(std::time::Duration::from_secs(60), async {
+                    tokio::select! {
+                        ready = ready_rx => ready.expect("read reached EOF"),
+                        result = &mut read => assert_eq!(false, true, "I06L_LOSS_AFTER_READ: returned before EOF gate: {result:?}"),
+                    }
+                }).await.expect("EOF watchdog");
+                client.reject_refreshes();
+                tokio::time::timeout(std::time::Duration::from_secs(60), signal.notified())
+                    .await
+                    .expect("actual owner loss watchdog");
+                assert_eq!(guard.is_lock_lost(), true, "I06L_LOSS_AFTER_READ");
+                release_tx.send(()).expect("release EOF");
+                let result = tokio::time::timeout(std::time::Duration::from_secs(60), read)
+                    .await
+                    .expect("read watchdog");
+                assert_eq!(
+                    matches!(
+                        result,
+                        Err(Error::NamespaceLockQuorumUnavailable {
+                            mode: "locked_config_observation",
+                            ..
+                        })
+                    ),
+                    true,
+                    "I06L_LOSS_AFTER_READ"
+                );
+                assert_eq!(store.gets.load(Ordering::SeqCst), 1);
+            } else {
+                client.reject_refreshes();
+                tokio::time::timeout(std::time::Duration::from_secs(60), signal.notified())
+                    .await
+                    .expect("actual owner loss watchdog");
+                assert_eq!(guard.is_lock_lost(), true, "I06L_LOSS_BEFORE_GET");
+                let result = super::read_config_observation_no_lock(store.clone(), object, 3, &guard).await;
+                assert_eq!(
+                    matches!(
+                        result,
+                        Err(Error::NamespaceLockQuorumUnavailable {
+                            mode: "locked_config_observation",
+                            ..
+                        })
+                    ),
+                    true,
+                    "I06L_LOSS_BEFORE_GET"
+                );
+                assert_eq!(store.gets.load(Ordering::SeqCst) == 0, true, "I06L_LOSS_BEFORE_GET");
+            }
+            assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        }
+    }
+
+    #[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));
@@ -68,7 +68,7 @@ impl WriteCommitGuard {
        self.guards.iter().any(|guard| guard.is_lock_lost() || guard.is_released())
    }

-    fn matches(&self, bucket: &str, object: &str) -> bool {
+    pub(crate) fn matches(&self, bucket: &str, object: &str) -> bool {
        self.bucket.as_ref() == bucket && self.object.as_ref() == object
    }
 }
@@ -1095,6 +1095,311 @@ 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 locked_observation_uses_real_outer_guard() {
+        let object = "config/locked-observation-real.bin";
+        let payload = vec![b'p'; OBSERVATION_HOOK_BODY_LEN];
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        let written = set
+            .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist locked observation fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; payload.len()]);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; payload.len()]);
+        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
+        let namespace = set
+            .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, object)
+            .await
+            .expect("namespace");
+        let guard = crate::object_api::WriteCommitGuard::acquire(&namespace, std::time::Duration::from_secs(5))
+            .await
+            .expect("matching real owner");
+        let mut contender = Box::pin(crate::object_api::WriteCommitGuard::acquire(
+            &namespace,
+            std::time::Duration::from_secs(30),
+        ));
+        assert_eq!(futures::poll!(contender.as_mut()).is_pending(), true, "I06L_OWNER_RETAINED");
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            crate::config::com::read_config_observation_no_lock(set.clone(), object, payload.len(), &guard),
+        )
+        .await
+        .expect("locked observation hang watchdog");
+        assert_eq!(result.is_ok(), true, "I06L_NO_NESTED_LOCK: {result:?}");
+        let (bytes, info) = result.expect("locked observation");
+        assert_eq!(bytes == payload, true, "I06L_HOOK_BYPASS");
+        assert_eq!(calls.load(Ordering::SeqCst) == 0, true, "I06L_HOOK_BYPASS");
+        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);
+        assert_eq!(futures::poll!(contender.as_mut()).is_pending(), true, "I06L_OWNER_RETAINED");
+        drop(guard);
+        let next = tokio::time::timeout(std::time::Duration::from_secs(10), contender)
+            .await
+            .expect("owner release watchdog")
+            .expect("competing owner acquired");
+        drop(next);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; payload.len()]);
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn locked_observation_preserves_empty_missing_and_bound() {
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        for (object, body) in [
+            ("config/locked-empty.bin", Vec::new()),
+            ("config/locked-bound.bin", b"raw".to_vec()),
+        ] {
+            let mut reader = PutObjReader::from_vec(body.clone());
+            let written = set
+                .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+                .await
+                .expect("persist boundary fixture");
+            let namespace = set
+                .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, object)
+                .await
+                .expect("namespace");
+            let guard = crate::object_api::WriteCommitGuard::acquire(&namespace, std::time::Duration::from_secs(5))
+                .await
+                .expect("matching boundary owner");
+            let result = crate::config::com::read_config_observation_no_lock(set.clone(), object, body.len(), &guard).await;
+            assert_eq!(result.is_ok(), true, "I06L_EMPTY_DISTINCT: {result:?}");
+            let (bytes, info) = result.expect("boundary observation");
+            assert_eq!(bytes, body);
+            assert_eq!(info.size, written.size);
+            if !body.is_empty() {
+                let result =
+                    crate::config::com::read_config_observation_no_lock(set.clone(), object, body.len() - 1, &guard).await;
+                assert_eq!(matches!(result, Err(Error::CorruptedFormat)), true, "I06L_BOUND_ENFORCED");
+            }
+        }
+        let object = "config/locked-absent.bin";
+        let namespace = set
+            .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, object)
+            .await
+            .expect("namespace");
+        let guard = crate::object_api::WriteCommitGuard::acquire(&namespace, std::time::Duration::from_secs(5))
+            .await
+            .expect("matching missing owner");
+        let result = crate::config::com::read_config_observation_no_lock(set, object, 0, &guard).await;
+        assert_eq!(matches!(result, Err(Error::ConfigNotFound)), true, "I06L_EMPTY_DISTINCT");
+    }
+
+    #[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 {
Author	SHA1	Message	Date
overtrue	f081082bd3	feat(ecstore): observe bounded config under retained write owner	2026-10-11 22:57:45 +08:00
overtrue	60d90f085a	feat(ecstore): add bounded persisted config observations	2026-10-11 20:29:27 +08:00
overtrue	dfa5d50565	feat(capture): persist first-decision provenance	2026-10-11 18:58:12 +08:00