Fix: keep sealed stage authority private during crash observation (#146)

CHANGELOG.md

@@ -8,6 +8,8 @@ after its public API and format compatibility policies are established.
 
 ## [Unreleased]
 
+- Sealed segment receipts no longer expose writable stages through `map_stage`. Repository crash injection uses a private-stage observation wrapper whose sealed conversion preserves stage identity and publisher authority without handing storage to callbacks (#146).
+
 - Incomplete version-one segment seals now refuse observed fixed-framing corruption before recovery assessment can authorize discard, preserving precise seal diagnostics (#171).
 
 - Retention recovery execution errors report the exact failed boundary, original typed cause, known namespace effects and uncertain effect/durability; retries freshly observe the store. Observed stage identity remains binding across reopening, and cleanup preserves verified pool evidence rather than promising the removed pathname survives (#99).

docs/formats/segment-store-v1/rationale.md

@@ -176,6 +176,10 @@ filesystem whose individual operations returned success. Each would let an
 unsupported platform manufacture the authority that the proof is meant to
 represent.
 
+## Sealed-stage observation
+
+Repository segment observation keeps writable stage authority private (#146). The former sealed `map_stage` callback could mutate an already synchronized stage while keeping the original receipt metadata; publication-time revalidation did not make that receipt truthful. `ObservedSegmentStage` now owns the stage and performs its actual write, flush and synchronization operations, exposing only lengths and before/after durability events to its observer. Its specialized sealed `without_observer` conversion drops the observer and preserves the same hidden stage and metadata together. There is no arbitrary user conversion of a sealed stage. Returning an error from an after-write hook cannot undo bytes; an `Interrupted` cause is retained inside a non-retryable I/O error so ordinary write retry cannot duplicate those effects. A new arbitrary unwrap trait or caller-supplied closure would reopen the capability escape and was rejected.
+
 ## Observation before recovery
 
 Store opening performs no repair. It produces either one verified reader

docs/formats/segment-store-v1/requirements.md

@@ -47,7 +47,7 @@ retention, or garbage collection.
 | `KEEP-SEGMENT-002` | Staged and sealed writer states are distinct, consuming types | `tests/segment_writer.rs` | Implemented in #15 |
 | `KEEP-SEGMENT-003` | Short, interrupted, zero-progress, invalid-count, storage, permission, flush, and synchronization failures retain exact phases and offsets | `tests/segment_writer/write_contract_laws.rs`, `tests/segment_writer/refusal_laws.rs`, `tests/segment_writer/durability_laws.rs` | Implemented in #15 |
 | `KEEP-SEGMENT-004` | Complete-segment admission verifies bounds, framing, checksums, logical identities, duplicate refusal, terminal state, and physical digest before exposure | `tests/segment.rs`, `tests/segment/identity_laws.rs`, `tests/segment/framing_laws.rs` | Implemented in #15 |
-| `KEEP-SEGMENT-005` | The public sealed receipt exposes no mutable stage handle | `src/adapters/sealed_segment.rs` | Production API implemented in #15; repository-task escape tracked in #146 |
+| `KEEP-SEGMENT-005` | The public sealed receipt exposes no mutable stage handle, including with repository-task observation enabled | `src/adapters/sealed_segment.rs` compile-fail law; `tests/observed_segment_stage.rs`; [capability evidence](../../testing-evidence/sealed-stage-observation.md) | Production API implemented in #15; repository-task escape removed for #146 |
 | `KEEP-SEGMENT-006` | Malformed, unsupported, partial, conflicting, and corrupt input returns boundary-typed errors | `tests/segment_header/mutation_laws.rs`, `tests/segment_record_header/framing_laws.rs`, `tests/segment_seal/framing_laws.rs`, `tests/segment/identity_laws.rs` | Implemented in #15 |
 | `KEEP-SEGMENT-007` | Record, nested-layout, segment-length, and temporary identity-index allocation remain explicitly bounded | `tests/segment_memory.rs`, `tests/segment_record_memory.rs`, `tests/segment_seal_memory.rs` | Implemented in #15 |
 | `KEEP-SEGMENT-008` | Filesystem staging uses exclusive fixed-name creation and never enumerates storage as a content index | `src/adapters/filesystem_segment_stage_tests.rs`, `src/adapters/filesystem_segment_stage.rs` | Implemented in #15 |

docs/testing-evidence/sealed-stage-observation.md

@@ -0,0 +1,30 @@
+# Sealed-stage observation capability
+
+Change kind: bug fix for #146, with a cohesive replacement observation boundary for its crash-harness consumer. Owner: `@flyingrobots`. The branch starts at main `6051abb25a9fd33ae7ee0de5614514b709a4d82a`; it is independent of the public-stage and catalog-admission follow-up branches. No on-disk format, identity, publication phase or new dependency is introduced. Removing the feature-gated public `map_stage` method intentionally breaks callers that requested writable authority from an already sealed receipt.
+
+## RED and protected boundary
+
+The permanent compile-fail example in `SealedSegment` attempts to call `map_stage` and write another byte through its callback. On the unfixed parent, `cargo test --doc --all-features sealed_segment --locked` failed with “Test compiled successfully, but it's marked compile_fail.” Regression commit `11be73d` preserves that RED. This is compiler-enforced capability evidence; it is not represented as a filesystem failure. With the arbitrary conversion removed, the same example passes by refusing access to that API.
+
+The replacement stores both stage and observer in private fields. Observation receives only requested/actual write lengths and explicit before/after flush/synchronize events. `without_observer` is implemented only for the built-in wrapper and returns another sealed receipt over the same stage; no callback or public extractor receives writable storage. Generic `close` still drops the stage. Filesystem selection still checks its original publisher authority and admitted segment coordinates. The usual `SegmentStage` exclusive-ownership precondition remains; a caller's independently duplicated storage handle is not made exclusive by a wrapper.
+
+## Runtime evidence and oracles
+
+| Claim | Evidence and oracle |
+| --- | --- |
+| Removing observation preserves bytes and private publisher provenance | Actual admitted ext4 stage equals the independently specified empty-segment golden; selection by its original publisher succeeds after observer removal. |
+| Short-write observation preserves ordinary writer output over the tested input family | Generated payload lengths 1 through 64, with a fixed repeating byte pattern, compare observed seven-byte-prefix writes against plain filesystem writing. Every case owns fresh storage. Ascending exhaustive exploration reports the smallest failing length in this family; the independent golden complements this shared-implementation differential oracle. |
+| Excessive observer limits refuse before writes | Exact header `SegmentWriteError::Write`, zero prior completed bytes and `InvalidInput`; the actual stage remains empty. |
+| A post-write interruption cannot retry completed effects | Exact header write refusal retains the nested original `Interrupted` cause, with a non-retryable outer kind; actual bytes equal exactly one golden header. The prior-completed-call offset is not a rollback claim for this failed call. |
+| Observation cannot fabricate successful durability | Faulting stage ports drive the production writer through the observation wrapper; exact prefix flush/synchronize failures preserve errno 5 and produce no sealed receipt. These are port-level fault simulations, not physical disk failure evidence. |
+| Crash injection remains connected to production operations | Complete existing debug and optimized process-death matrices pass after replacing the writable decorator with `CrashSegmentObserver`. Header, record and seal interruption offsets, before/during/after choices, flush/sync ordering, restart bytes and immutable admission oracles remain unchanged. |
+
+Separate copied mutations demonstrated runtime RED for skipped underlying flush, skipped underlying synchronization, clamped excessive limits, retryable post-effect interruption, and zeroed writes. Each used a dedicated build directory. Zeroed writes also failed the generated differential assertion at its first payload length. These are distinct outcome calibrations, not a mutation score or harness-case-count claim. The mutation sources were not committed.
+
+## Execution and limits
+
+All Rust execution uses copied Docker source, pinned Rust 1.96.0 and dedicated output on Linux aarch64. Filesystem laws use actual production platform admission on private ext4 scratch; no writable host checkout mount or fabricated admission proof is used. The capability regression is run with `--all-features`, since that is what exposed the old method. Both feature configurations are checked with warnings-denied Clippy. The final PR records exact candidate and hosted-check SHAs alongside full debug/release, formatting, source-structure, documentation and existing golden/corruption/memory evidence.
+
+Filesystem laws are medium; the faulting durability-port laws and capability example are small/static evidence. No random seeds, sleeps or uncontrolled scheduling drive the oracles. The differential input family is bounded and deterministic; it does not establish equivalence for arbitrary payloads or malicious storage implementations. Replay with `cargo test --test observed_segment_stage --all-features --locked`, its `--release` variant, and the all-feature doctest command above. Re-run the production campaigns with `cargo xtask durability-crash-matrix` and its optimized invocation.
+
+Existing resource-enforcement and suite-budget gaps remain those disclosed by the binding testing enforcement profile; this change claims neither a new sandbox policy nor measured latency SLOs. Process death and simulated errno failures do not establish physical power-loss behavior. No performance optimization is claimed. Retire the laws only if the capability is removed or stronger boundary evidence demonstrably subsumes the risk. Original roadmap checkboxes and unrelated recovery/admission scope remain unchanged.

src/adapters/exports.rs

@@ -64,6 +64,8 @@ pub use super::layout_encode_error::LayoutEncodeError;
 pub use super::layout_id_binary_error::LayoutIdBinaryParseError;
 pub use super::layout_id_text_error::LayoutIdTextParseError;
 pub use super::layout_record::CanonicalLayoutRecord;
+#[cfg(feature = "repository-tasks")]
+pub use super::observed_segment_stage::ObservedSegmentStage;
 pub use super::opened_reusable_segment::OpenedReusableSegment;
 pub use super::publication_head_decode_error::PublicationHeadDecodeError;
 pub use super::recovery::*;
@@ -92,6 +94,8 @@ pub use super::segment_seal::SegmentSeal;
 pub use super::segment_seal_error::SegmentSealError;
 pub use super::segment_stage::SegmentStage;
 pub use super::segment_stage_create_error::SegmentStageCreateError;
+#[cfg(feature = "repository-tasks")]
+pub use super::segment_stage_observer::{SegmentStageDurabilityEvent, SegmentStageObserver};
 pub use super::segment_write_error::SegmentWriteError;
 pub use super::segment_write_phase::{SegmentDurabilityPhase, SegmentWritePhase};
 pub use super::staged_segment::StagedSegment;

src/adapters/mod.rs

@@ -151,6 +151,8 @@ mod layout_record_format;
 mod layout_record_framing;
 mod loaded_segment;
 mod lower_hex;
+#[cfg(feature = "repository-tasks")]
+mod observed_segment_stage;
 mod opened_reusable_segment;
 mod physical_pool_name;
 mod publication_head_decode_error;
@@ -213,6 +215,8 @@ mod segment_seal_hash;
 mod segment_stage;
 mod segment_stage_create_error;
 mod segment_stage_create_error_display;
+#[cfg(feature = "repository-tasks")]
+mod segment_stage_observer;
 mod segment_stage_write;
 mod segment_write_error;
 mod segment_write_error_display;

src/adapters/observed_segment_stage.rs

@@ -0,0 +1,85 @@
+//! This module owns transparent stage observation with private writable authority.
+
+use std::io::{self, Write};
+
+use super::{SealedSegment, SegmentStage, SegmentStageDurabilityEvent, SegmentStageObserver};
+
+/// A repository stage decorator whose observer never receives its inner stage.
+///
+/// Owns the stage until close or sealing. Construction allocates nothing; actual
+/// writes and durability operations block according to the supplied stage and
+/// observer. The supplied stage must satisfy [`SegmentStage`]'s ownership contract.
+#[must_use]
+pub struct ObservedSegmentStage<S, O> {
+    stage: S,
+    observer: O,
+}
+
+impl<S, O> ObservedSegmentStage<S, O> {
+    /// Installs observation before the stage is consumed by the writer.
+    pub const fn new(stage: S, observer: O) -> Self {
+        Self { stage, observer }
+    }
+}
+
+impl<S: SegmentStage, O: SegmentStageObserver> Write for ObservedSegmentStage<S, O> {
+    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+        let allowed = self.observer.before_write(bytes.len())?;
+        let prefix = bytes.get(..allowed).ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "observer write limit exceeds input",
+            )
+        })?;
+        let written = self.stage.write(prefix)?;
+        if written > prefix.len() {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "stage exceeded observed write limit",
+            ));
+        }
+        self.observer.after_write(written).map_err(|source| {
+            // Interrupted means no write happened to Write::write_all. The
+            // observer runs after real effects, so preserve the cause without
+            // allowing a caller to silently repeat those bytes.
+            if source.kind() == io::ErrorKind::Interrupted {
+                io::Error::other(source)
+            } else {
+                source
+            }
+        })?;
+        Ok(written)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.observer
+            .durability(SegmentStageDurabilityEvent::BeforeFlush)?;
+        self.stage.flush()?;
+        self.observer
+            .durability(SegmentStageDurabilityEvent::AfterFlush)
+    }
+}
+
+impl<S: SegmentStage, O: SegmentStageObserver> SegmentStage for ObservedSegmentStage<S, O> {
+    fn synchronize(&mut self) -> io::Result<()> {
+        self.observer
+            .durability(SegmentStageDurabilityEvent::BeforeSynchronize)?;
+        self.stage.synchronize()?;
+        self.observer
+            .durability(SegmentStageDurabilityEvent::AfterSynchronize)
+    }
+}
+
+impl<S: SegmentStage, O: SegmentStageObserver> SealedSegment<ObservedSegmentStage<S, O>> {
+    /// Removes observation while keeping the sealed stage inaccessible to callers.
+    ///
+    /// Drops only the observer. The original stage, exact metadata and completed
+    /// durability evidence stay together; no user callback receives the stage.
+    /// Filesystem publication still checks actual bytes and publisher authority.
+    pub fn without_observer(self) -> SealedSegment<S> {
+        let (wrapped, count, length, digest) = self.into_parts();
+        let ObservedSegmentStage { stage, observer } = wrapped;
+        drop(observer);
+        SealedSegment::admitted(stage, count, length, digest)
+    }
+}

src/adapters/sealed_segment.rs

@@ -7,6 +7,19 @@ use super::{ClosedSegment, SegmentDigest, SegmentStage};
 ///
 /// The stage remains unpublished. This type exposes no mutable stage handle,
 /// makes no directory-durability claim, and is not a catalog reference.
+///
+/// Sealing cannot hand writable authority to a conversion callback, including
+/// with `repository-tasks` enabled:
+///
+/// ```compile_fail
+/// use keep::{SealedSegment, SegmentStage};
+/// fn rewrite<S: SegmentStage>(sealed: SealedSegment<S>) {
+///     let _ = sealed.map_stage(|mut stage| {
+///         let _ = std::io::Write::write_all(&mut stage, b"!");
+///         stage
+///     });
+/// }
+/// ```
 #[must_use]
 pub struct SealedSegment<S>
 where
@@ -53,22 +66,6 @@ where
         self.digest
     }
 
-    /// Replaces a repository-only storage decorator while preserving sealed
-    /// metadata.
-    ///
-    /// This supports transparent storage-port decorators. The mapping does not
-    /// change, revalidate, or publish the sealed bytes; authority-bound
-    /// adapters still validate the returned stage before publication.
-    #[cfg(feature = "repository-tasks")]
-    #[doc(hidden)]
-    pub fn map_stage<T>(self, map: impl FnOnce(S) -> T) -> SealedSegment<T>
-    where
-        T: SegmentStage,
-    {
-        let (stage, record_count, segment_length, digest) = self.into_parts();
-        SealedSegment::admitted(map(stage), record_count, segment_length, digest)
-    }
-
     pub(super) fn into_parts(self) -> (S, u32, u64, SegmentDigest) {
         let Self {
             _stage: stage,

src/adapters/segment_stage_observer.rs

@@ -0,0 +1,41 @@
+//! This module owns repository fault-observation hooks without stage authority.
+
+use std::io;
+
+/// An observation boundary around a real stage durability operation.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum SegmentStageDurabilityEvent {
+    /// Before flushing buffered bytes.
+    BeforeFlush,
+    /// After a successful flush.
+    AfterFlush,
+    /// Before synchronizing persisted bytes.
+    BeforeSynchronize,
+    /// After a successful synchronization.
+    AfterSynchronize,
+}
+
+/// Repository fault hooks that can interrupt operations but cannot access storage.
+///
+/// The wrapper always performs actual writes, flushes and synchronization itself.
+/// Observers can block or fail at a boundary, or limit a write to a strict prefix.
+/// No hook receives a writable stage or can authorize a sealed receipt.
+pub trait SegmentStageObserver {
+    /// Selects how many of the requested bytes may reach the next actual write.
+    ///
+    /// # Errors
+    /// Returns the injected pre-write failure. Limits beyond `requested` are refused.
+    fn before_write(&mut self, requested: usize) -> io::Result<usize>;
+
+    /// Observes the actual successful write count before the caller continues.
+    ///
+    /// # Errors
+    /// Returns an injected failure after the write; this does not undo its effects.
+    fn after_write(&mut self, written: usize) -> io::Result<()>;
+
+    /// Observes a boundary before or after an actual durability operation.
+    ///
+    /// # Errors
+    /// Returns the injected failure at this boundary, without rolling back effects.
+    fn durability(&mut self, event: SegmentStageDurabilityEvent) -> io::Result<()>;
+}

src/lib.rs

@@ -55,9 +55,6 @@ mod profile;
 mod reference;
 mod retention;
 
-#[cfg(feature = "repository-tasks")]
-#[doc(hidden)]
-pub use adapters::RepositoryInitializationStorage;
 pub use adapters::{
     AdmittedCatalog, AdmittedRecoveryStageBytes, AdmittedSegment, AdmittedSegmentRecord,
     AdmittedStoreFormatMarker, AdmittedStoreMigrationIntent, AdmittedStoreMigrationReceipt,
@@ -167,6 +164,12 @@ pub use adapters::{
     StoreMigrationRecoveryReceipt, StoreMigrationRecoveryStorage, StoreMigrationResidue,
     StoreMigrationStageDecodeError, plan_store_migration_recovery, recover_store_migration,
 };
+#[cfg(feature = "repository-tasks")]
+#[doc(hidden)]
+pub use adapters::{
+    ObservedSegmentStage, RepositoryInitializationStorage, SegmentStageDurabilityEvent,
+    SegmentStageObserver,
+};
 pub use blob::{
     BlobHashError, BlobHasher, BlobId, BlobLength, BlobReadError, ByteLength, ByteOffset,
     ByteRange, ByteRangeError,