Add bounded streaming execution kernel

Cargo.lock

@@ -1612,6 +1612,18 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "remora-stream"
+version = "0.0.0"
+dependencies = [
+ "getrandom 0.3.4",
+ "remora-domain",
+ "serde",
+ "sha2 0.11.0",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "remora-worker"
 version = "0.0.0"

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "crates/remora-domain",
     "crates/remora-durable",
     "crates/remora-identity",
+    "crates/remora-stream",
     "crates/remora-worker",
 ]
 resolver = "3"

Makefile

@@ -1,4 +1,4 @@
-.PHONY: check-dependencies check-fixture-lib check-kind-sample check-durable check-coordination check-identity check-credentials check-fast check-full check-fast-host check-full-host rust-check web-install web-check web-test-fast web-test-full
+.PHONY: check-dependencies check-fixture-lib check-kind-sample check-durable check-coordination check-identity check-credentials check-stream check-fast check-full check-fast-host check-full-host rust-check web-install web-check web-test-fast web-test-full
 
 CONTAINER_ENGINE ?= nerdctl
 CONTAINER_RUN_FLAGS ?= --memory=16g --memory-swap=24g --cpus=4
@@ -67,3 +67,9 @@ check-identity:
 
 check-credentials:
 	./tests/kind/run-credentials-job.sh
+
+check-stream:
+	cargo fmt --all --check
+	cargo clippy --package remora-stream --all-targets --all-features -- -D warnings
+	cargo nextest run --package remora-stream --all-features
+	cargo test --package remora-stream --doc

crates/remora-stream/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "remora-stream"
+edition.workspace = true
+publish = false
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+getrandom = "0.3.4"
+remora-domain.workspace = true
+serde = { workspace = true, features = ["derive"] }
+sha2.workspace = true
+thiserror.workspace = true
+tokio = { workspace = true, features = ["macros", "sync", "time"] }
+
+[dev-dependencies]
+sha2.workspace = true
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }

crates/remora-stream/src/cancel.rs

@@ -0,0 +1,51 @@
+use tokio::sync::watch;
+
+/// Cooperative cancellation token for the copy loop and bounded I/O.
+///
+/// Cancel wakes waiters so in-flight operations bounded by
+/// [`crate::copy_stream`] can return without waiting for their full
+/// no-progress timeout.
+///
+/// A receiver is retained so `watch::Sender::send` always has a live
+/// subscriber and the cancelled flag is stored.
+#[derive(Clone, Debug)]
+pub struct CancelToken {
+    tx: watch::Sender<bool>,
+    rx: watch::Receiver<bool>,
+}
+
+impl Default for CancelToken {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl CancelToken {
+    #[must_use]
+    pub fn new() -> Self {
+        let (tx, rx) = watch::channel(false);
+        Self { tx, rx }
+    }
+
+    pub fn cancel(&self) {
+        let _ = self.tx.send(true);
+    }
+
+    #[must_use]
+    pub fn is_cancelled(&self) -> bool {
+        *self.rx.borrow()
+    }
+
+    /// Completes when [`Self::cancel`] has been called.
+    pub async fn cancelled(&self) {
+        let mut rx = self.rx.clone();
+        if *rx.borrow_and_update() {
+            return;
+        }
+        while rx.changed().await.is_ok() {
+            if *rx.borrow_and_update() {
+                return;
+            }
+        }
+    }
+}

crates/remora-stream/src/copy.rs

@@ -0,0 +1,455 @@
+use std::future::Future;
+use std::time::{Duration, Instant};
+
+use tokio::time::timeout;
+
+use crate::cancel::CancelToken;
+use crate::error::{StreamError, StreamSide};
+use crate::failure::{StructuredFailure, classify_stream_error};
+use crate::integrity::{IntegrityState, Sha256Accountant};
+use crate::ports::{SourceBody, TargetBody, TargetFinish};
+use crate::progress::{TransferCheckpoint, TransferProgress};
+
+/// Race an I/O future against cancellation and the no-progress bound.
+async fn bounded_io<F, T>(
+    cancel: &CancelToken,
+    bound: Duration,
+    side: StreamSide,
+    fut: F,
+) -> Result<T, StreamError>
+where
+    F: Future<Output = Result<T, StreamError>>,
+{
+    let started = Instant::now();
+    tokio::select! {
+        biased;
+        () = cancel.cancelled() => Err(StreamError::Cancelled),
+        result = timeout(bound, fut) => match result {
+            Ok(inner) => inner,
+            Err(_) => {
+                let elapsed_ms =
+                    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
+                Err(StreamError::NoProgress { side, elapsed_ms })
+            }
+        },
+    }
+}
+
+/// Bounded-buffer copy configuration.
+///
+/// Memory used by the kernel for payload data is one buffer of `buffer_size`
+/// bytes plus fixed integrity state, independent of total transfer size.
+///
+/// `initial_completed_bytes` is the absolute prefix already retained before this
+/// invocation. When it is non-zero, callers must seed integrity with
+/// [`copy_stream_seeded`] so the accountant's `bytes_hashed` matches that
+/// prefix; otherwise the copy fails closed.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CopyConfig {
+    pub buffer_size: usize,
+    pub progress_interval_bytes: u64,
+    pub checkpoint_interval_bytes: u64,
+    pub no_progress_timeout: Duration,
+    pub initial_completed_bytes: u64,
+    pub total_bytes: Option<u64>,
+    pub completed_members: u64,
+    pub total_members: Option<u64>,
+}
+
+impl CopyConfig {
+    pub const DEFAULT_BUFFER_SIZE: usize = 64 * 1024;
+    pub const MIN_BUFFER_SIZE: usize = 4 * 1024;
+    pub const MAX_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+    #[must_use]
+    pub fn sanitized(self) -> Self {
+        let buffer_size = self
+            .buffer_size
+            .clamp(Self::MIN_BUFFER_SIZE, Self::MAX_BUFFER_SIZE);
+        let progress_interval_bytes = if self.progress_interval_bytes == 0 {
+            buffer_size as u64
+        } else {
+            self.progress_interval_bytes
+        };
+        let checkpoint_interval_bytes = if self.checkpoint_interval_bytes == 0 {
+            progress_interval_bytes.saturating_mul(16)
+        } else {
+            self.checkpoint_interval_bytes
+        };
+        let no_progress_timeout = if self.no_progress_timeout.is_zero() {
+            Duration::from_secs(60)
+        } else {
+            self.no_progress_timeout
+        };
+        Self {
+            buffer_size,
+            progress_interval_bytes,
+            checkpoint_interval_bytes,
+            no_progress_timeout,
+            initial_completed_bytes: self.initial_completed_bytes,
+            total_bytes: self.total_bytes,
+            completed_members: self.completed_members,
+            total_members: self.total_members,
+        }
+    }
+}
+
+impl Default for CopyConfig {
+    fn default() -> Self {
+        Self {
+            buffer_size: Self::DEFAULT_BUFFER_SIZE,
+            progress_interval_bytes: Self::DEFAULT_BUFFER_SIZE as u64,
+            checkpoint_interval_bytes: (Self::DEFAULT_BUFFER_SIZE as u64).saturating_mul(16),
+            no_progress_timeout: Duration::from_secs(60),
+            initial_completed_bytes: 0,
+            total_bytes: None,
+            completed_members: 0,
+            total_members: None,
+        }
+    }
+}
+
+/// Hooks invoked when progress or checkpoint thresholds advance.
+pub trait CopyHooks: Send {
+    fn on_progress(
+        &mut self,
+        progress: &TransferProgress,
+    ) -> impl Future<Output = Result<(), StreamError>> + Send;
+
+    fn on_checkpoint(
+        &mut self,
+        checkpoint: &TransferCheckpoint,
+    ) -> impl Future<Output = Result<(), StreamError>> + Send;
+}
+
+impl<H: CopyHooks + ?Sized> CopyHooks for &mut H {
+    fn on_progress(
+        &mut self,
+        progress: &TransferProgress,
+    ) -> impl Future<Output = Result<(), StreamError>> + Send {
+        (**self).on_progress(progress)
+    }
+
+    fn on_checkpoint(
+        &mut self,
+        checkpoint: &TransferCheckpoint,
+    ) -> impl Future<Output = Result<(), StreamError>> + Send {
+        (**self).on_checkpoint(checkpoint)
+    }
+}
+
+/// No-op hooks for callers that record progress elsewhere.
+pub struct NoopHooks;
+
+impl CopyHooks for NoopHooks {
+    async fn on_progress(&mut self, _progress: &TransferProgress) -> Result<(), StreamError> {
+        Ok(())
+    }
+
+    async fn on_checkpoint(&mut self, _checkpoint: &TransferCheckpoint) -> Result<(), StreamError> {
+        Ok(())
+    }
+}
+
+/// Successful completion of one member copy.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CopyResult {
+    pub progress: TransferProgress,
+    pub integrity: IntegrityState,
+    pub target: TargetFinish,
+    /// Configured payload buffer length used for this copy.
+    pub buffer_len: usize,
+    /// Reserved capacity of the payload buffer (never grown during the copy).
+    pub buffer_capacity: usize,
+}
+
+/// Terminal result of a bounded copy.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum CopyOutcome {
+    Completed(CopyResult),
+    Cancelled {
+        progress: TransferProgress,
+        integrity: IntegrityState,
+        buffer_len: usize,
+        buffer_capacity: usize,
+    },
+    Failed {
+        failure: StructuredFailure,
+        progress: TransferProgress,
+        integrity: IntegrityState,
+        buffer_len: usize,
+        buffer_capacity: usize,
+    },
+}
+
+/// Copy from `source` into `target` with a single bounded buffer and a fresh
+/// integrity accountant (session starts at zero hashed bytes).
+pub async fn copy_stream<S, T, H>(
+    source: S,
+    target: T,
+    config: CopyConfig,
+    cancel: &CancelToken,
+    hooks: H,
+) -> CopyOutcome
+where
+    S: SourceBody,
+    T: TargetBody,
+    H: CopyHooks,
+{
+    copy_stream_seeded(
+        source,
+        target,
+        config,
+        cancel,
+        hooks,
+        Sha256Accountant::new(),
+    )
+    .await
+}
+
+/// Copy with a pre-seeded integrity accountant for resume.
+///
+/// `accountant.bytes_hashed()` must equal `config.initial_completed_bytes`. A
+/// mismatch fails closed so absolute progress and integrity stay consistent.
+pub async fn copy_stream_seeded<S, T, H>(
+    mut source: S,
+    mut target: T,
+    config: CopyConfig,
+    cancel: &CancelToken,
+    mut hooks: H,
+    mut accountant: Sha256Accountant,
+) -> CopyOutcome
+where
+    S: SourceBody,
+    T: TargetBody,
+    H: CopyHooks,
+{
+    let config = config.sanitized();
+    let mut buffer = vec![0u8; config.buffer_size];
+    let buffer_len = buffer.len();
+    let buffer_capacity = buffer.capacity();
+    let initial = config.initial_completed_bytes;
+    let mut completed_bytes = initial;
+    let mut session_bytes = 0u64;
+    let mut last_progress_at = initial;
+    let mut last_checkpoint_at = initial;
+
+    let progress_snapshot = |completed_bytes: u64, session_bytes: u64| {
+        TransferProgress::new(
+            completed_bytes,
+            session_bytes,
+            config.total_bytes,
+            config.completed_members,
+            config.total_members,
+        )
+    };
+
+    let fail = |error: StreamError,
+                completed_bytes: u64,
+                session_bytes: u64,
+                accountant: Sha256Accountant| {
+        let progress = progress_snapshot(completed_bytes, session_bytes);
+        let integrity = accountant.finalize();
+        if error.is_cancelled() {
+            CopyOutcome::Cancelled {
+                progress,
+                integrity,
+                buffer_len,
+                buffer_capacity,
+            }
+        } else {
+            CopyOutcome::Failed {
+                failure: classify_stream_error(&error),
+                progress,
+                integrity,
+                buffer_len,
+                buffer_capacity,
+            }
+        }
+    };
+
+    if accountant.bytes_hashed() != initial {
+        return fail(
+            StreamError::internal(format!(
+                "integrity seed covers {} bytes but initial_completed_bytes is {initial}",
+                accountant.bytes_hashed()
+            )),
+            completed_bytes,
+            session_bytes,
+            accountant,
+        );
+    }
+
+    loop {
+        if cancel.is_cancelled() {
+            return fail(
+                StreamError::Cancelled,
+                completed_bytes,
+                session_bytes,
+                accountant,
+            );
+        }
+
+        let n = match bounded_io(
+            cancel,
+            config.no_progress_timeout,
+            StreamSide::Source,
+            source.read(&mut buffer),
+        )
+        .await
+        {
+            Ok(0) => break,
+            Ok(n) => n,
+            Err(error) => {
+                return fail(error, completed_bytes, session_bytes, accountant);
+            }
+        };
+
+        if n > buffer.len() {
+            return fail(
+                StreamError::internal(format!(
+                    "source read returned {n} bytes into a {}-byte buffer",
+                    buffer.len()
+                )),
+                completed_bytes,
+                session_bytes,
+                accountant,
+            );
+        }
+
+        if cancel.is_cancelled() {
+            return fail(
+                StreamError::Cancelled,
+                completed_bytes,
+                session_bytes,
+                accountant,
+            );
+        }
+
+        let chunk = &buffer[..n];
+        if let Err(error) = bounded_io(
+            cancel,
+            config.no_progress_timeout,
+            StreamSide::Target,
+            target.write_all(chunk),
+        )
+        .await
+        {
+            return fail(error, completed_bytes, session_bytes, accountant);
+        }
+
+        accountant.update(chunk);
+        session_bytes = session_bytes.saturating_add(n as u64);
+        completed_bytes = initial.saturating_add(session_bytes);
+
+        if completed_bytes.saturating_sub(last_progress_at) >= config.progress_interval_bytes {
+            let progress = progress_snapshot(completed_bytes, session_bytes);
+            if let Err(error) = hooks.on_progress(&progress).await {
+                return fail(error, completed_bytes, session_bytes, accountant);
+            }
+            last_progress_at = completed_bytes;
+            if cancel.is_cancelled() {
+                return fail(
+                    StreamError::Cancelled,
+                    completed_bytes,
+                    session_bytes,
+                    accountant,
+                );
+            }
+        }
+
+        if completed_bytes.saturating_sub(last_checkpoint_at) >= config.checkpoint_interval_bytes {
+            let checkpoint =
+                TransferCheckpoint::new(completed_bytes, session_bytes, accountant.snapshot());
+            if let Err(error) = hooks.on_checkpoint(&checkpoint).await {
+                return fail(error, completed_bytes, session_bytes, accountant);
+            }
+            last_checkpoint_at = completed_bytes;
+            if cancel.is_cancelled() {
+                return fail(
+                    StreamError::Cancelled,
+                    completed_bytes,
+                    session_bytes,
+                    accountant,
+                );
+            }
+        }
+    }
+
+    if cancel.is_cancelled() {
+        return fail(
+            StreamError::Cancelled,
+            completed_bytes,
+            session_bytes,
+            accountant,
+        );
+    }
+
+    // Final progress/checkpoint at EOF when bytes advanced past the last report.
+    if completed_bytes > last_progress_at || session_bytes == 0 {
+        let progress = progress_snapshot(completed_bytes, session_bytes);
+        if let Err(error) = hooks.on_progress(&progress).await {
+            return fail(error, completed_bytes, session_bytes, accountant);
+        }
+        if cancel.is_cancelled() {
+            return fail(
+                StreamError::Cancelled,
+                completed_bytes,
+                session_bytes,
+                accountant,
+            );
+        }
+    }
+    if completed_bytes > last_checkpoint_at && session_bytes > 0 {
+        let checkpoint =
+            TransferCheckpoint::new(completed_bytes, session_bytes, accountant.snapshot());
+        if let Err(error) = hooks.on_checkpoint(&checkpoint).await {
+            return fail(error, completed_bytes, session_bytes, accountant);
+        }
+        if cancel.is_cancelled() {
+            return fail(
+                StreamError::Cancelled,
+                completed_bytes,
+                session_bytes,
+                accountant,
+            );
+        }
+    }
+
+    let integrity = accountant.finalize();
+    let progress = progress_snapshot(completed_bytes, session_bytes);
+    match bounded_io(
+        cancel,
+        config.no_progress_timeout,
+        StreamSide::Target,
+        target.finish(),
+    )
+    .await
+    {
+        Ok(target_finish) => CopyOutcome::Completed(CopyResult {
+            progress,
+            integrity,
+            target: target_finish,
+            buffer_len,
+            buffer_capacity,
+        }),
+        Err(error) => {
+            if error.is_cancelled() {
+                CopyOutcome::Cancelled {
+                    progress,
+                    integrity,
+                    buffer_len,
+                    buffer_capacity,
+                }
+            } else {
+                CopyOutcome::Failed {
+                    failure: classify_stream_error(&error),
+                    progress,
+                    integrity,
+                    buffer_len,
+                    buffer_capacity,
+                }
+            }
+        }
+    }
+}

crates/remora-stream/src/error.rs

@@ -0,0 +1,116 @@
+use std::fmt;
+
+use remora_domain::RuntimeFailureClass;
+
+/// Which I/O boundary produced a stream error.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum StreamSide {
+    Source,
+    Target,
+}
+
+/// Kernel and adapter stream errors before public redaction.
+///
+/// Internal diagnostic text may be retained for operator logs only when the
+/// adapter has already scrubbed secrets. Public projections use
+/// [`crate::classify_stream_error`].
+#[derive(Debug, thiserror::Error)]
+pub enum StreamError {
+    #[error("authorization failure on {side:?}")]
+    Authorization { side: StreamSide },
+    #[error("source not found")]
+    SourceNotFound,
+    #[error("source changed")]
+    SourceChanged,
+    #[error("transient failure on {side:?}: {detail}")]
+    Transient { side: StreamSide, detail: String },
+    #[error("integrity failure: {detail}")]
+    Integrity { detail: String },
+    #[error("capacity exhausted on {side:?}")]
+    Capacity { side: StreamSide },
+    #[error("internal stream error: {detail}")]
+    Internal { detail: String },
+    #[error("no progress for {elapsed_ms}ms on {side:?}")]
+    NoProgress { side: StreamSide, elapsed_ms: u64 },
+    #[error("transfer cancelled")]
+    Cancelled,
+}
+
+impl StreamError {
+    #[must_use]
+    pub fn side(&self) -> Option<StreamSide> {
+        match self {
+            Self::Authorization { side }
+            | Self::Transient { side, .. }
+            | Self::Capacity { side }
+            | Self::NoProgress { side, .. } => Some(*side),
+            Self::SourceNotFound | Self::SourceChanged | Self::Integrity { .. } => {
+                Some(StreamSide::Source)
+            }
+            Self::Internal { .. } | Self::Cancelled => None,
+        }
+    }
+
+    /// Map to an attempt-reportable runtime class when applicable.
+    ///
+    /// Cancellation is a protocol input and is not a runtime failure class.
+    #[must_use]
+    pub fn runtime_class(&self) -> Option<RuntimeFailureClass> {
+        match self {
+            Self::Authorization { .. } => Some(RuntimeFailureClass::Authorization),
+            Self::SourceNotFound => Some(RuntimeFailureClass::SourceNotFound),
+            Self::SourceChanged => Some(RuntimeFailureClass::SourceChanged),
+            Self::Transient {
+                side: StreamSide::Source,
+                ..
+            }
+            | Self::NoProgress {
+                side: StreamSide::Source,
+                ..
+            } => Some(RuntimeFailureClass::TransientSource),
+            Self::Transient {
+                side: StreamSide::Target,
+                ..
+            }
+            | Self::NoProgress {
+                side: StreamSide::Target,
+                ..
+            } => Some(RuntimeFailureClass::TransientTarget),
+            Self::Integrity { .. } => Some(RuntimeFailureClass::Integrity),
+            Self::Capacity { .. } => Some(RuntimeFailureClass::Capacity),
+            Self::Internal { .. } => Some(RuntimeFailureClass::Internal),
+            Self::Cancelled => None,
+        }
+    }
+
+    #[must_use]
+    pub fn is_cancelled(&self) -> bool {
+        matches!(self, Self::Cancelled)
+    }
+
+    pub fn transient_source(detail: impl fmt::Display) -> Self {
+        Self::Transient {
+            side: StreamSide::Source,
+            detail: detail.to_string(),
+        }
+    }
+
+    pub fn transient_target(detail: impl fmt::Display) -> Self {
+        Self::Transient {
+            side: StreamSide::Target,
+            detail: detail.to_string(),
+        }
+    }
+
+    pub fn integrity(detail: impl fmt::Display) -> Self {
+        Self::Integrity {
+            detail: detail.to_string(),
+        }
+    }
+
+    pub fn internal(detail: impl fmt::Display) -> Self {
+        Self::Internal {
+            detail: detail.to_string(),
+        }
+    }
+}

crates/remora-stream/src/failure.rs

@@ -0,0 +1,329 @@
+use std::collections::BTreeMap;
+
+use remora_domain::{FailureClass, FailureDisposition, RuntimeFailureClass};
+use serde::{Deserialize, Serialize};
+
+use crate::error::{StreamError, StreamSide};
+
+/// Allowlisted scalar for public failure details after typed construction.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(untagged)]
+pub enum SafeDetailValue {
+    String(String),
+    Integer(i64),
+    Boolean(bool),
+    Null,
+}
+
+/// Closed set of public reason codes that may appear in stream failure details.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum PublicReason {
+    NoProgress,
+}
+
+impl PublicReason {
+    #[must_use]
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::NoProgress => "no_progress",
+        }
+    }
+}
+
+/// Typed public details for a stream-classified failure.
+///
+/// Only fixed enum labels and integers are accepted. Free-form strings are not
+/// representable, so credentials, URLs, and exception text cannot enter the
+/// public projection through this type. Keys are further filtered per class.
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
+pub struct PublicDetails {
+    side: Option<StreamSide>,
+    reason: Option<PublicReason>,
+    elapsed_ms: Option<u64>,
+    byte_offset: Option<u64>,
+    member_index: Option<u64>,
+}
+
+impl PublicDetails {
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    #[must_use]
+    pub fn with_side(mut self, side: StreamSide) -> Self {
+        self.side = Some(side);
+        self
+    }
+
+    #[must_use]
+    pub fn with_reason(mut self, reason: PublicReason) -> Self {
+        self.reason = Some(reason);
+        self
+    }
+
+    #[must_use]
+    pub fn with_elapsed_ms(mut self, elapsed_ms: u64) -> Self {
+        self.elapsed_ms = Some(elapsed_ms);
+        self
+    }
+
+    #[must_use]
+    pub fn with_byte_offset(mut self, byte_offset: u64) -> Self {
+        self.byte_offset = Some(byte_offset);
+        self
+    }
+
+    #[must_use]
+    pub fn with_member_index(mut self, member_index: u64) -> Self {
+        self.member_index = Some(member_index);
+        self
+    }
+
+    fn render(self, class: FailureClass) -> BTreeMap<String, SafeDetailValue> {
+        let mut out = BTreeMap::new();
+        if key_allowed(class, "side")
+            && let Some(side) = self.side
+        {
+            out.insert(
+                "side".into(),
+                SafeDetailValue::String(side_label(side).into()),
+            );
+        }
+        if key_allowed(class, "reason")
+            && let Some(reason) = self.reason
+        {
+            out.insert(
+                "reason".into(),
+                SafeDetailValue::String(reason.as_str().into()),
+            );
+        }
+        if key_allowed(class, "elapsed_ms")
+            && let Some(elapsed_ms) = self.elapsed_ms
+        {
+            out.insert(
+                "elapsed_ms".into(),
+                SafeDetailValue::Integer(i64::try_from(elapsed_ms).unwrap_or(i64::MAX)),
+            );
+        }
+        if key_allowed(class, "byte_offset")
+            && let Some(byte_offset) = self.byte_offset
+        {
+            out.insert(
+                "byte_offset".into(),
+                SafeDetailValue::Integer(i64::try_from(byte_offset).unwrap_or(i64::MAX)),
+            );
+        }
+        if key_allowed(class, "member_index")
+            && let Some(member_index) = self.member_index
+        {
+            out.insert(
+                "member_index".into(),
+                SafeDetailValue::Integer(i64::try_from(member_index).unwrap_or(i64::MAX)),
+            );
+        }
+        out
+    }
+}
+
+/// Attempt-reportable or protocol-visible structured failure after redaction.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct StructuredFailure {
+    class: FailureClass,
+    message: String,
+    safe_details: BTreeMap<String, SafeDetailValue>,
+}
+
+impl StructuredFailure {
+    #[must_use]
+    pub fn class(&self) -> FailureClass {
+        self.class
+    }
+
+    #[must_use]
+    pub fn message(&self) -> &str {
+        &self.message
+    }
+
+    #[must_use]
+    pub fn safe_details(&self) -> &BTreeMap<String, SafeDetailValue> {
+        &self.safe_details
+    }
+
+    #[must_use]
+    pub fn disposition(&self) -> Option<FailureDisposition> {
+        match self.class {
+            FailureClass::Authorization
+            | FailureClass::SourceNotFound
+            | FailureClass::SourceChanged
+            | FailureClass::Validation
+            | FailureClass::Cancelled => Some(FailureDisposition::Permanent),
+            FailureClass::TransientSource
+            | FailureClass::TransientTarget
+            | FailureClass::Integrity
+            | FailureClass::Capacity
+            | FailureClass::Internal
+            | FailureClass::LostAttempt => Some(FailureDisposition::Retryable),
+        }
+    }
+
+    #[must_use]
+    pub fn runtime_class(&self) -> Option<RuntimeFailureClass> {
+        match self.class {
+            FailureClass::Authorization => Some(RuntimeFailureClass::Authorization),
+            FailureClass::SourceNotFound => Some(RuntimeFailureClass::SourceNotFound),
+            FailureClass::SourceChanged => Some(RuntimeFailureClass::SourceChanged),
+            FailureClass::TransientSource => Some(RuntimeFailureClass::TransientSource),
+            FailureClass::TransientTarget => Some(RuntimeFailureClass::TransientTarget),
+            FailureClass::Integrity => Some(RuntimeFailureClass::Integrity),
+            FailureClass::Capacity => Some(RuntimeFailureClass::Capacity),
+            FailureClass::Internal => Some(RuntimeFailureClass::Internal),
+            FailureClass::Validation | FailureClass::Cancelled | FailureClass::LostAttempt => None,
+        }
+    }
+
+    /// Build a public failure with the fixed class message and typed details.
+    ///
+    /// Caller-supplied free-form messages are not accepted. Detail keys are
+    /// filtered by class allowlist; values are only enum labels or integers.
+    #[must_use]
+    pub fn public(class: FailureClass, details: PublicDetails) -> Self {
+        Self {
+            class,
+            message: public_message_for_class(class).into(),
+            safe_details: details.render(class),
+        }
+    }
+
+    /// Redacted internal failure for unrecognized or unsafe errors.
+    #[must_use]
+    pub fn redacted_internal() -> Self {
+        Self::public(FailureClass::Internal, PublicDetails::new())
+    }
+}
+
+/// Map a stream error to a public structured failure.
+///
+/// Credentials, upstream bodies, raw URLs, and internal exception text are
+/// excluded from the public message and details.
+#[must_use]
+pub fn classify_stream_error(error: &StreamError) -> StructuredFailure {
+    match error {
+        StreamError::Cancelled => {
+            StructuredFailure::public(FailureClass::Cancelled, PublicDetails::new())
+        }
+        StreamError::Authorization { side } => StructuredFailure::public(
+            FailureClass::Authorization,
+            PublicDetails::new().with_side(*side),
+        ),
+        StreamError::SourceNotFound => {
+            StructuredFailure::public(FailureClass::SourceNotFound, PublicDetails::new())
+        }
+        StreamError::SourceChanged => {
+            StructuredFailure::public(FailureClass::SourceChanged, PublicDetails::new())
+        }
+        StreamError::Transient { side, .. } => {
+            let class = match side {
+                StreamSide::Source => FailureClass::TransientSource,
+                StreamSide::Target => FailureClass::TransientTarget,
+            };
+            StructuredFailure::public(class, PublicDetails::new().with_side(*side))
+        }
+        StreamError::NoProgress { side, elapsed_ms } => {
+            let class = match side {
+                StreamSide::Source => FailureClass::TransientSource,
+                StreamSide::Target => FailureClass::TransientTarget,
+            };
+            StructuredFailure::public(
+                class,
+                PublicDetails::new()
+                    .with_side(*side)
+                    .with_reason(PublicReason::NoProgress)
+                    .with_elapsed_ms(*elapsed_ms),
+            )
+        }
+        StreamError::Integrity { .. } => {
+            StructuredFailure::public(FailureClass::Integrity, PublicDetails::new())
+        }
+        StreamError::Capacity { side } => StructuredFailure::public(
+            FailureClass::Capacity,
+            PublicDetails::new().with_side(*side),
+        ),
+        StreamError::Internal { .. } => StructuredFailure::redacted_internal(),
+    }
+}
+
+#[must_use]
+pub fn public_message_for_class(class: FailureClass) -> &'static str {
+    match class {
+        FailureClass::Validation => "The request is invalid.",
+        FailureClass::Authorization => "Authorization failed for this transfer.",
+        FailureClass::SourceNotFound => "The source was not found.",
+        FailureClass::SourceChanged => "The source changed during transfer.",
+        FailureClass::TransientSource => "A temporary source error interrupted the transfer.",
+        FailureClass::TransientTarget => "A temporary target error interrupted the transfer.",
+        FailureClass::Integrity => "Transfer integrity validation failed.",
+        FailureClass::Capacity => "Capacity limits prevented the transfer from continuing.",
+        FailureClass::Internal => "An internal error interrupted the transfer.",
+        FailureClass::Cancelled => "The transfer was cancelled.",
+        FailureClass::LostAttempt => "The attempt was lost and may be retried.",
+    }
+}
+
+fn side_label(side: StreamSide) -> &'static str {
+    match side {
+        StreamSide::Source => "source",
+        StreamSide::Target => "target",
+    }
+}
+
+fn key_allowed(class: FailureClass, key: &str) -> bool {
+    match class {
+        FailureClass::TransientSource | FailureClass::TransientTarget => matches!(
+            key,
+            "side" | "reason" | "elapsed_ms" | "byte_offset" | "member_index"
+        ),
+        FailureClass::Authorization | FailureClass::Capacity => key == "side",
+        FailureClass::Integrity => matches!(key, "byte_offset" | "member_index"),
+        FailureClass::Validation
+        | FailureClass::SourceNotFound
+        | FailureClass::SourceChanged
+        | FailureClass::Internal
+        | FailureClass::Cancelled
+        | FailureClass::LostAttempt => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn public_constructor_uses_fixed_message_only() {
+        let failure = StructuredFailure::public(
+            FailureClass::TransientSource,
+            PublicDetails::new().with_reason(PublicReason::NoProgress),
+        );
+        assert_eq!(
+            failure.message(),
+            public_message_for_class(FailureClass::TransientSource)
+        );
+        assert!(!failure.message().contains("api"));
+        assert_eq!(
+            failure.safe_details().get("reason"),
+            Some(&SafeDetailValue::String("no_progress".into()))
+        );
+    }
+
+    #[test]
+    fn per_class_allowlist_drops_disallowed_keys() {
+        let failure = StructuredFailure::public(
+            FailureClass::Internal,
+            PublicDetails::new()
+                .with_side(StreamSide::Source)
+                .with_reason(PublicReason::NoProgress)
+                .with_elapsed_ms(12),
+        );
+        assert!(failure.safe_details().is_empty());
+    }
+}

crates/remora-stream/src/integrity.rs

@@ -0,0 +1,132 @@
+use sha2::{Digest, Sha256};
+
+/// Hex-encoded independent SHA-256 digest of transferred bytes.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct IntegrityDigest(String);
+
+impl IntegrityDigest {
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl std::fmt::Display for IntegrityDigest {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+/// Running integrity state for one transfer member.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct IntegrityState {
+    bytes_hashed: u64,
+    digest: Option<IntegrityDigest>,
+}
+
+impl IntegrityState {
+    #[must_use]
+    pub fn bytes_hashed(&self) -> u64 {
+        self.bytes_hashed
+    }
+
+    #[must_use]
+    pub fn digest(&self) -> Option<&IntegrityDigest> {
+        self.digest.as_ref()
+    }
+}
+
+/// Independent SHA-256 accountant updated as bytes leave the copy loop.
+#[derive(Clone)]
+pub struct Sha256Accountant {
+    hasher: Sha256,
+    bytes: u64,
+}
+
+impl Default for Sha256Accountant {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Sha256Accountant {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            hasher: Sha256::new(),
+            bytes: 0,
+        }
+    }
+
+    #[must_use]
+    pub fn bytes_hashed(&self) -> u64 {
+        self.bytes
+    }
+
+    pub fn update(&mut self, data: &[u8]) {
+        if data.is_empty() {
+            return;
+        }
+        self.hasher.update(data);
+        self.bytes = self.bytes.saturating_add(data.len() as u64);
+    }
+
+    #[must_use]
+    pub fn snapshot(&self) -> IntegrityState {
+        let digest = {
+            let hasher = self.hasher.clone();
+            let finalized = hasher.finalize();
+            IntegrityDigest(format_sha256(finalized.as_ref()))
+        };
+        IntegrityState {
+            bytes_hashed: self.bytes,
+            digest: Some(digest),
+        }
+    }
+
+    #[must_use]
+    pub fn finalize(self) -> IntegrityState {
+        let digest = IntegrityDigest(format_sha256(self.hasher.finalize().as_ref()));
+        IntegrityState {
+            bytes_hashed: self.bytes,
+            digest: Some(digest),
+        }
+    }
+}
+
+fn format_sha256(bytes: &[u8]) -> String {
+    let mut out = String::with_capacity(7 + bytes.len() * 2);
+    out.push_str("sha256:");
+    for byte in bytes {
+        use std::fmt::Write as _;
+        let _ = write!(out, "{byte:02x}");
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_digest_matches_sha256_of_empty() {
+        let state = Sha256Accountant::new().finalize();
+        assert_eq!(state.bytes_hashed(), 0);
+        assert_eq!(
+            state.digest().map(IntegrityDigest::as_str),
+            Some("sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855")
+        );
+    }
+
+    #[test]
+    fn known_vector_abc() {
+        let mut accountant = Sha256Accountant::new();
+        accountant.update(b"abc");
+        let state = accountant.finalize();
+        assert_eq!(state.bytes_hashed(), 3);
+        assert_eq!(
+            state.digest().map(IntegrityDigest::as_str),
+            Some("sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad")
+        );
+    }
+}

crates/remora-stream/src/lib.rs

@@ -0,0 +1,31 @@
+//! Streaming execution kernel.
+//!
+//! Source and target adapter ports, a bounded-buffer copy loop, cancellation,
+//! progress and checkpoint hooks, integrity accounting, retry timing, and
+//! structured failure mapping. Concrete source and storage adapters implement
+//! the ports; protocol services consume classified outcomes.
+
+mod cancel;
+mod copy;
+mod error;
+mod failure;
+mod integrity;
+mod ports;
+mod progress;
+mod timing;
+
+pub use cancel::CancelToken;
+pub use copy::{
+    CopyConfig, CopyHooks, CopyOutcome, CopyResult, NoopHooks, copy_stream, copy_stream_seeded,
+};
+pub use error::{StreamError, StreamSide};
+pub use failure::{
+    PublicDetails, PublicReason, SafeDetailValue, StructuredFailure, classify_stream_error,
+    public_message_for_class,
+};
+pub use integrity::{IntegrityDigest, IntegrityState, Sha256Accountant};
+pub use ports::{SourceBody, TargetBody, TargetFinish};
+pub use progress::{TransferCheckpoint, TransferProgress};
+pub use timing::{
+    RetryTiming, capped_server_hint, full_jitter_delay, full_jitter_delay_with, saturating_backoff,
+};

crates/remora-stream/src/ports.rs

@@ -0,0 +1,28 @@
+use std::future::Future;
+
+use crate::error::StreamError;
+
+/// Result of finishing a target write for one member.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct TargetFinish {
+    pub bytes_written: u64,
+}
+
+/// Streaming source body. Implementations never buffer the complete object.
+pub trait SourceBody: Send {
+    /// Read the next chunk into `buf`.
+    ///
+    /// Returns `0` at end-of-stream. On success with `n > 0`, `n` must be at
+    /// most `buf.len()` and only `buf[..n]` is defined. Values greater than
+    /// `buf.len()` are a contract violation; the copy loop fails closed.
+    fn read(&mut self, buf: &mut [u8]) -> impl Future<Output = Result<usize, StreamError>> + Send;
+}
+
+/// Streaming target body for create-only attempt object writes.
+pub trait TargetBody: Send {
+    /// Write the complete slice. Partial writes are not exposed.
+    fn write_all(&mut self, data: &[u8]) -> impl Future<Output = Result<(), StreamError>> + Send;
+
+    /// Finish the member write after EOF from the source.
+    fn finish(self) -> impl Future<Output = Result<TargetFinish, StreamError>> + Send;
+}

crates/remora-stream/src/progress.rs

@@ -0,0 +1,96 @@
+use crate::integrity::IntegrityState;
+
+/// Monotonic transfer progress observed by the copy loop.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct TransferProgress {
+    /// Absolute completed bytes, including `initial_completed_bytes`.
+    completed_bytes: u64,
+    /// Bytes written during this copy invocation only.
+    session_bytes: u64,
+    total_bytes: Option<u64>,
+    completed_members: u64,
+    total_members: Option<u64>,
+}
+
+impl TransferProgress {
+    #[must_use]
+    pub fn new(
+        completed_bytes: u64,
+        session_bytes: u64,
+        total_bytes: Option<u64>,
+        completed_members: u64,
+        total_members: Option<u64>,
+    ) -> Self {
+        Self {
+            completed_bytes,
+            session_bytes,
+            total_bytes,
+            completed_members,
+            total_members,
+        }
+    }
+
+    #[must_use]
+    pub fn completed_bytes(&self) -> u64 {
+        self.completed_bytes
+    }
+
+    #[must_use]
+    pub fn session_bytes(&self) -> u64 {
+        self.session_bytes
+    }
+
+    #[must_use]
+    pub fn total_bytes(&self) -> Option<u64> {
+        self.total_bytes
+    }
+
+    #[must_use]
+    pub fn completed_members(&self) -> u64 {
+        self.completed_members
+    }
+
+    #[must_use]
+    pub fn total_members(&self) -> Option<u64> {
+        self.total_members
+    }
+}
+
+/// Checkpoint boundary emitted when retained progress should be durable.
+///
+/// Concrete source and destination proofs are adapter-owned. Integrity covers
+/// every byte hashed by the accountant supplied to the copy (prior seed plus
+/// this session). Absolute digests require a seed whose `bytes_hashed` equals
+/// `initial_completed_bytes`.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct TransferCheckpoint {
+    completed_bytes: u64,
+    session_bytes: u64,
+    integrity: IntegrityState,
+}
+
+impl TransferCheckpoint {
+    #[must_use]
+    pub fn new(completed_bytes: u64, session_bytes: u64, integrity: IntegrityState) -> Self {
+        Self {
+            completed_bytes,
+            session_bytes,
+            integrity,
+        }
+    }
+
+    #[must_use]
+    pub fn completed_bytes(&self) -> u64 {
+        self.completed_bytes
+    }
+
+    #[must_use]
+    pub fn session_bytes(&self) -> u64 {
+        self.session_bytes
+    }
+
+    #[must_use]
+    pub fn integrity(&self) -> &IntegrityState {
+        &self.integrity
+    }
+}

crates/remora-stream/src/timing.rs

@@ -0,0 +1,169 @@
+use std::time::Duration;
+
+/// Adapter-selected timing bounds below the JobProtocol attempt budget.
+///
+/// Connect, response-start, and no-progress timeouts map to structured runtime
+/// failures. There is no elapsed wall-clock job deadline.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct RetryTiming {
+    pub connect_timeout: Duration,
+    pub response_start_timeout: Duration,
+    pub no_progress_timeout: Duration,
+    pub base_backoff: Duration,
+    pub max_backoff: Duration,
+    pub max_server_hint: Duration,
+}
+
+impl RetryTiming {
+    /// Conservative defaults suitable for synthetic and small-object transfers.
+    #[must_use]
+    pub const fn default_bounds() -> Self {
+        Self {
+            connect_timeout: Duration::from_secs(10),
+            response_start_timeout: Duration::from_secs(30),
+            no_progress_timeout: Duration::from_secs(60),
+            base_backoff: Duration::from_millis(100),
+            max_backoff: Duration::from_secs(30),
+            max_server_hint: Duration::from_secs(60),
+        }
+    }
+
+    /// Validate and clamp configuration; zero or inverted bounds become defaults.
+    #[must_use]
+    pub fn sanitized(self) -> Self {
+        let defaults = Self::default_bounds();
+        let base = if self.base_backoff.is_zero() {
+            defaults.base_backoff
+        } else {
+            self.base_backoff
+        };
+        let max_backoff = if self.max_backoff < base {
+            base
+        } else {
+            self.max_backoff
+        };
+        let max_server_hint = if self.max_server_hint.is_zero() {
+            defaults.max_server_hint
+        } else {
+            self.max_server_hint
+        };
+        Self {
+            connect_timeout: nonzero_or(self.connect_timeout, defaults.connect_timeout),
+            response_start_timeout: nonzero_or(
+                self.response_start_timeout,
+                defaults.response_start_timeout,
+            ),
+            no_progress_timeout: nonzero_or(self.no_progress_timeout, defaults.no_progress_timeout),
+            base_backoff: base,
+            max_backoff,
+            max_server_hint,
+        }
+    }
+}
+
+fn nonzero_or(value: Duration, fallback: Duration) -> Duration {
+    if value.is_zero() { fallback } else { value }
+}
+
+/// Exponential backoff ceiling before jitter: `min(max, base * 2^attempt)`.
+#[must_use]
+pub fn saturating_backoff(attempt: u32, timing: &RetryTiming) -> Duration {
+    let timing = timing.sanitized();
+    let shift = attempt.min(16);
+    let factor = 1u64.checked_shl(shift).unwrap_or(u64::MAX);
+    let base_ms = u64::try_from(timing.base_backoff.as_millis()).unwrap_or(u64::MAX);
+    let scaled = base_ms.saturating_mul(factor);
+    let max_ms = u64::try_from(timing.max_backoff.as_millis()).unwrap_or(u64::MAX);
+    Duration::from_millis(scaled.min(max_ms))
+}
+
+/// Full-jitter delay in `[0, saturating_backoff]`.
+///
+/// If the OS RNG is unavailable, returns the full backoff ceiling so retries
+/// do not collapse to a synchronized zero delay.
+#[must_use]
+pub fn full_jitter_delay(attempt: u32, timing: &RetryTiming) -> Duration {
+    let ceiling = saturating_backoff(attempt, timing);
+    match random_u64() {
+        Some(random) => full_jitter_delay_with(ceiling, random),
+        None => ceiling,
+    }
+}
+
+/// Full-jitter with an injected random value for deterministic tests.
+#[must_use]
+pub fn full_jitter_delay_with(ceiling: Duration, random: u64) -> Duration {
+    if ceiling.is_zero() {
+        return Duration::ZERO;
+    }
+    let ceiling_ms = u64::try_from(ceiling.as_millis()).unwrap_or(u64::MAX);
+    if ceiling_ms == 0 {
+        return Duration::ZERO;
+    }
+    let chosen = random % (ceiling_ms.saturating_add(1));
+    Duration::from_millis(chosen)
+}
+
+/// Cap or reject a server-supplied retry hint.
+///
+/// Zero, missing, or excessive hints yield `None` so adapters ignore them.
+#[must_use]
+pub fn capped_server_hint(hint: Option<Duration>, timing: &RetryTiming) -> Option<Duration> {
+    let timing = timing.sanitized();
+    let hint = hint?;
+    if hint.is_zero() || hint > timing.max_server_hint {
+        return None;
+    }
+    Some(hint)
+}
+
+fn random_u64() -> Option<u64> {
+    let mut bytes = [0u8; 8];
+    getrandom::fill(&mut bytes).ok()?;
+    Some(u64::from_le_bytes(bytes))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn backoff_grows_then_caps() {
+        let timing = RetryTiming {
+            base_backoff: Duration::from_millis(100),
+            max_backoff: Duration::from_secs(1),
+            ..RetryTiming::default_bounds()
+        };
+        assert_eq!(saturating_backoff(0, &timing), Duration::from_millis(100));
+        assert_eq!(saturating_backoff(1, &timing), Duration::from_millis(200));
+        assert_eq!(saturating_backoff(2, &timing), Duration::from_millis(400));
+        assert_eq!(saturating_backoff(10, &timing), Duration::from_secs(1));
+    }
+
+    #[test]
+    fn full_jitter_stays_within_ceiling() {
+        let ceiling = Duration::from_millis(1000);
+        for random in [0, 1, 500, 999, 1000, u64::MAX] {
+            let delay = full_jitter_delay_with(ceiling, random);
+            assert!(delay <= ceiling);
+        }
+    }
+
+    #[test]
+    fn server_hints_are_capped_or_ignored() {
+        let timing = RetryTiming {
+            max_server_hint: Duration::from_secs(5),
+            ..RetryTiming::default_bounds()
+        };
+        assert_eq!(
+            capped_server_hint(Some(Duration::from_secs(3)), &timing),
+            Some(Duration::from_secs(3))
+        );
+        assert_eq!(
+            capped_server_hint(Some(Duration::from_secs(6)), &timing),
+            None
+        );
+        assert_eq!(capped_server_hint(Some(Duration::ZERO), &timing), None);
+        assert_eq!(capped_server_hint(None, &timing), None);
+    }
+}

crates/remora-stream/tests/kernel.rs

@@ -0,0 +1,939 @@
+//! Streaming execution kernel acceptance matrix.
+//!
+//! Synthetic sources and targets exercise the bounded copy loop without any
+//! backend fixture. Memory bound evidence is the single reserved payload buffer.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::time::Duration;
+
+use remora_domain::{FailureClass, FailureDisposition, RuntimeFailureClass};
+use remora_stream::{
+    CancelToken, CopyConfig, CopyHooks, CopyOutcome, PublicDetails, PublicReason, RetryTiming,
+    SafeDetailValue, Sha256Accountant, SourceBody, StreamError, StreamSide, StructuredFailure,
+    TargetBody, TargetFinish, TransferCheckpoint, TransferProgress, capped_server_hint,
+    classify_stream_error, copy_stream, copy_stream_seeded, full_jitter_delay_with,
+    public_message_for_class, saturating_backoff,
+};
+use tokio::sync::Notify;
+
+const KILOBYTE: u64 = 1024;
+const MEGABYTE: u64 = 1024 * KILOBYTE;
+const GIGABYTE: u64 = 1024 * MEGABYTE;
+
+/// Patterned source that never materializes the full payload.
+struct PatternSource {
+    remaining: u64,
+    offset: u64,
+    fail_after: Option<u64>,
+    stall: Option<Duration>,
+    stall_after: Option<u64>,
+    max_chunk: usize,
+    peak_fill: Arc<AtomicUsize>,
+    oversize_read: bool,
+}
+
+impl PatternSource {
+    fn new(total: u64) -> Self {
+        Self {
+            remaining: total,
+            offset: 0,
+            fail_after: None,
+            stall: None,
+            stall_after: None,
+            max_chunk: usize::MAX,
+            peak_fill: Arc::new(AtomicUsize::new(0)),
+            oversize_read: false,
+        }
+    }
+
+    fn starting_at(total: u64, offset: u64) -> Self {
+        let mut source = Self::new(total.saturating_sub(offset));
+        source.offset = offset;
+        source
+    }
+
+    fn fail_after(mut self, bytes: u64) -> Self {
+        self.fail_after = Some(bytes);
+        self
+    }
+
+    fn stall_after(mut self, bytes: u64, stall: Duration) -> Self {
+        self.stall_after = Some(bytes);
+        self.stall = Some(stall);
+        self
+    }
+
+    fn max_chunk(mut self, max_chunk: usize) -> Self {
+        self.max_chunk = max_chunk;
+        self
+    }
+
+    fn oversize_read(mut self) -> Self {
+        self.oversize_read = true;
+        self
+    }
+}
+
+impl SourceBody for PatternSource {
+    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, StreamError> {
+        if self.oversize_read {
+            return Ok(buf.len().saturating_add(1));
+        }
+        if let Some(limit) = self.fail_after
+            && self.offset >= limit
+        {
+            return Err(StreamError::transient_source(
+                "synthetic mid-stream interruption with token=super-secret-token",
+            ));
+        }
+        if let (Some(after), Some(stall)) = (self.stall_after, self.stall)
+            && self.offset >= after
+        {
+            tokio::time::sleep(stall).await;
+        }
+        if self.remaining == 0 {
+            return Ok(0);
+        }
+        let want = (buf.len().min(self.max_chunk) as u64).min(self.remaining) as usize;
+        for (index, slot) in buf[..want].iter_mut().enumerate() {
+            let absolute = self.offset + index as u64;
+            *slot = pattern_byte(absolute);
+        }
+        self.peak_fill.fetch_max(want, Ordering::SeqCst);
+        self.offset += want as u64;
+        self.remaining -= want as u64;
+        Ok(want)
+    }
+}
+
+fn pattern_byte(offset: u64) -> u8 {
+    (offset
+        .wrapping_mul(0x9E37_79B9_7F4A_7C15)
+        .wrapping_add(0xA24B_AED4_96E9_7C8D)
+        >> 56) as u8
+}
+
+fn fill_pattern(offset: u64, len: usize) -> Vec<u8> {
+    (0..len)
+        .map(|index| pattern_byte(offset + index as u64))
+        .collect()
+}
+
+struct CountingTarget {
+    written: u64,
+    absolute_base: u64,
+    expected: Option<u64>,
+    delay_per_write: Option<Duration>,
+    stall_first_write: Option<Duration>,
+    fail_after: Option<u64>,
+    bytes: Option<Vec<u8>>,
+    peak_write: Arc<AtomicUsize>,
+}
+
+impl CountingTarget {
+    fn discarding() -> Self {
+        Self {
+            written: 0,
+            absolute_base: 0,
+            expected: None,
+            delay_per_write: None,
+            stall_first_write: None,
+            fail_after: None,
+            bytes: None,
+            peak_write: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    fn collecting() -> Self {
+        Self {
+            written: 0,
+            absolute_base: 0,
+            expected: None,
+            delay_per_write: None,
+            stall_first_write: None,
+            fail_after: None,
+            bytes: Some(Vec::new()),
+            peak_write: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    fn with_delay(mut self, delay: Duration) -> Self {
+        self.delay_per_write = Some(delay);
+        self
+    }
+
+    fn stall_first_write(mut self, stall: Duration) -> Self {
+        self.stall_first_write = Some(stall);
+        self
+    }
+
+    fn fail_after(mut self, bytes: u64) -> Self {
+        self.fail_after = Some(bytes);
+        self
+    }
+
+    fn expect_total(mut self, total: u64) -> Self {
+        self.expected = Some(total);
+        self
+    }
+
+    fn absolute_base(mut self, base: u64) -> Self {
+        self.absolute_base = base;
+        self
+    }
+}
+
+impl TargetBody for CountingTarget {
+    async fn write_all(&mut self, data: &[u8]) -> Result<(), StreamError> {
+        if let Some(limit) = self.fail_after
+            && self.written >= limit
+        {
+            return Err(StreamError::transient_target(
+                "target write failed; Authorization: Bearer leaked-token",
+            ));
+        }
+        if let Some(stall) = self.stall_first_write.take() {
+            tokio::time::sleep(stall).await;
+        }
+        if let Some(delay) = self.delay_per_write {
+            tokio::time::sleep(delay).await;
+        }
+        self.peak_write.fetch_max(data.len(), Ordering::SeqCst);
+        if let Some(bytes) = &mut self.bytes {
+            bytes.extend_from_slice(data);
+        } else if let Some(expected) = self.expected {
+            if !data.is_empty() {
+                let absolute = self.absolute_base + self.written;
+                assert_eq!(data[0], pattern_byte(absolute));
+                let last = data.len() - 1;
+                assert_eq!(data[last], pattern_byte(absolute + last as u64));
+            }
+            let _ = expected;
+        }
+        self.written += data.len() as u64;
+        Ok(())
+    }
+
+    async fn finish(self) -> Result<TargetFinish, StreamError> {
+        if let Some(expected) = self.expected {
+            assert_eq!(self.written, expected);
+        }
+        Ok(TargetFinish {
+            bytes_written: self.written,
+        })
+    }
+}
+
+#[derive(Default)]
+struct RecordingHooks {
+    progress_events: Vec<u64>,
+    checkpoint_events: Vec<u64>,
+    progress_count: Arc<AtomicU64>,
+    checkpoint_count: Arc<AtomicU64>,
+}
+
+impl CopyHooks for RecordingHooks {
+    async fn on_progress(&mut self, progress: &TransferProgress) -> Result<(), StreamError> {
+        self.progress_events.push(progress.completed_bytes());
+        self.progress_count.fetch_add(1, Ordering::SeqCst);
+        Ok(())
+    }
+
+    async fn on_checkpoint(&mut self, checkpoint: &TransferCheckpoint) -> Result<(), StreamError> {
+        self.checkpoint_events.push(checkpoint.completed_bytes());
+        self.checkpoint_count.fetch_add(1, Ordering::SeqCst);
+        // Integrity covers prior seed + session; equals absolute completed bytes.
+        assert_eq!(
+            checkpoint.integrity().bytes_hashed(),
+            checkpoint.completed_bytes()
+        );
+        assert!(checkpoint.integrity().digest().is_some());
+        Ok(())
+    }
+}
+
+fn config_for(total: u64, buffer_size: usize) -> CopyConfig {
+    CopyConfig {
+        buffer_size,
+        progress_interval_bytes: buffer_size as u64,
+        checkpoint_interval_bytes: (buffer_size as u64).saturating_mul(4),
+        no_progress_timeout: Duration::from_secs(5),
+        initial_completed_bytes: 0,
+        total_bytes: Some(total),
+        completed_members: 0,
+        total_members: Some(1),
+    }
+}
+
+#[tokio::test]
+async fn small_transfer_completes_with_known_digest() {
+    let payload = b"remora-stream-kernel";
+    let source = SliceSource {
+        data: payload.to_vec(),
+        position: 0,
+    };
+    let target = CountingTarget::collecting();
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let outcome = copy_stream(
+        source,
+        target,
+        config_for(payload.len() as u64, 64),
+        &cancel,
+        &mut hooks,
+    )
+    .await;
+
+    let CopyOutcome::Completed(result) = outcome else {
+        panic!("expected completion: {outcome:?}");
+    };
+    assert_eq!(result.progress.completed_bytes(), payload.len() as u64);
+    assert_eq!(result.progress.session_bytes(), payload.len() as u64);
+    assert_eq!(result.target.bytes_written, payload.len() as u64);
+    // Requested 64 is clamped up to MIN_BUFFER_SIZE.
+    assert_eq!(result.buffer_len, CopyConfig::MIN_BUFFER_SIZE);
+    assert!(result.buffer_capacity >= result.buffer_len);
+    assert!(result.buffer_capacity < result.buffer_len.saturating_add(4096));
+    assert_eq!(
+        result.integrity.digest().map(|d| d.as_str()),
+        Some(sha256_hex(payload).as_str())
+    );
+    assert!(!hooks.progress_events.is_empty());
+}
+
+struct SliceSource {
+    data: Vec<u8>,
+    position: usize,
+}
+
+impl SourceBody for SliceSource {
+    async fn read(&mut self, buf: &mut [u8]) -> Result<usize, StreamError> {
+        if self.position >= self.data.len() {
+            return Ok(0);
+        }
+        let n = buf.len().min(self.data.len() - self.position);
+        buf[..n].copy_from_slice(&self.data[self.position..self.position + n]);
+        self.position += n;
+        Ok(n)
+    }
+}
+
+fn sha256_hex(data: &[u8]) -> String {
+    use sha2::{Digest, Sha256};
+    let digest = Sha256::digest(data);
+    let mut out = String::from("sha256:");
+    for byte in digest {
+        use std::fmt::Write as _;
+        let _ = write!(out, "{byte:02x}");
+    }
+    out
+}
+
+#[tokio::test]
+async fn multi_gigabyte_stream_keeps_bounded_memory() {
+    // Two gigabytes of synthetic patterned data. The kernel holds one buffer.
+    let total = 2 * GIGABYTE;
+    let buffer_size = 256 * 1024;
+    let source = PatternSource::new(total).max_chunk(buffer_size);
+    let peak_fill = Arc::clone(&source.peak_fill);
+    let target = CountingTarget::discarding().expect_total(total);
+    let peak_write = Arc::clone(&target.peak_write);
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+
+    let outcome = copy_stream(
+        source,
+        target,
+        config_for(total, buffer_size),
+        &cancel,
+        &mut hooks,
+    )
+    .await;
+
+    let CopyOutcome::Completed(result) = outcome else {
+        panic!("expected multi-GB completion: {outcome:?}");
+    };
+    assert_eq!(result.progress.completed_bytes(), total);
+    assert_eq!(result.target.bytes_written, total);
+    assert_eq!(result.integrity.bytes_hashed(), total);
+    assert!(result.integrity.digest().is_some());
+
+    // Bounded memory evidence: one configured buffer, no growth beyond page slack.
+    assert_eq!(result.buffer_len, buffer_size);
+    assert!(result.buffer_capacity >= buffer_size);
+    assert!(result.buffer_capacity < buffer_size.saturating_add(4096));
+    assert!(peak_fill.load(Ordering::SeqCst) <= buffer_size);
+    assert!(peak_write.load(Ordering::SeqCst) <= buffer_size);
+    assert!(hooks.progress_count.load(Ordering::SeqCst) > 0);
+    assert!(hooks.checkpoint_count.load(Ordering::SeqCst) > 0);
+}
+
+#[tokio::test]
+async fn mid_stream_interruption_classifies_and_preserves_progress() {
+    let total = 8 * MEGABYTE;
+    let fail_after = 3 * MEGABYTE;
+    let buffer_size = 64 * 1024;
+    let source = PatternSource::new(total).fail_after(fail_after);
+    let target = CountingTarget::discarding();
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+
+    let outcome = copy_stream(
+        source,
+        target,
+        config_for(total, buffer_size),
+        &cancel,
+        &mut hooks,
+    )
+    .await;
+
+    let CopyOutcome::Failed {
+        failure,
+        progress,
+        integrity,
+        buffer_len,
+        buffer_capacity,
+    } = outcome
+    else {
+        panic!("expected failure: {outcome:?}");
+    };
+
+    assert_eq!(failure.class(), FailureClass::TransientSource);
+    assert_eq!(failure.disposition(), Some(FailureDisposition::Retryable));
+    assert_eq!(
+        failure.runtime_class(),
+        Some(RuntimeFailureClass::TransientSource)
+    );
+    // Public projection must not carry the synthetic secret token.
+    assert!(!failure.message().contains("super-secret-token"));
+    assert!(!format!("{failure:?}").contains("super-secret-token"));
+    assert!(progress.completed_bytes() >= fail_after.saturating_sub(buffer_size as u64));
+    assert!(progress.completed_bytes() < total);
+    assert_eq!(integrity.bytes_hashed(), progress.completed_bytes());
+    assert_eq!(buffer_len, buffer_size);
+    assert!(buffer_capacity >= buffer_len);
+}
+
+#[tokio::test]
+async fn backpressure_still_completes_with_bounded_buffer() {
+    let total = 512 * KILOBYTE;
+    let buffer_size = 16 * 1024;
+    let source = PatternSource::new(total).max_chunk(buffer_size);
+    let target = CountingTarget::discarding()
+        .expect_total(total)
+        .with_delay(Duration::from_millis(1));
+    let peak_write = Arc::clone(&target.peak_write);
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+
+    let outcome = copy_stream(
+        source,
+        target,
+        config_for(total, buffer_size),
+        &cancel,
+        &mut hooks,
+    )
+    .await;
+
+    let CopyOutcome::Completed(result) = outcome else {
+        panic!("expected completion under backpressure: {outcome:?}");
+    };
+    assert_eq!(result.progress.completed_bytes(), total);
+    assert!(peak_write.load(Ordering::SeqCst) <= buffer_size);
+}
+
+#[tokio::test]
+async fn no_progress_timeout_maps_to_transient_source() {
+    let total = MEGABYTE;
+    let buffer_size = 64 * 1024;
+    let source = PatternSource::new(total).stall_after(0, Duration::from_millis(200));
+    let target = CountingTarget::discarding();
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let mut config = config_for(total, buffer_size);
+    config.no_progress_timeout = Duration::from_millis(50);
+
+    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;
+
+    let CopyOutcome::Failed {
+        failure, progress, ..
+    } = outcome
+    else {
+        panic!("expected no-progress failure: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::TransientSource);
+    assert_eq!(
+        failure.safe_details().get("reason"),
+        Some(&SafeDetailValue::String("no_progress".into()))
+    );
+    assert_eq!(progress.completed_bytes(), 0);
+}
+
+#[tokio::test]
+async fn no_progress_timeout_maps_to_transient_target() {
+    let total = MEGABYTE;
+    let buffer_size = 64 * 1024;
+    let source = PatternSource::new(total).max_chunk(buffer_size);
+    let target = CountingTarget::discarding().stall_first_write(Duration::from_millis(200));
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let mut config = config_for(total, buffer_size);
+    config.no_progress_timeout = Duration::from_millis(50);
+
+    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;
+
+    let CopyOutcome::Failed {
+        failure, progress, ..
+    } = outcome
+    else {
+        panic!("expected target no-progress failure: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::TransientTarget);
+    assert_eq!(
+        failure.safe_details().get("reason"),
+        Some(&SafeDetailValue::String("no_progress".into()))
+    );
+    assert_eq!(
+        failure.safe_details().get("side"),
+        Some(&SafeDetailValue::String("target".into()))
+    );
+    assert_eq!(progress.completed_bytes(), 0);
+}
+
+#[tokio::test]
+async fn cooperative_cancellation_stops_transfer() {
+    let total = 64 * MEGABYTE;
+    let buffer_size = 64 * 1024;
+    let source = PatternSource::new(total).max_chunk(4 * 1024);
+    let target = CountingTarget::discarding();
+    let cancel = CancelToken::new();
+    let cancel_for_task = cancel.clone();
+    let notify = Arc::new(Notify::new());
+    let notify_wait = Arc::clone(&notify);
+    let checkpoints_after_cancel = Arc::new(AtomicU64::new(0));
+    let checkpoints_for_hooks = Arc::clone(&checkpoints_after_cancel);
+
+    struct CancelAfterProgress {
+        cancel: CancelToken,
+        notify: Arc<Notify>,
+        armed: bool,
+        checkpoints_after_cancel: Arc<AtomicU64>,
+    }
+
+    impl CopyHooks for CancelAfterProgress {
+        async fn on_progress(&mut self, progress: &TransferProgress) -> Result<(), StreamError> {
+            if !self.armed && progress.completed_bytes() > 0 {
+                self.armed = true;
+                self.cancel.cancel();
+                self.notify.notify_one();
+            }
+            Ok(())
+        }
+
+        async fn on_checkpoint(
+            &mut self,
+            _checkpoint: &TransferCheckpoint,
+        ) -> Result<(), StreamError> {
+            if self.armed {
+                self.checkpoints_after_cancel.fetch_add(1, Ordering::SeqCst);
+            }
+            Ok(())
+        }
+    }
+
+    let hooks = CancelAfterProgress {
+        cancel: cancel_for_task,
+        notify,
+        armed: false,
+        checkpoints_after_cancel: checkpoints_for_hooks,
+    };
+
+    let copy = tokio::spawn(async move {
+        copy_stream(
+            source,
+            target,
+            config_for(total, buffer_size),
+            &cancel,
+            hooks,
+        )
+        .await
+    });
+
+    notify_wait.notified().await;
+    let outcome = copy.await.expect("join");
+
+    match outcome {
+        CopyOutcome::Cancelled {
+            progress,
+            integrity,
+            ..
+        } => {
+            assert!(progress.completed_bytes() > 0);
+            assert!(progress.completed_bytes() < total);
+            assert_eq!(integrity.bytes_hashed(), progress.completed_bytes());
+            assert_eq!(checkpoints_after_cancel.load(Ordering::SeqCst), 0);
+        }
+        other => panic!("expected cancellation: {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn oversize_source_read_fails_closed() {
+    let source = PatternSource::new(MEGABYTE).oversize_read();
+    let target = CountingTarget::discarding();
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+
+    let outcome = copy_stream(
+        source,
+        target,
+        config_for(MEGABYTE, 64 * 1024),
+        &cancel,
+        &mut hooks,
+    )
+    .await;
+
+    let CopyOutcome::Failed {
+        failure, progress, ..
+    } = outcome
+    else {
+        panic!("expected oversize-read failure: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::Internal);
+    assert_eq!(progress.completed_bytes(), 0);
+}
+
+#[tokio::test]
+async fn resume_with_seeded_integrity_matches_full_digest() {
+    let total = 256 * 1024u64;
+    let half = total / 2;
+    let buffer_size = 16 * 1024;
+
+    let mut accountant = Sha256Accountant::new();
+    let prefix = fill_pattern(0, half as usize);
+    accountant.update(&prefix);
+    assert_eq!(accountant.bytes_hashed(), half);
+
+    let source = PatternSource::starting_at(total, half).max_chunk(buffer_size);
+    let target = CountingTarget::discarding()
+        .expect_total(half)
+        .absolute_base(half);
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let mut config = config_for(total, buffer_size);
+    config.initial_completed_bytes = half;
+    config.progress_interval_bytes = 32 * 1024;
+    config.checkpoint_interval_bytes = 64 * 1024;
+
+    let outcome = copy_stream_seeded(source, target, config, &cancel, &mut hooks, accountant).await;
+
+    let CopyOutcome::Completed(result) = outcome else {
+        panic!("expected resumed completion: {outcome:?}");
+    };
+    assert_eq!(result.progress.completed_bytes(), total);
+    assert_eq!(result.progress.session_bytes(), half);
+    assert_eq!(result.target.bytes_written, half);
+    assert_eq!(result.integrity.bytes_hashed(), total);
+
+    let full = fill_pattern(0, total as usize);
+    assert_eq!(
+        result.integrity.digest().map(|d| d.as_str()),
+        Some(sha256_hex(&full).as_str())
+    );
+}
+
+#[tokio::test]
+async fn mismatched_integrity_seed_fails_closed() {
+    let mut accountant = Sha256Accountant::new();
+    accountant.update(b"prefix");
+    let source = PatternSource::new(KILOBYTE);
+    let target = CountingTarget::discarding();
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let mut config = config_for(KILOBYTE, 4 * 1024);
+    config.initial_completed_bytes = 100; // does not match accountant
+
+    let outcome = copy_stream_seeded(source, target, config, &cancel, &mut hooks, accountant).await;
+
+    let CopyOutcome::Failed { failure, .. } = outcome else {
+        panic!("expected seed mismatch failure: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::Internal);
+}
+
+#[tokio::test]
+async fn non_zero_initial_without_seed_fails_closed() {
+    let source = PatternSource::new(KILOBYTE);
+    let target = CountingTarget::discarding();
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let mut config = config_for(KILOBYTE, 4 * 1024);
+    config.initial_completed_bytes = 64;
+
+    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;
+    let CopyOutcome::Failed { failure, .. } = outcome else {
+        panic!("expected unseeded resume failure: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::Internal);
+}
+
+#[tokio::test]
+async fn target_failure_redacts_authorization_header() {
+    let total = MEGABYTE;
+    let buffer_size = 32 * 1024;
+    let source = PatternSource::new(total);
+    let target = CountingTarget::discarding().fail_after(128 * 1024);
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+
+    let outcome = copy_stream(
+        source,
+        target,
+        config_for(total, buffer_size),
+        &cancel,
+        &mut hooks,
+    )
+    .await;
+
+    let CopyOutcome::Failed { failure, .. } = outcome else {
+        panic!("expected target failure: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::TransientTarget);
+    assert!(!failure.message().contains("leaked-token"));
+    assert!(!failure.message().contains("Bearer"));
+    assert_eq!(
+        failure.message(),
+        public_message_for_class(FailureClass::TransientTarget)
+    );
+}
+
+#[test]
+fn structured_failure_mapping_covers_runtime_classes() {
+    let cases = [
+        (
+            StreamError::Authorization {
+                side: StreamSide::Source,
+            },
+            FailureClass::Authorization,
+            FailureDisposition::Permanent,
+        ),
+        (
+            StreamError::SourceNotFound,
+            FailureClass::SourceNotFound,
+            FailureDisposition::Permanent,
+        ),
+        (
+            StreamError::SourceChanged,
+            FailureClass::SourceChanged,
+            FailureDisposition::Permanent,
+        ),
+        (
+            StreamError::transient_source("x"),
+            FailureClass::TransientSource,
+            FailureDisposition::Retryable,
+        ),
+        (
+            StreamError::transient_target("x"),
+            FailureClass::TransientTarget,
+            FailureDisposition::Retryable,
+        ),
+        (
+            StreamError::integrity("x"),
+            FailureClass::Integrity,
+            FailureDisposition::Retryable,
+        ),
+        (
+            StreamError::Capacity {
+                side: StreamSide::Target,
+            },
+            FailureClass::Capacity,
+            FailureDisposition::Retryable,
+        ),
+        (
+            StreamError::internal("stack trace at https://evil.example/path?token=1"),
+            FailureClass::Internal,
+            FailureDisposition::Retryable,
+        ),
+        (
+            StreamError::Cancelled,
+            FailureClass::Cancelled,
+            FailureDisposition::Permanent,
+        ),
+    ];
+
+    for (error, class, disposition) in cases {
+        let failure = classify_stream_error(&error);
+        assert_eq!(failure.class(), class);
+        assert_eq!(failure.disposition(), Some(disposition));
+        assert!(!failure.message().contains("https://"));
+        assert!(!failure.message().contains("token="));
+        assert!(!failure.message().contains("stack trace"));
+    }
+}
+
+#[test]
+fn public_failure_rejects_free_form_secrets() {
+    let secret = "synthetic-api-key-4f7c9a";
+    // Typed constructor cannot accept free-form message or reason text.
+    let failure = StructuredFailure::public(
+        FailureClass::TransientSource,
+        PublicDetails::new()
+            .with_side(StreamSide::Source)
+            .with_reason(PublicReason::NoProgress),
+    );
+    assert_eq!(
+        failure.message(),
+        public_message_for_class(FailureClass::TransientSource)
+    );
+    assert!(!failure.message().contains(secret));
+    assert_eq!(
+        failure.safe_details().get("reason"),
+        Some(&SafeDetailValue::String("no_progress".into()))
+    );
+    assert!(!format!("{failure:?}").contains(secret));
+}
+
+#[tokio::test]
+async fn target_finish_honors_no_progress_timeout() {
+    use std::future::pending;
+
+    struct EmptySource;
+    impl SourceBody for EmptySource {
+        async fn read(&mut self, _buf: &mut [u8]) -> Result<usize, StreamError> {
+            Ok(0)
+        }
+    }
+
+    struct StalledFinish;
+    impl TargetBody for StalledFinish {
+        async fn write_all(&mut self, _data: &[u8]) -> Result<(), StreamError> {
+            Ok(())
+        }
+        async fn finish(self) -> Result<TargetFinish, StreamError> {
+            pending().await
+        }
+    }
+
+    let cancel = CancelToken::new();
+    let config = CopyConfig {
+        no_progress_timeout: Duration::from_millis(20),
+        ..CopyConfig::default()
+    };
+    let outcome = tokio::time::timeout(
+        Duration::from_millis(200),
+        copy_stream(
+            EmptySource,
+            StalledFinish,
+            config,
+            &cancel,
+            RecordingHooks::default(),
+        ),
+    )
+    .await
+    .expect("finish must not hang past no-progress bound");
+
+    let CopyOutcome::Failed { failure, .. } = outcome else {
+        panic!("expected target no-progress on stalled finish: {outcome:?}");
+    };
+    assert_eq!(failure.class(), FailureClass::TransientTarget);
+    assert_eq!(
+        failure.safe_details().get("reason"),
+        Some(&SafeDetailValue::String("no_progress".into()))
+    );
+}
+
+#[tokio::test]
+async fn target_finish_honors_cancellation() {
+    use std::future::pending;
+
+    struct EmptySource;
+    impl SourceBody for EmptySource {
+        async fn read(&mut self, _buf: &mut [u8]) -> Result<usize, StreamError> {
+            Ok(0)
+        }
+    }
+
+    struct StalledFinish;
+    impl TargetBody for StalledFinish {
+        async fn write_all(&mut self, _data: &[u8]) -> Result<(), StreamError> {
+            Ok(())
+        }
+        async fn finish(self) -> Result<TargetFinish, StreamError> {
+            pending().await
+        }
+    }
+
+    let cancel = CancelToken::new();
+    let cancel_for_task = cancel.clone();
+    tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        cancel_for_task.cancel();
+    });
+
+    // Long no-progress so cancellation, not timeout, must win.
+    let config = CopyConfig {
+        no_progress_timeout: Duration::from_secs(30),
+        ..CopyConfig::default()
+    };
+    let outcome = tokio::time::timeout(
+        Duration::from_millis(200),
+        copy_stream(
+            EmptySource,
+            StalledFinish,
+            config,
+            &cancel,
+            RecordingHooks::default(),
+        ),
+    )
+    .await
+    .expect("finish must return promptly on cancel");
+
+    assert!(
+        matches!(outcome, CopyOutcome::Cancelled { .. }),
+        "expected cancellation during finish: {outcome:?}"
+    );
+}
+
+#[test]
+fn retry_timing_helpers_honor_bounds() {
+    let timing = RetryTiming {
+        base_backoff: Duration::from_millis(50),
+        max_backoff: Duration::from_millis(400),
+        max_server_hint: Duration::from_secs(2),
+        ..RetryTiming::default_bounds()
+    };
+    assert!(saturating_backoff(0, &timing) <= timing.max_backoff);
+    assert!(saturating_backoff(20, &timing) <= timing.max_backoff);
+    let delay = full_jitter_delay_with(Duration::from_millis(100), 42);
+    assert!(delay <= Duration::from_millis(100));
+    assert_eq!(
+        capped_server_hint(Some(Duration::from_secs(1)), &timing),
+        Some(Duration::from_secs(1))
+    );
+    assert_eq!(
+        capped_server_hint(Some(Duration::from_secs(10)), &timing),
+        None
+    );
+}
+
+#[tokio::test]
+async fn progress_and_checkpoint_hooks_fire_on_intervals() {
+    let total = 256 * 1024;
+    let buffer_size = 16 * 1024;
+    let source = PatternSource::new(total);
+    let target = CountingTarget::discarding().expect_total(total);
+    let cancel = CancelToken::new();
+    let mut hooks = RecordingHooks::default();
+    let mut config = config_for(total, buffer_size);
+    config.progress_interval_bytes = 32 * 1024;
+    config.checkpoint_interval_bytes = 64 * 1024;
+
+    let outcome = copy_stream(source, target, config, &cancel, &mut hooks).await;
+    assert!(matches!(outcome, CopyOutcome::Completed(_)));
+    assert!(hooks.progress_events.len() >= 4);
+    assert!(hooks.checkpoint_events.len() >= 2);
+    assert!(hooks.progress_events.is_sorted());
+    assert!(hooks.checkpoint_events.is_sorted());
+    assert_eq!(*hooks.progress_events.last().unwrap(), total);
+}