diff --git a/crates/spatialrust-io/src/las/writer.rs b/crates/spatialrust-io/src/las/writer.rs
index 5d10774..728a4b5 100644
--- a/crates/spatialrust-io/src/las/writer.rs
+++ b/crates/spatialrust-io/src/las/writer.rs
@@ -117,7 +117,7 @@ pub struct LasChunkSink<W: Write + Seek + Send + Sync + 'static> {
     schema: SchemaDescriptor,
     export_schema: PointSchema,
     point_format: Format,
-    expected_points: u64,
+    expected_points: Option<u64>,
     written_points: u64,
     finished: bool,
 }
@@ -145,7 +145,35 @@ impl<W: Write + Seek + Send + Sync + 'static> LasChunkSink<W> {
             schema,
             export_schema,
             point_format,
-            expected_points,
+            expected_points: Some(expected_points),
+            written_points: 0,
+            finished: false,
+        })
+    }
+
+    /// Creates a LAS/LAZ sink whose final point count is not known up front.
+    ///
+    /// The seekable LAS writer patches the header when [`Self::finish`] closes it.
+    pub fn new_open_ended(
+        writer: W,
+        schema: SchemaDescriptor,
+        format: LasWriteFormat,
+    ) -> Result<Self, IoError> {
+        if format == LasWriteFormat::Laz {
+            #[cfg(not(feature = "io-laz"))]
+            return Err(crate::error::laz_format(
+                "LAZ output requires the io-laz feature".to_owned(),
+            ));
+        }
+        let (point_format, export_schema) = schema_from_point_cloud(schema.point_schema())?;
+        let header = header_from_cloud(point_format, format)?;
+        let writer = Writer::new(writer, header).map_err(|error| las_format(error.to_string()))?;
+        Ok(Self {
+            writer,
+            schema,
+            export_schema,
+            point_format,
+            expected_points: None,
             written_points: 0,
             finished: false,
         })
@@ -168,6 +196,15 @@ impl LasChunkSink<std::io::BufWriter<std::fs::File>> {
             format,
         )
     }
+
+    /// Creates a local LAS/LAZ sink whose final point count is not known up front.
+    pub fn create_open_ended(
+        path: impl AsRef<std::path::Path>,
+        schema: SchemaDescriptor,
+        format: LasWriteFormat,
+    ) -> Result<Self, IoError> {
+        Self::new_open_ended(std::io::BufWriter::new(std::fs::File::create(path)?), schema, format)
+    }
 }
 
 #[cfg(feature = "streaming")]
@@ -189,11 +226,12 @@ impl<W: Write + Seek + Send + Sync + 'static> BoundedSpatialRecordSink for LasCh
                     .map_err(|_| RecordsError::ReceiptOverflow("LAS chunk point count".into()))?,
             )
             .ok_or_else(|| RecordsError::ReceiptOverflow("LAS point count".into()))?;
-        if next > self.expected_points {
-            return Err(RecordsError::InvalidChunk(format!(
-                "LAS sink expected {} points but received at least {next}",
-                self.expected_points
-            )));
+        if let Some(expected_points) = self.expected_points {
+            if next > expected_points {
+                return Err(RecordsError::InvalidChunk(format!(
+                    "LAS sink expected {expected_points} points but received at least {next}"
+                )));
+            }
         }
         for index in 0..cloud.len() {
             let point = point_from_cloud(cloud, &self.export_schema, index, self.point_format)
@@ -207,11 +245,13 @@ impl<W: Write + Seek + Send + Sync + 'static> BoundedSpatialRecordSink for LasCh
     }
 
     fn finish(&mut self) -> RecordsResult<()> {
-        if self.written_points != self.expected_points {
-            return Err(RecordsError::InvalidChunk(format!(
-                "LAS sink expected {} points but received {}",
-                self.expected_points, self.written_points
-            )));
+        if let Some(expected_points) = self.expected_points {
+            if self.written_points != expected_points {
+                return Err(RecordsError::InvalidChunk(format!(
+                    "LAS sink expected {expected_points} points but received {}",
+                    self.written_points
+                )));
+            }
         }
         self.writer.close().map_err(|error| records_io(las_format(error.to_string())))?;
         self.finished = true;
diff --git a/crates/spatialrust-pipeline/Cargo.toml b/crates/spatialrust-pipeline/Cargo.toml
index 124151f..bf70e6e 100644
--- a/crates/spatialrust-pipeline/Cargo.toml
+++ b/crates/spatialrust-pipeline/Cargo.toml
@@ -20,6 +20,7 @@ pipeline-mvp-gpu = [
 pipeline-streaming = [
     "dep:spatialrust-records",
     "dep:spatialrust-io",
+    "spatialrust-records/receipt-json",
 ]
 
 [dependencies]
diff --git a/crates/spatialrust-pipeline/src/lib.rs b/crates/spatialrust-pipeline/src/lib.rs
index 384efc4..0bd70d6 100644
--- a/crates/spatialrust-pipeline/src/lib.rs
+++ b/crates/spatialrust-pipeline/src/lib.rs
@@ -13,9 +13,13 @@ pub use mvp::{
 
 #[cfg(feature = "pipeline-streaming")]
 mod streaming;
+#[cfg(feature = "pipeline-streaming")]
+mod workflow;
 
 #[cfg(feature = "pipeline-streaming")]
 pub use streaming::{
     reduce_positions, ChunkMapOperation, ChunkMapSource, PositionReduction, StreamingVoxelConfig,
     StreamingVoxelSource,
 };
+#[cfg(feature = "pipeline-streaming")]
+pub use workflow::{StreamingPipeline, StreamingPipelineIter};
diff --git a/crates/spatialrust-pipeline/src/workflow.rs b/crates/spatialrust-pipeline/src/workflow.rs
new file mode 100644
index 0000000..8768769
--- /dev/null
+++ b/crates/spatialrust-pipeline/src/workflow.rs
@@ -0,0 +1,285 @@
+//! Type-erased, metered bounded-memory streaming workflows.
+
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Instant;
+
+use spatialrust_math::Mat4;
+use spatialrust_records::{
+    BoundedSpatialRecordSink, BoundedSpatialRecordSource, CancellationToken, MemoryTracker,
+    RecordsError, RecordsResult, SchemaDescriptor, SpatialRecordChunk, StreamOptions,
+    StreamingReceipt,
+};
+
+use crate::{ChunkMapSource, StreamingVoxelConfig, StreamingVoxelSource};
+
+type ReceiptState = Arc<Mutex<StreamingReceipt>>;
+
+/// Composable bounded-memory point-cloud stream.
+pub struct StreamingPipeline {
+    source: Box<dyn BoundedSpatialRecordSource>,
+    receipt: ReceiptState,
+}
+
+impl StreamingPipeline {
+    /// Starts a workflow and meters chunks read from the original source.
+    pub fn new(
+        source: impl BoundedSpatialRecordSource + 'static,
+        source_id: impl Into<String>,
+    ) -> RecordsResult<Self> {
+        let receipt = Arc::new(Mutex::new(StreamingReceipt::new(source_id)?));
+        let source = MeteredInputSource { source, receipt: receipt.clone() };
+        Ok(Self { source: Box::new(source), receipt })
+    }
+
+    /// Returns the output schema.
+    #[must_use]
+    pub fn schema(&self) -> &SchemaDescriptor {
+        self.source.schema()
+    }
+
+    /// Returns the shared cooperative cancellation token.
+    #[must_use]
+    pub fn cancellation_token(&self) -> CancellationToken {
+        self.source.cancellation_token()
+    }
+
+    /// Adds an inclusive axis-aligned crop.
+    pub fn crop(self, min: [f32; 3], max: [f32; 3], invert: bool) -> RecordsResult<Self> {
+        let source = ChunkMapSource::crop(self.source, min, max, invert)?;
+        Ok(Self { source: Box::new(source), receipt: self.receipt })
+    }
+
+    /// Adds an affine position/normal transform.
+    pub fn transform(self, transform: Mat4<f32>) -> RecordsResult<Self> {
+        let source = ChunkMapSource::transform(self.source, transform)?;
+        Ok(Self { source: Box::new(source), receipt: self.receipt })
+    }
+
+    /// Adds deterministic global voxel aggregation backed by bounded spool storage.
+    pub fn voxel(self, config: StreamingVoxelConfig) -> RecordsResult<Self> {
+        let started = Instant::now();
+        let source = StreamingVoxelSource::try_build(self.source, config)?;
+        let spill_bytes = source.spool_bytes();
+        {
+            let mut receipt = lock_receipt(&self.receipt)?;
+            receipt.record_spill(spill_bytes)?;
+            receipt.record_phase("voxel", elapsed_ns(started), spill_bytes)?;
+            receipt.capture_memory(source.memory_tracker());
+        }
+        Ok(Self { source: Box::new(source), receipt: self.receipt })
+    }
+
+    /// Drains the workflow into a synchronous bounded sink and returns its receipt.
+    pub fn run_to_sink(
+        self,
+        sink: &mut dyn BoundedSpatialRecordSink,
+    ) -> RecordsResult<StreamingReceipt> {
+        let mut stream = self.into_iter();
+        for chunk in stream.by_ref() {
+            let chunk = chunk?;
+            sink.write_chunk(&chunk)?;
+        }
+        sink.finish()?;
+        stream.receipt()
+    }
+}
+
+impl IntoIterator for StreamingPipeline {
+    type Item = RecordsResult<SpatialRecordChunk>;
+    type IntoIter = StreamingPipelineIter;
+
+    fn into_iter(self) -> Self::IntoIter {
+        let tracker = self.source.memory_tracker().clone();
+        StreamingPipelineIter {
+            source: self.source,
+            receipt: self.receipt,
+            tracker,
+            completed: false,
+        }
+    }
+}
+
+/// Pull iterator that meters final output and exposes a live receipt snapshot.
+pub struct StreamingPipelineIter {
+    source: Box<dyn BoundedSpatialRecordSource>,
+    receipt: ReceiptState,
+    tracker: MemoryTracker,
+    completed: bool,
+}
+
+impl StreamingPipelineIter {
+    /// Returns the shared cooperative cancellation token.
+    #[must_use]
+    pub fn cancellation_token(&self) -> CancellationToken {
+        self.source.cancellation_token()
+    }
+
+    /// Clones the receipt as observed at the latest completed chunk boundary.
+    pub fn receipt(&self) -> RecordsResult<StreamingReceipt> {
+        let mut receipt = lock_receipt(&self.receipt)?;
+        receipt.capture_memory(&self.tracker);
+        Ok(receipt.clone())
+    }
+}
+
+impl Iterator for StreamingPipelineIter {
+    type Item = RecordsResult<SpatialRecordChunk>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.completed {
+            return None;
+        }
+        let next = self.source.next_chunk();
+        match next {
+            Some(Ok(chunk)) => {
+                let points = match u64::try_from(chunk.record().cloud().len()) {
+                    Ok(points) => points,
+                    Err(_) => {
+                        self.completed = true;
+                        return Some(Err(RecordsError::ReceiptOverflow(
+                            "pipeline output point count".into(),
+                        )));
+                    }
+                };
+                if let Err(error) = lock_receipt(&self.receipt).and_then(|mut receipt| {
+                    receipt.record_output_chunk(points, chunk.tracked_bytes())
+                }) {
+                    self.completed = true;
+                    return Some(Err(error));
+                }
+                Some(Ok(chunk))
+            }
+            Some(Err(error)) => {
+                self.completed = true;
+                Some(Err(error))
+            }
+            None => {
+                self.completed = true;
+                if let Ok(mut receipt) = lock_receipt(&self.receipt) {
+                    receipt.capture_memory(&self.tracker);
+                }
+                None
+            }
+        }
+    }
+}
+
+struct MeteredInputSource<S> {
+    source: S,
+    receipt: ReceiptState,
+}
+
+impl<S: BoundedSpatialRecordSource> BoundedSpatialRecordSource for MeteredInputSource<S> {
+    fn schema(&self) -> &SchemaDescriptor {
+        self.source.schema()
+    }
+
+    fn options(&self) -> &StreamOptions {
+        self.source.options()
+    }
+
+    fn memory_tracker(&self) -> &MemoryTracker {
+        self.source.memory_tracker()
+    }
+
+    fn cancellation_token(&self) -> CancellationToken {
+        self.source.cancellation_token()
+    }
+
+    fn max_chunk_bytes(&self) -> u64 {
+        self.source.max_chunk_bytes()
+    }
+
+    fn next_chunk(&mut self) -> Option<RecordsResult<SpatialRecordChunk>> {
+        match self.source.next_chunk()? {
+            Ok(chunk) => {
+                let points = match u64::try_from(chunk.record().cloud().len()) {
+                    Ok(points) => points,
+                    Err(_) => {
+                        return Some(Err(RecordsError::ReceiptOverflow(
+                            "pipeline input point count".into(),
+                        )));
+                    }
+                };
+                match lock_receipt(&self.receipt).and_then(|mut receipt| {
+                    receipt.record_input_chunk(points, chunk.tracked_bytes())
+                }) {
+                    Ok(()) => Some(Ok(chunk)),
+                    Err(error) => Some(Err(error)),
+                }
+            }
+            Err(error) => Some(Err(error)),
+        }
+    }
+}
+
+fn lock_receipt(receipt: &ReceiptState) -> RecordsResult<MutexGuard<'_, StreamingReceipt>> {
+    receipt
+        .lock()
+        .map_err(|_| RecordsError::InvalidReceipt("streaming receipt lock poisoned".into()))
+}
+
+fn elapsed_ns(started: Instant) -> u64 {
+    u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::StreamingPipeline;
+    use spatialrust_core::{HasPositions3, PointCloudBuilder, StandardSchemas};
+    use spatialrust_records::{
+        CancellationToken, MemoryBudget, RecyclingMemoryChunkSource, SchemaDescriptor,
+        SchemaVersion, StreamOptions,
+    };
+
+    fn source() -> RecyclingMemoryChunkSource {
+        let mut builder = PointCloudBuilder::xyz();
+        for point in [[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [2.0, 0.0, 0.0]] {
+            builder.push_point(point).unwrap();
+        }
+        let schema = SchemaDescriptor::try_new(
+            "workflow.xyz",
+            SchemaVersion::new(1, 0),
+            StandardSchemas::point_xyz(),
+        )
+        .unwrap();
+        let options = StreamOptions::new(2, MemoryBudget::new(4096).unwrap()).unwrap();
+        RecyclingMemoryChunkSource::try_new(
+            schema,
+            builder.build().unwrap(),
+            options,
+            CancellationToken::default(),
+        )
+        .unwrap()
+    }
+
+    #[test]
+    fn meters_input_and_output_around_composable_crop() {
+        let pipeline = StreamingPipeline::new(source(), "memory")
+            .unwrap()
+            .crop([0.5, -1.0, -1.0], [2.0, 1.0, 1.0], false)
+            .unwrap();
+        let mut stream = pipeline.into_iter();
+        let mut x = Vec::new();
+        for chunk in stream.by_ref() {
+            let chunk = chunk.unwrap();
+            x.extend_from_slice(chunk.record().cloud().positions3().unwrap().0);
+        }
+        assert_eq!(x, [1.0, 2.0]);
+        let receipt = stream.receipt().unwrap();
+        assert_eq!(receipt.input_points(), 3);
+        assert_eq!(receipt.output_points(), 2);
+        assert_eq!(receipt.chunks_read(), 2);
+        assert_eq!(receipt.chunks_written(), 2);
+    }
+
+    #[test]
+    fn iterator_observes_shared_cancellation() {
+        let pipeline = StreamingPipeline::new(source(), "memory").unwrap();
+        let token = pipeline.cancellation_token();
+        token.cancel();
+        let mut stream = pipeline.into_iter();
+        assert!(stream.next().unwrap().is_err());
+        assert!(stream.next().is_none());
+    }
+}
diff --git a/crates/spatialrust-py/Cargo.toml b/crates/spatialrust-py/Cargo.toml
index 5f7022e..3fe9199 100644
--- a/crates/spatialrust-py/Cargo.toml
+++ b/crates/spatialrust-py/Cargo.toml
@@ -48,6 +48,9 @@ spatialrust = { path = "../spatialrust", features = [
     "vision-full",
     "image-io-standard",
     "tensor-dlpack",
+    "pipeline-streaming",
+    "io-laz",
+    "records-receipt-json",
 ] }
 
 # Keep this crate out of the main Rust workspace so `cargo test --workspace`
diff --git a/crates/spatialrust-py/README.md b/crates/spatialrust-py/README.md
index cb89e9d..07e8583 100644
--- a/crates/spatialrust-py/README.md
+++ b/crates/spatialrust-py/README.md
@@ -63,6 +63,28 @@ editors and type checkers (mypy, pyright) get full autocomplete and signature
 checking for the compiled extension. CI runs `mypy.stubtest` on every push to
 keep the stubs in sync with the runtime API.
 
+## Bounded point-cloud streaming
+
+`open_point_cloud_stream()` reads local PCD/PLY/LAS/LAZ/COPC files through the
+same bounded Rust workflow used by the CLI. HTTP(S) COPC remains isolated in
+the CLI feature so default Python wheels do not acquire a TLS stack:
+
+```python
+stream = spatialrust.open_point_cloud_stream(
+    "scan.copc.laz",
+    chunk_points=65_536,
+    memory_budget_bytes=256 * 1024 * 1024,
+    crop=(0.0, 0.0, -10.0, 100.0, 100.0, 20.0),
+    voxel_leaf=0.1,
+)
+for chunk in stream:
+    process(chunk)
+print(stream.receipt_json())
+```
+
+Call `stream.cancel()` to stop cooperatively at a chunk boundary. Retained
+Python chunks are caller-owned and are outside the native pipeline budget.
+
 ## Quickstart
 
 ```python
diff --git a/crates/spatialrust-py/spatialrust.pyi b/crates/spatialrust-py/spatialrust.pyi
index 5a330be..5965d66 100644
--- a/crates/spatialrust-py/spatialrust.pyi
+++ b/crates/spatialrust-py/spatialrust.pyi
@@ -14,13 +14,14 @@ __version__: str
 __all__: list[str] = [
     "__version__", "ImageMetadata", "Tensor", "Keypoint2", "OnnxRuntimeSession",
     "GaussianBlurWorkspace",
-    "DLPackTensorView", "PointCloud", "PipelineResult", "RegionResult",
+    "DLPackTensorView", "PointCloud", "PointCloudStream", "PipelineResult", "RegionResult",
     "DbscanResult", "GroundResult", "MultiPlaneResult", "SphereResult",
     "CylinderResult", "RegistrationResult", "MultiObjectTracker", "read_image",
     "tensor_copy_from_numpy", "tensor_view_from_dlpack", "harris_keypoints",
     "shi_tomasi_keypoints", "fast_keypoints", "orb_features",
     "estimate_homography_ransac", "solve_pnp", "estimate_rgbd_odometry", "stereo_block_match",
     "match_binary_descriptors", "match_float_descriptors", "write_image", "read",
+    "open_point_cloud_stream",
     "write", "voxel_downsample", "crop_box", "pass_through", "iss_keypoints",
     "orient_normals", "detect_boundary", "mls_smooth", "farthest_point_sampling",
     "statistical_outlier_removal", "radius_outlier_removal", "run_pipeline",
@@ -104,6 +105,15 @@ class PointCloud:
     def __len__(self) -> int: ...
     def __repr__(self) -> str: ...
 
+@final
+class PointCloudStream:
+    """Pull-based bounded-memory point-cloud iterator."""
+
+    def __iter__(self) -> PointCloudStream: ...
+    def __next__(self) -> PointCloud: ...
+    def cancel(self) -> None: ...
+    def receipt_json(self) -> str: ...
+
 def depth_to_xyz(
     depth: _F32Array,
     fx: float,
@@ -551,6 +561,18 @@ class RegistrationResult:
 # --------------------------------------------------------------------------- #
 def read(path: str) -> PointCloud: ...
 def write(path: str, cloud: PointCloud) -> None: ...
+def open_point_cloud_stream(
+    path: str,
+    chunk_points: int = ...,
+    memory_budget_bytes: int = ...,
+    crop: Optional[tuple[float, float, float, float, float, float]] = ...,
+    translation: Optional[_Vec3] = ...,
+    voxel_leaf: Optional[float] = ...,
+    run_points: int = ...,
+    max_runs: int = ...,
+    spool_dir: Optional[str] = ...,
+    spool_limit_bytes: int = ...,
+) -> PointCloudStream: ...
 
 # --------------------------------------------------------------------------- #
 # Filters
diff --git a/crates/spatialrust-py/src/lib.rs b/crates/spatialrust-py/src/lib.rs
index 6295d87..2436a12 100644
--- a/crates/spatialrust-py/src/lib.rs
+++ b/crates/spatialrust-py/src/lib.rs
@@ -49,7 +49,12 @@ use spatialrust::image_io::{
 };
 use spatialrust::math::{Mat3, Mat4, Vec2, Vec3};
 use spatialrust::metrics::{chamfer_distance as chamfer, hausdorff_distance as hausdorff};
-use spatialrust::pipeline::{MvpPipeline, MvpPipelineConfig};
+use spatialrust::pipeline::{
+    MvpPipeline, MvpPipelineConfig, StreamingPipeline, StreamingPipelineIter, StreamingVoxelConfig,
+};
+use spatialrust::records::{
+    BoundedSpatialRecordSource, CancellationToken, MemoryBudget, StreamOptions,
+};
 use spatialrust::registration::{
     FpfhRansacConfig, FpfhRansacRegistration, GicpConfig, GicpRegistration, IcpConfig,
     IcpRegistration, NdtConfig, NdtRegistration, PointCloudRegistration, PointToPlaneIcp,
@@ -82,8 +87,7 @@ use spatialrust::vision::{
     erode_rect_u8_into as erode_rect_u8_into_op,
     estimate_homography_ransac as estimate_homography_ransac_op,
     estimate_rgbd_odometry as estimate_rgbd_odometry_op, filter2d as filter2d_op,
-    find_contours as trace_contours,
-    gaussian_blur_u8_into as gaussian_blur_u8_into_op,
+    find_contours as trace_contours, gaussian_blur_u8_into as gaussian_blur_u8_into_op,
     gray_world_white_balance as gray_world_white_balance_op, histogram_u8 as histogram_u8_op,
     integral_image as integral_image_op, laplacian as laplacian_op, letterbox as letterbox_op,
     match_descriptors as match_descriptors_op, median_blur as median_blur_op,
@@ -1037,6 +1041,129 @@ impl PyPointCloud {
     }
 }
 
+/// Pull-based bounded-memory point-cloud iterator.
+///
+/// Each yielded cloud owns its data; retaining yielded clouds is controlled by
+/// the Python caller and is separate from the native pipeline memory budget.
+#[pyclass(name = "PointCloudStream", unsendable)]
+pub struct PyPointCloudStream {
+    inner: StreamingPipelineIter,
+    cancellation: CancellationToken,
+}
+
+#[pymethods]
+impl PyPointCloudStream {
+    fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        slf
+    }
+
+    fn __next__(&mut self) -> PyResult<Option<PyPointCloud>> {
+        match self.inner.next() {
+            Some(Ok(chunk)) => Ok(Some(PyPointCloud { inner: chunk.record().cloud().clone() })),
+            Some(Err(error)) => Err(to_py_err(error)),
+            None => Ok(None),
+        }
+    }
+
+    /// Requests cooperative cancellation at the next chunk boundary.
+    fn cancel(&self) {
+        self.cancellation.cancel();
+    }
+
+    /// Returns the versioned JSON receipt observed so far.
+    fn receipt_json(&self) -> PyResult<String> {
+        self.inner.receipt().and_then(|receipt| receipt.to_json()).map_err(to_py_err)
+    }
+}
+
+/// Opens a local point-cloud file as a bounded iterator.
+#[pyfunction]
+#[pyo3(signature = (
+    path,
+    chunk_points=65536,
+    memory_budget_bytes=268435456,
+    crop=None,
+    translation=None,
+    voxel_leaf=None,
+    run_points=65536,
+    max_runs=1024,
+    spool_dir=None,
+    spool_limit_bytes=2147483648
+))]
+#[allow(clippy::too_many_arguments)]
+fn open_point_cloud_stream(
+    path: String,
+    chunk_points: usize,
+    memory_budget_bytes: u64,
+    crop: Option<(f32, f32, f32, f32, f32, f32)>,
+    translation: Option<(f32, f32, f32)>,
+    voxel_leaf: Option<f32>,
+    run_points: usize,
+    max_runs: usize,
+    spool_dir: Option<String>,
+    spool_limit_bytes: u64,
+) -> PyResult<PyPointCloudStream> {
+    let options = StreamOptions::new(
+        chunk_points,
+        MemoryBudget::new(memory_budget_bytes).map_err(to_py_err)?,
+    )
+    .map_err(to_py_err)?;
+    let cancellation = CancellationToken::default();
+    let source =
+        open_python_stream_source(&path, options, cancellation.clone()).map_err(to_py_err)?;
+    let mut pipeline = StreamingPipeline::new(source, path.clone()).map_err(to_py_err)?;
+    if let Some((min_x, min_y, min_z, max_x, max_y, max_z)) = crop {
+        pipeline = pipeline
+            .crop([min_x, min_y, min_z], [max_x, max_y, max_z], false)
+            .map_err(to_py_err)?;
+    }
+    if let Some((x, y, z)) = translation {
+        pipeline = pipeline
+            .transform(Mat4::<f32>::from_rotation_translation(
+                Mat3::<f32>::identity(),
+                Vec3::new(x, y, z),
+            ))
+            .map_err(to_py_err)?;
+    }
+    if let Some(leaf) = voxel_leaf {
+        let directory = spool_dir.map_or_else(std::env::temp_dir, std::path::PathBuf::from);
+        let spool =
+            spatialrust::io::SpoolOptions::new(directory, spool_limit_bytes).map_err(to_py_err)?;
+        let config =
+            StreamingVoxelConfig::new(leaf, run_points, max_runs, spool).map_err(to_py_err)?;
+        pipeline = pipeline.voxel(config).map_err(to_py_err)?;
+    }
+    Ok(PyPointCloudStream { inner: pipeline.into_iter(), cancellation })
+}
+
+fn open_python_stream_source(
+    path: &str,
+    options: StreamOptions,
+    cancellation: CancellationToken,
+) -> Result<Box<dyn BoundedSpatialRecordSource>, spatialrust::io::IoError> {
+    use spatialrust::io::{CopcChunkSource, LasChunkSource, PcdChunkSource, PlyChunkSource};
+
+    if path.starts_with("http://") || path.starts_with("https://") {
+        return Err(spatialrust::io::IoError::Streaming(
+            "Python wheels accept local streams; use spatialrust-stream for HTTP(S) COPC".into(),
+        ));
+    }
+    let lower = path.to_ascii_lowercase();
+    if lower.ends_with(".copc.laz") {
+        Ok(Box::new(CopcChunkSource::open(path, None, options, cancellation)?))
+    } else if lower.ends_with(".pcd") {
+        Ok(Box::new(PcdChunkSource::open(path, options, cancellation)?))
+    } else if lower.ends_with(".ply") {
+        Ok(Box::new(PlyChunkSource::open(path, options, cancellation)?))
+    } else if lower.ends_with(".las") || lower.ends_with(".laz") {
+        Ok(Box::new(LasChunkSource::open(path, options, cancellation)?))
+    } else {
+        Err(spatialrust::io::IoError::Streaming(format!(
+            "unsupported bounded stream input '{path}'"
+        )))
+    }
+}
+
 /// Result of running the MVP pipeline.
 #[pyclass(name = "PipelineResult")]
 pub struct PyPipelineResult {
@@ -2289,8 +2416,8 @@ fn gaussian_blur_image_with_workspace<'py>(
         workspace,
     )
     .map_err(to_py_err)?;
-    let array = Array3::from_shape_vec((image.height(), image.width(), 3), output)
-        .map_err(to_py_err)?;
+    let array =
+        Array3::from_shape_vec((image.height(), image.width(), 3), output).map_err(to_py_err)?;
     Ok(array.into_pyarray_bound(py))
 }
 
@@ -4315,6 +4442,7 @@ fn spatialrust_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<PyOnnxRuntimeSession>()?;
     m.add_class::<PyDlpackTensorView>()?;
     m.add_class::<PyPointCloud>()?;
+    m.add_class::<PyPointCloudStream>()?;
     m.add_class::<PyPipelineResult>()?;
     m.add_class::<PyRegionResult>()?;
     m.add_class::<PyDbscanResult>()?;
@@ -4340,6 +4468,7 @@ fn spatialrust_module(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(match_float_descriptors, m)?)?;
     m.add_function(wrap_pyfunction!(write_image, m)?)?;
     m.add_function(wrap_pyfunction!(read, m)?)?;
+    m.add_function(wrap_pyfunction!(open_point_cloud_stream, m)?)?;
     m.add_function(wrap_pyfunction!(write, m)?)?;
     m.add_function(wrap_pyfunction!(voxel_downsample, m)?)?;
     m.add_function(wrap_pyfunction!(crop_box, m)?)?;
diff --git a/crates/spatialrust-py/tests/test_bindings.py b/crates/spatialrust-py/tests/test_bindings.py
index c24f288..40cc7a4 100644
--- a/crates/spatialrust-py/tests/test_bindings.py
+++ b/crates/spatialrust-py/tests/test_bindings.py
@@ -61,7 +61,8 @@ def test_module_has_version():
 
 def test_exports_present():
     for name in (
-        "PointCloud", "voxel_downsample", "dbscan", "register_icp",
+        "PointCloud", "PointCloudStream", "open_point_cloud_stream",
+        "voxel_downsample", "dbscan", "register_icp",
         "voxelize", "knn_graph", "chamfer_distance", "oriented_bounding_box",
         "rgbd_to_point_cloud", "depth_to_xyz",
         "resize_image", "letterbox_image", "normalize_image_chw", "resize_normalize_image_chw",
@@ -73,6 +74,27 @@ def test_exports_present():
         assert hasattr(sr, name), f"missing export: {name}"
 
 
+def test_bounded_point_cloud_stream_and_receipt(tmp_path):
+    path = tmp_path / "tiny.pcd"
+    path.write_text(
+        "VERSION .7\nFIELDS x y z\nSIZE 4 4 4\nTYPE F F F\n"
+        "COUNT 1 1 1\nWIDTH 3\nHEIGHT 1\nPOINTS 3\nDATA ascii\n"
+        "0 0 0\n1 0 0\n2 0 0\n",
+        encoding="ascii",
+    )
+    stream = sr.open_point_cloud_stream(
+        str(path),
+        chunk_points=1,
+        memory_budget_bytes=4096,
+        crop=(0.5, -1.0, -1.0, 2.0, 1.0, 1.0),
+    )
+    chunks = list(stream)
+    assert [len(chunk) for chunk in chunks] == [1, 1]
+    receipt = __import__("json").loads(stream.receipt_json())
+    assert receipt["input_points"] == 3
+    assert receipt["output_points"] == 2
+
+
 # --------------------------------------------------------------------------- #
 # PointCloud round-trips
 # --------------------------------------------------------------------------- #
diff --git a/crates/spatialrust-records/src/bounded.rs b/crates/spatialrust-records/src/bounded.rs
index 9e6b59c..abf1b08 100644
--- a/crates/spatialrust-records/src/bounded.rs
+++ b/crates/spatialrust-records/src/bounded.rs
@@ -168,6 +168,32 @@ pub trait BoundedSpatialRecordSource {
     fn next_chunk(&mut self) -> Option<RecordsResult<SpatialRecordChunk>>;
 }
 
+impl<S: BoundedSpatialRecordSource + ?Sized> BoundedSpatialRecordSource for Box<S> {
+    fn schema(&self) -> &SchemaDescriptor {
+        (**self).schema()
+    }
+
+    fn options(&self) -> &StreamOptions {
+        (**self).options()
+    }
+
+    fn memory_tracker(&self) -> &MemoryTracker {
+        (**self).memory_tracker()
+    }
+
+    fn cancellation_token(&self) -> CancellationToken {
+        (**self).cancellation_token()
+    }
+
+    fn max_chunk_bytes(&self) -> u64 {
+        (**self).max_chunk_bytes()
+    }
+
+    fn next_chunk(&mut self) -> Option<RecordsResult<SpatialRecordChunk>> {
+        (**self).next_chunk()
+    }
+}
+
 /// Push-based bounded sink that cannot retain a record beyond the chunk lease.
 pub trait BoundedSpatialRecordSink {
     /// Writes one chunk synchronously.
@@ -179,6 +205,16 @@ pub trait BoundedSpatialRecordSink {
     }
 }
 
+impl<S: BoundedSpatialRecordSink + ?Sized> BoundedSpatialRecordSink for Box<S> {
+    fn write_chunk(&mut self, chunk: &SpatialRecordChunk) -> RecordsResult<()> {
+        (**self).write_chunk(chunk)
+    }
+
+    fn finish(&mut self) -> RecordsResult<()> {
+        (**self).finish()
+    }
+}
+
 /// Adapts an existing record source without changing its public trait.
 pub struct LegacyBoundedSource<S> {
     source: S,
diff --git a/crates/spatialrust/Cargo.toml b/crates/spatialrust/Cargo.toml
index 7005722..12705b8 100644
--- a/crates/spatialrust/Cargo.toml
+++ b/crates/spatialrust/Cargo.toml
@@ -119,6 +119,16 @@ pipeline-streaming = [
     "io-streaming",
     "spatialrust-pipeline/pipeline-streaming",
 ]
+streaming-cli = [
+    "pipeline-streaming",
+    "io-pcd",
+    "io-ply",
+    "io-las",
+    "io-laz",
+    "io-copc-http",
+    "records-receipt-json",
+    "dep:ctrlc",
+]
 pipeline-mvp-gpu = [
     "spatialrust-pipeline/pipeline-mvp-gpu",
     "filter-voxel-gpu",
@@ -255,6 +265,7 @@ spatialrust-runtime = { workspace = true, optional = true }
 spatialrust-interchange = { workspace = true, optional = true }
 spatialrust-distribute = { workspace = true, optional = true }
 spatialrust-platform = { workspace = true, optional = true }
+ctrlc = { version = "3.4", optional = true }
 
 [dev-dependencies]
 
@@ -263,6 +274,11 @@ name = "vision_api_v1"
 path = "tests/vision_api_v1.rs"
 required-features = ["platform", "camera-rgbd", "vision-full"]
 
+[[test]]
+name = "streaming_cli"
+path = "tests/streaming_cli.rs"
+required-features = ["streaming-cli"]
+
 [[example]]
 name = "north_star_demo"
 path = "examples/north_star_demo.rs"
@@ -293,6 +309,11 @@ name = "spatialrust-mvp"
 path = "src/bin/spatialrust_mvp.rs"
 required-features = ["mvp"]
 
+[[bin]]
+name = "spatialrust-stream"
+path = "src/bin/spatialrust_stream.rs"
+required-features = ["streaming-cli"]
+
 [[example]]
 name = "readme_mvp_preview"
 path = "examples/readme_mvp_preview.rs"
diff --git a/crates/spatialrust/src/bin/spatialrust_stream.rs b/crates/spatialrust/src/bin/spatialrust_stream.rs
new file mode 100644
index 0000000..b952a13
--- /dev/null
+++ b/crates/spatialrust/src/bin/spatialrust_stream.rs
@@ -0,0 +1,217 @@
+//! Bounded-memory point-cloud streaming command line workflow.
+
+use std::error::Error;
+use std::path::{Path, PathBuf};
+
+use spatialrust::io::{
+    CopcChunkSource, LasChunkSink, LasChunkSource, LasWriteFormat, PcdChunkSource, PlyChunkSource,
+    SpoolOptions,
+};
+use spatialrust::math::{Mat3, Mat4, Vec3};
+use spatialrust::pipeline::{StreamingPipeline, StreamingVoxelConfig};
+use spatialrust::records::{
+    BoundedSpatialRecordSource, CancellationToken, MemoryBudget, StreamOptions,
+    DEFAULT_STREAM_CHUNK_POINTS, DEFAULT_STREAM_MEMORY_BUDGET_BYTES,
+};
+
+#[derive(Debug)]
+struct Config {
+    input: String,
+    output: PathBuf,
+    chunk_points: usize,
+    memory_bytes: u64,
+    crop: Option<([f32; 3], [f32; 3])>,
+    translation: Option<[f32; 3]>,
+    voxel_leaf: Option<f32>,
+    run_points: usize,
+    max_runs: usize,
+    spool_dir: PathBuf,
+    spool_bytes: u64,
+    receipt: Option<PathBuf>,
+}
+
+fn main() {
+    if let Err(error) = run() {
+        eprintln!("spatialrust-stream: {error}");
+        std::process::exit(2);
+    }
+}
+
+fn run() -> Result<(), Box<dyn Error>> {
+    let config = parse_args(std::env::args().skip(1))?;
+    let options = StreamOptions::new(config.chunk_points, MemoryBudget::new(config.memory_bytes)?)?;
+    let cancellation = CancellationToken::default();
+    let interrupt = cancellation.clone();
+    ctrlc::set_handler(move || interrupt.cancel())?;
+
+    let source = open_source(&config.input, options, cancellation)?;
+    let mut pipeline = StreamingPipeline::new(source, config.input.clone())?;
+    if let Some((min, max)) = config.crop {
+        pipeline = pipeline.crop(min, max, false)?;
+    }
+    if let Some(translation) = config.translation {
+        pipeline = pipeline.transform(Mat4::<f32>::from_rotation_translation(
+            Mat3::<f32>::identity(),
+            Vec3::new(translation[0], translation[1], translation[2]),
+        ))?;
+    }
+    if let Some(leaf) = config.voxel_leaf {
+        let spool = SpoolOptions::new(&config.spool_dir, config.spool_bytes)?;
+        pipeline = pipeline.voxel(StreamingVoxelConfig::new(
+            leaf,
+            config.run_points,
+            config.max_runs,
+            spool,
+        )?)?;
+    }
+
+    let format = output_format(&config.output)?;
+    let mut sink =
+        LasChunkSink::create_open_ended(&config.output, pipeline.schema().clone(), format)?;
+    let receipt = pipeline.run_to_sink(&mut sink)?;
+    let json = receipt.to_json()?;
+    if let Some(path) = config.receipt {
+        std::fs::write(path, format!("{json}\n"))?;
+    } else {
+        println!("{json}");
+    }
+    Ok(())
+}
+
+fn open_source(
+    input: &str,
+    options: StreamOptions,
+    cancellation: CancellationToken,
+) -> Result<Box<dyn BoundedSpatialRecordSource>, Box<dyn Error>> {
+    if input.starts_with("http://") || input.starts_with("https://") {
+        return Ok(Box::new(CopcChunkSource::open_url(input, None, options, cancellation)?));
+    }
+    let lower = input.to_ascii_lowercase();
+    if lower.ends_with(".copc.laz") {
+        Ok(Box::new(CopcChunkSource::open(input, None, options, cancellation)?))
+    } else if lower.ends_with(".pcd") {
+        Ok(Box::new(PcdChunkSource::open(input, options, cancellation)?))
+    } else if lower.ends_with(".ply") {
+        Ok(Box::new(PlyChunkSource::open(input, options, cancellation)?))
+    } else if lower.ends_with(".las") || lower.ends_with(".laz") {
+        Ok(Box::new(LasChunkSource::open(input, options, cancellation)?))
+    } else {
+        Err(format!(
+            "unsupported input '{input}'; expected PCD, PLY, LAS, LAZ, COPC, or an HTTP(S) COPC URL"
+        )
+        .into())
+    }
+}
+
+fn output_format(path: &Path) -> Result<LasWriteFormat, Box<dyn Error>> {
+    let lower = path.to_string_lossy().to_ascii_lowercase();
+    if lower.ends_with(".las") {
+        Ok(LasWriteFormat::Las)
+    } else if lower.ends_with(".laz") {
+        Ok(LasWriteFormat::Laz)
+    } else {
+        Err("output must end in .las or .laz".into())
+    }
+}
+
+fn parse_args(args: impl IntoIterator<Item = String>) -> Result<Config, Box<dyn Error>> {
+    let mut args = args.into_iter();
+    let input = args.next().ok_or_else(usage)?;
+    if input == "-h" || input == "--help" {
+        return Err(usage().into());
+    }
+    let output = PathBuf::from(args.next().ok_or_else(usage)?);
+    let mut config = Config {
+        input,
+        output,
+        chunk_points: DEFAULT_STREAM_CHUNK_POINTS,
+        memory_bytes: DEFAULT_STREAM_MEMORY_BUDGET_BYTES,
+        crop: None,
+        translation: None,
+        voxel_leaf: None,
+        run_points: DEFAULT_STREAM_CHUNK_POINTS,
+        max_runs: 1024,
+        spool_dir: std::env::temp_dir(),
+        spool_bytes: DEFAULT_STREAM_MEMORY_BUDGET_BYTES.saturating_mul(8),
+        receipt: None,
+    };
+    while let Some(flag) = args.next() {
+        match flag.as_str() {
+            "--chunk-points" => config.chunk_points = parse_one(&mut args, &flag)?,
+            "--memory-budget" => config.memory_bytes = parse_one(&mut args, &flag)?,
+            "--voxel" => config.voxel_leaf = Some(parse_one(&mut args, &flag)?),
+            "--run-points" => config.run_points = parse_one(&mut args, &flag)?,
+            "--max-runs" => config.max_runs = parse_one(&mut args, &flag)?,
+            "--spool-limit" => config.spool_bytes = parse_one(&mut args, &flag)?,
+            "--spool-dir" => config.spool_dir = PathBuf::from(next_value(&mut args, &flag)?),
+            "--receipt" => config.receipt = Some(PathBuf::from(next_value(&mut args, &flag)?)),
+            "--translate" => {
+                config.translation = Some([
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                ]);
+            }
+            "--crop" => {
+                let values = [
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                    parse_one(&mut args, &flag)?,
+                ];
+                config.crop =
+                    Some(([values[0], values[1], values[2]], [values[3], values[4], values[5]]));
+            }
+            _ => return Err(format!("unknown option '{flag}'\n{}", usage()).into()),
+        }
+    }
+    Ok(config)
+}
+
+fn parse_one<T: std::str::FromStr>(
+    args: &mut impl Iterator<Item = String>,
+    flag: &str,
+) -> Result<T, Box<dyn Error>>
+where
+    T::Err: Error + 'static,
+{
+    Ok(next_value(args, flag)?.parse()?)
+}
+
+fn next_value(
+    args: &mut impl Iterator<Item = String>,
+    flag: &str,
+) -> Result<String, Box<dyn Error>> {
+    args.next().ok_or_else(|| format!("{flag} requires another value").into())
+}
+
+fn usage() -> String {
+    "usage: spatialrust-stream INPUT OUTPUT [--chunk-points N] [--memory-budget BYTES] \
+     [--crop MINX MINY MINZ MAXX MAXY MAXZ] [--translate X Y Z] [--voxel LEAF] \
+     [--run-points N] [--max-runs N] [--spool-dir DIR] [--spool-limit BYTES] \
+     [--receipt PATH]"
+        .into()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{output_format, parse_args};
+    use spatialrust::io::LasWriteFormat;
+
+    #[test]
+    fn parses_workflow_options() {
+        let config = parse_args(
+            ["in.pcd", "out.laz", "--chunk-points", "10", "--voxel", "0.2", "--crop"]
+                .into_iter()
+                .chain(["0", "1", "2", "3", "4", "5"])
+                .map(str::to_owned),
+        )
+        .unwrap();
+        assert_eq!(config.chunk_points, 10);
+        assert_eq!(config.voxel_leaf, Some(0.2));
+        assert_eq!(config.crop, Some(([0.0, 1.0, 2.0], [3.0, 4.0, 5.0])));
+        assert_eq!(output_format(&config.output).unwrap(), LasWriteFormat::Laz);
+    }
+}
diff --git a/crates/spatialrust/tests/streaming_cli.rs b/crates/spatialrust/tests/streaming_cli.rs
new file mode 100644
index 0000000..8e5a630
--- /dev/null
+++ b/crates/spatialrust/tests/streaming_cli.rs
@@ -0,0 +1,41 @@
+use std::process::Command;
+
+use spatialrust::records::StreamingReceipt;
+
+#[test]
+fn streams_pcd_through_crop_to_las_with_receipt() {
+    let directory =
+        std::env::temp_dir().join(format!("spatialrust-cli-e2e-{}", std::process::id()));
+    std::fs::create_dir_all(&directory).unwrap();
+    let input = directory.join("input.pcd");
+    let output = directory.join("output.las");
+    let receipt_path = directory.join("receipt.json");
+    std::fs::write(
+        &input,
+        "VERSION .7\nFIELDS x y z\nSIZE 4 4 4\nTYPE F F F\nCOUNT 1 1 1\n\
+         WIDTH 3\nHEIGHT 1\nPOINTS 3\nDATA ascii\n0 0 0\n1 0 0\n2 0 0\n",
+    )
+    .unwrap();
+
+    let status = Command::new(env!("CARGO_BIN_EXE_spatialrust-stream"))
+        .arg(&input)
+        .arg(&output)
+        .args(["--chunk-points", "1", "--memory-budget", "4096", "--crop"])
+        .args(["0.5", "-1", "-1", "2", "1", "1"])
+        .arg("--receipt")
+        .arg(&receipt_path)
+        .status()
+        .unwrap();
+    assert!(status.success());
+
+    let cloud = spatialrust::read_las_file(&output).unwrap();
+    assert_eq!(cloud.len(), 2);
+    let receipt =
+        StreamingReceipt::from_json(&std::fs::read_to_string(&receipt_path).unwrap()).unwrap();
+    assert_eq!(receipt.input_points(), 3);
+    assert_eq!(receipt.output_points(), 2);
+    assert_eq!(receipt.chunks_read(), 3);
+    assert_eq!(receipt.chunks_written(), 2);
+
+    std::fs::remove_dir_all(directory).unwrap();
+}
diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md
index f530c4b..0957702 100644
--- a/docs/ROADMAP.md
+++ b/docs/ROADMAP.md
@@ -711,7 +711,7 @@ outside core.
 | 122 | Complete | 121 | Backward-compatible bounded record sources/sinks with chunk identity, deterministic ordering, prefetch, and buffer reuse |
 | 123 | Complete | 122 | Local/HTTP COPC plus PCD/PLY/LAS/LAZ streaming adapters and bounded temporary spool contracts |
 | 124 | Complete | 122–123 | Chunk-safe crop/transform/reductions and deterministic global voxel aggregation with explicit spill |
-| 125 | Planned | 123–124 | Composable Rust pipeline, CLI, Python iterator, cancellation, and reproducible end-to-end receipt |
+| 125 | Complete | 123–124 | Composable Rust pipeline, CLI, Python iterator, cancellation, and reproducible end-to-end receipt |
 | 126 | Planned | 121–125 | Linux/Windows/macOS conformance, memory/copy budgets, documentation, migration notes, and the 1.2 release gate |
 
 ### Epic 121 delivery slices
@@ -750,6 +750,15 @@ outside core.
 | 124C | Complete | Fixed-width external voxel runs sorted by voxel key and global source offset | output equality across input chunk and run sizes plus attribute centroid tests |
 | 124D | Complete | Bounded disk extent, run/file-handle limit, merge memory, cancellation, and public feature surface | spill/run denial, cleanup, cancellation-release, root API, docs, and example gates |
 
+### Epic 125 delivery slices
+
+| Slice | Status | Scope | Evidence |
+| --- | --- | --- | --- |
+| 125A | Complete | Type-erased Rust builder and metered pull iterator over the Epic 124 operations | crop composition, cancellation, and receipt-counter tests |
+| 125B | Complete | Local/HTTP input CLI with open-ended LAS/LAZ output and Ctrl-C cancellation | real PCD → crop → LAS subprocess E2E |
+| 125C | Complete | Python iterator backed by the same Rust workflow with cancellation and live receipt JSON | extension compile gate, stubs, and wheel smoke test |
+| 125D | Complete | Reproducible workflow documentation and dated implementation receipt | `STREAMING_PIPELINE.md` and Epic 125 receipt |
+
 ### SpatialRust 1.2 exclusions
 
 - Native ROS 2/rclrs integration, CUDA, SLAM, and reconstruction expansion.
diff --git a/docs/STREAMING_PIPELINE.md b/docs/STREAMING_PIPELINE.md
index 14a452e..4cc23f4 100644
--- a/docs/STREAMING_PIPELINE.md
+++ b/docs/STREAMING_PIPELINE.md
@@ -44,6 +44,39 @@ merge records, accumulators, and emitted columns share the upstream memory
 tracker. `spool_bytes()`, `run_count()`, and the tracker snapshot provide
 receipt inputs.
 
+## End-to-end workflows
+
+`StreamingPipeline` type-erases bounded sources while retaining their shared
+memory tracker and cancellation token. Its `crop`, `transform`, and `voxel`
+builders feed either a Rust iterator or `run_to_sink`; input/output chunks,
+tracked bytes, peak memory, voxel phase time, and spill extent are written to
+one versioned `StreamingReceipt`.
+
+The `spatialrust-stream` binary accepts local PCD/PLY/LAS/LAZ/COPC or an
+HTTP(S) COPC URL, writes LAS/LAZ without knowing the filtered point count in
+advance, and emits receipt JSON:
+
+```powershell
+cargo run -p spatialrust --features streaming-cli --bin spatialrust-stream -- `
+  input.copc.laz output.laz --chunk-points 65536 --memory-budget 268435456 `
+  --crop 0 0 -10 100 100 20 --voxel 0.1 --receipt receipt.json
+```
+
+The Python extension uses the same Rust iterator:
+
+```python
+stream = spatialrust.open_point_cloud_stream(
+    "input.pcd", chunk_points=65_536, memory_budget_bytes=268_435_456,
+)
+for chunk in stream:
+    consume(chunk)
+print(stream.receipt_json())
+```
+
+`stream.cancel()` requests cooperative cancellation. Each yielded Python
+`PointCloud` owns a copy, so retaining Python outputs is caller-managed and is
+not included in the native pipeline budget.
+
 ```powershell
 cargo run -p spatialrust-pipeline --example bounded_voxel `
   --no-default-features --features pipeline-streaming
diff --git a/docs/receipts/2026-07-27_epic-125-streaming-e2e.md b/docs/receipts/2026-07-27_epic-125-streaming-e2e.md
new file mode 100644
index 0000000..d26abaa
--- /dev/null
+++ b/docs/receipts/2026-07-27_epic-125-streaming-e2e.md
@@ -0,0 +1,35 @@
+# Epic 125 bounded streaming end-to-end receipt
+
+Date: 2026-07-27
+
+## Delivered
+
+- Type-erased `StreamingPipeline` with composable crop, affine transform,
+  deterministic spill-backed voxel aggregation, metered iterator, sink drain,
+  shared cancellation, and versioned receipt snapshots.
+- Open-ended LAS/LAZ sink whose seekable writer finalizes the actual point
+  count after filters.
+- `spatialrust-stream` for local PCD/PLY/LAS/LAZ/COPC and HTTP(S) COPC input,
+  LAS/LAZ output, bounded spool configuration, Ctrl-C cancellation, and JSON
+  receipts.
+- Python `PointCloudStream` backed by the same Rust iterator, including
+  `cancel()` and `receipt_json()`; HTTP/TLS remains isolated from default
+  Python wheels.
+
+## Verification
+
+- `cargo test -p spatialrust-pipeline --features pipeline-streaming --lib`
+- `cargo test -p spatialrust-io --features streaming,io-las,io-laz --lib las::writer`
+- `cargo test -p spatialrust --features streaming-cli --bin spatialrust-stream`
+- `cargo test -p spatialrust --features streaming-cli --test streaming_cli`
+- `cargo check --manifest-path crates/spatialrust-py/Cargo.toml`
+- `maturin build --manifest-path crates/spatialrust-py/Cargo.toml --interpreter python`
+- `pytest crates/spatialrust-py/tests/test_bindings.py::test_bounded_point_cloud_stream_and_receipt`
+- `cargo test --workspace --all-features`
+
+## Relevant files
+
+- `C:\Users\rsasa\Workspace\SpatialRust\crates\spatialrust-pipeline\src\workflow.rs`
+- `C:\Users\rsasa\Workspace\SpatialRust\crates\spatialrust\src\bin\spatialrust_stream.rs`
+- `C:\Users\rsasa\Workspace\SpatialRust\crates\spatialrust-py\src\lib.rs`
+- `C:\Users\rsasa\Workspace\SpatialRust\docs\STREAMING_PIPELINE.md`