diff --git a/Cargo.toml b/Cargo.toml
index 9f814162..851cfef5 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,6 +16,7 @@ path = "rust-arroyo/src/lib.rs"
 ssl = ["rdkafka/ssl"]
 
 [dependencies]
+async-stream = "0.3"
 chrono = "0.4.26"
 coarsetime = "0.1.33"
 once_cell = "1.18.0"
@@ -26,12 +27,15 @@ serde = { version = "1.0.137", features = ["derive"] }
 serde_json = "1.0.81"
 thiserror = "1.0"
 tokio = { version = "1.19.2", features = ["full"] }
+futures = "0.3"
+tokio-stream = { version = "0.1", features = ["time"] }
+tokio-util = "0.7"
 tracing = "0.1.40"
 uuid = { version = "1.5.0", features = ["v4"] }
 parking_lot = "0.12.1"
 
 [dev-dependencies]
-tracing-subscriber = "0.3.18"
+tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 
 [[example]]
 name = "base_processor"
@@ -44,3 +48,7 @@ path = "rust-arroyo/examples/transform_and_produce.rs"
 [[example]]
 name = "base_consumer"
 path = "rust-arroyo/examples/base_consumer.rs"
+
+[[example]]
+name = "transform_and_produce_pull"
+path = "rust-arroyo/examples/transform_and_produce_pull.rs"
diff --git a/rust-arroyo/examples/transform_and_produce_pull.rs b/rust-arroyo/examples/transform_and_produce_pull.rs
new file mode 100644
index 00000000..8d7b1b48
--- /dev/null
+++ b/rust-arroyo/examples/transform_and_produce_pull.rs
@@ -0,0 +1,97 @@
+/// Pull-based version of transform_and_produce.
+///
+/// Pipeline:
+///   KafkaSource → apply(reverse) → on_next(produce) → on_reject(log) → commit
+///
+/// `PipelineRunner::run()` handles the rebalance restart loop —
+/// the closure is called once per partition assignment with fresh
+/// stages, handlers, and tracker.
+extern crate sentry_arroyo;
+
+use std::time::Duration;
+
+use sentry_arroyo::backends::kafka::config::KafkaConfig;
+use sentry_arroyo::backends::kafka::producer::KafkaProducer;
+use sentry_arroyo::backends::kafka::types::KafkaPayload;
+use sentry_arroyo::backends::kafka::InitialOffset;
+use sentry_arroyo::processing::stream::{
+    KafkaProducerHandler, KafkaSource, LogHandler, OffsetTracker, PipelineEnvelope, PipelineExt,
+    PipelineRunner, Stage, StageResult,
+};
+use sentry_arroyo::types::{Topic, TopicOrPartition};
+
+/// A Stage that reverses the string payload.
+struct ReverseStage;
+
+impl Stage for ReverseStage {
+    type In = KafkaPayload;
+    type Out = KafkaPayload;
+
+    async fn process(&self, envelope: PipelineEnvelope<KafkaPayload>) -> StageResult<KafkaPayload> {
+        let reversed = envelope.map_payload(|p| {
+            let bytes = p.payload().unwrap();
+            let s = std::str::from_utf8(bytes).unwrap();
+            let reversed: String = s.chars().rev().collect();
+            println!("transforming: {:?} -> {:?}", s, reversed);
+            KafkaPayload::new(
+                p.key().cloned(),
+                p.headers().cloned(),
+                Some(reversed.into_bytes()),
+            )
+        });
+
+        StageResult::Emit(reversed)
+    }
+
+    fn name(&self) -> &'static str {
+        "reverse_string"
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt::init();
+
+    let consumer_config = KafkaConfig::new_consumer_config(
+        vec!["0.0.0.0:9092".to_string()],
+        "my_group".to_string(),
+        InitialOffset::Latest,
+        false,
+        30_000,
+        None,
+    );
+    let source = KafkaSource::new(consumer_config, &[Topic::new("test_in")]);
+
+    let producer_config = KafkaConfig::new_producer_config(vec!["0.0.0.0:9092".to_string()], None);
+    // The pipeline reads left to right, top to bottom:
+    //   stream                     — async stream of Kafka messages
+    //     .apply(&stage)           — transform/filter/batch each message
+    //     .on_next(&handler)       — side-effect on successful items (produce, upload)
+    //     .on_reject(&handler)     — handle rejected items (DLQ, log)
+    //     .commit(&mut tracker)    — track offsets, flush on interval
+    //
+    // PipelineRunner::run() handles the lifecycle:
+    //   - calls the closure once per partition assignment
+    //   - on rebalance: closure is called again with a fresh stream
+    //   - on shutdown or stream end: exits
+    let result = PipelineRunner::run(&source, |stream, committer| async {
+        let reverse = ReverseStage;
+        let producer = KafkaProducer::new(producer_config.clone());
+        let produce_handler =
+            KafkaProducerHandler::new(producer, TopicOrPartition::Topic(Topic::new("test_out")));
+        let error_handler = LogHandler;
+        let mut tracker = OffsetTracker::new(Duration::from_secs(1), committer);
+
+        stream
+            .apply(&reverse)
+            .on_next(&produce_handler)
+            .on_reject(&error_handler)
+            .commit(&mut tracker)
+            .await
+    })
+    .await;
+
+    if let Err(e) = result {
+        tracing::error!("Pipeline stopped: {}", e);
+    }
+}
diff --git a/rust-arroyo/src/backends/kafka/mod.rs b/rust-arroyo/src/backends/kafka/mod.rs
index 295ae8fc..86103dba 100644
--- a/rust-arroyo/src/backends/kafka/mod.rs
+++ b/rust-arroyo/src/backends/kafka/mod.rs
@@ -82,7 +82,7 @@ impl KafkaConsumerState {
 }
 
 /// Treat recoverable `librdkafka` errors as an empty poll the same way Python treats `KafkaError._TRANSPORT`.
-fn kafka_poll_error_is_recoverable(err: &KafkaError) -> bool {
+pub fn kafka_poll_error_is_recoverable(err: &KafkaError) -> bool {
     matches!(
         err,
         KafkaError::MessageConsumption(RDKafkaErrorCode::BrokerTransportFailure)
diff --git a/rust-arroyo/src/processing/mod.rs b/rust-arroyo/src/processing/mod.rs
index 2238c18f..d24b0d09 100644
--- a/rust-arroyo/src/processing/mod.rs
+++ b/rust-arroyo/src/processing/mod.rs
@@ -22,6 +22,7 @@ use crate::{counter, timer};
 pub mod dlq;
 mod metrics_buffer;
 pub mod strategies;
+pub mod stream;
 
 use strategies::{ProcessingStrategy, ProcessingStrategyFactory};
 
diff --git a/rust-arroyo/src/processing/stream/ext.rs b/rust-arroyo/src/processing/stream/ext.rs
new file mode 100644
index 00000000..25233848
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/ext.rs
@@ -0,0 +1,199 @@
+use std::time::Instant;
+
+use futures::stream::Stream;
+use futures::StreamExt;
+
+use super::offset_tracker::OffsetTracker;
+use crate::{counter, timer};
+
+use super::handlers::next::NextHandler;
+use super::handlers::rejection::{RejectionHandler, RejectionMetadata};
+use super::pipeline_envelope::PipelineEnvelope;
+use super::stage::{PipelineExit, Stage, StageResult};
+
+/// Run a stage on an envelope and record metrics.
+async fn run_stage<S: Stage>(stage: &S, envelope: PipelineEnvelope<S::In>) -> StageResult<S::Out> {
+    let start = Instant::now();
+    let result = stage.process(envelope).await;
+
+    timer!("arroyo.stage.duration", start.elapsed(), "stage" => stage.name());
+    match &result {
+        StageResult::Emit(_) => {
+            counter!("arroyo.stage.success", 1, "stage" => stage.name());
+        }
+        StageResult::Drop { .. } => {
+            counter!("arroyo.stage.drop", 1, "stage" => stage.name());
+        }
+        StageResult::Skip => {
+            counter!("arroyo.stage.skip", 1, "stage" => stage.name());
+        }
+        StageResult::Reject { .. } => {
+            counter!("arroyo.stage.reject", 1, "stage" => stage.name());
+        }
+        StageResult::Fail(_) => {
+            counter!("arroyo.stage.fail", 1, "stage" => stage.name());
+        }
+        StageResult::Exit(_) => {}
+    }
+
+    result
+}
+
+/// Extension trait that adds pipeline combinators to any
+/// Stream<Item = StageResult<T>>.
+///
+/// Combinators: `.apply()`, `.apply_concurrent()`, `.on_next()`,
+/// `.on_reject()`, `.commit()`.
+///
+/// See `PipelineRunner` for the recommended way to run a pipeline
+/// with rebalance handling.
+pub trait PipelineExt<T: Send>: Stream<Item = StageResult<T>> + Sized {
+    /// Apply a processing stage sequentially to each Emit envelope.
+    /// Equivalent to apply_concurrent(stage, 1).
+    fn apply<'a, S>(self, stage: &'a S) -> impl Stream<Item = StageResult<S::Out>> + 'a
+    where
+        S: Stage<In = T>,
+        Self: 'a,
+        T: 'a,
+    {
+        self.apply_concurrent(stage, 1)
+    }
+
+    /// Apply a processing stage concurrently to up to `concurrency` Emit
+    /// envelopes at once. Results are yielded in input order.
+    /// Non-Emit items pass through immediately.
+    fn apply_concurrent<'a, S>(
+        self,
+        stage: &'a S,
+        concurrency: usize,
+    ) -> impl Stream<Item = StageResult<S::Out>> + 'a
+    where
+        S: Stage<In = T>,
+        Self: 'a,
+        T: 'a,
+    {
+        assert!(concurrency > 0, "concurrency must be at least 1");
+        self.map(move |item| async move {
+            match item {
+                StageResult::Emit(e) => run_stage(stage, e).await,
+                StageResult::Drop { metadata } => StageResult::Drop { metadata },
+                StageResult::Skip => StageResult::Skip,
+                StageResult::Reject {
+                    metadata,
+                    raw,
+                    reason,
+                } => StageResult::Reject {
+                    metadata,
+                    raw,
+                    reason,
+                },
+                StageResult::Fail(err) => StageResult::Fail(err),
+                StageResult::Exit(reason) => StageResult::Exit(reason),
+            }
+        })
+        .buffered(concurrency)
+    }
+
+    /// Call the next handler for each Emit envelope.
+    /// If the handler fails, the item becomes Fail.
+    /// All other variants pass through untouched.
+    fn on_next<'a, H>(self, handler: &'a H) -> impl Stream<Item = StageResult<T>> + 'a
+    where
+        H: NextHandler<T>,
+        Self: 'a,
+        T: 'a,
+    {
+        self.then(move |item| async move {
+            let envelope = match item {
+                StageResult::Emit(e) => e,
+                other => return other,
+            };
+
+            match handler.handle(&envelope).await {
+                Ok(()) => StageResult::Emit(envelope),
+                Err(produce_err) => StageResult::Fail(produce_err),
+            }
+        })
+    }
+
+    /// Call the rejection handler for each Reject item.
+    /// All other variants pass through untouched.
+    /// If the handler itself fails, the item becomes Fail.
+    fn on_reject<'a, H>(self, handler: &'a H) -> impl Stream<Item = StageResult<T>> + 'a
+    where
+        H: RejectionHandler,
+        Self: 'a,
+        T: 'a,
+    {
+        self.then(move |item| async move {
+            match item {
+                StageResult::Reject {
+                    metadata,
+                    raw,
+                    reason,
+                } => {
+                    let rejected = RejectionMetadata {
+                        metadata: metadata.clone(),
+                        raw: raw.clone(),
+                        reason,
+                    };
+
+                    match handler.handle(&rejected).await {
+                        Ok(()) => StageResult::Reject {
+                            metadata,
+                            raw,
+                            reason,
+                        },
+                        Err(handler_err) => StageResult::Fail(handler_err),
+                    }
+                }
+                other => other,
+            }
+        })
+    }
+
+    /// Terminal: drive the pipeline to completion.
+    /// Tracks offsets for Emit, Drop, and Reject items.
+    /// Returns the exit reason (Rebalance, Shutdown, or Complete).
+    /// Fail stops the pipeline with an error.
+    #[allow(async_fn_in_trait)]
+    async fn commit(
+        self,
+        tracker: &mut OffsetTracker<'_>,
+    ) -> Result<PipelineExit, Box<dyn std::error::Error + Send>> {
+        let mut stream = Box::pin(self);
+
+        while let Some(item) = stream.next().await {
+            match item {
+                StageResult::Emit(envelope) => {
+                    tracker.track(envelope.metadata.partition, envelope.metadata.offset + 1);
+                    tracker.record_latency(envelope.metadata.timestamp);
+                }
+                StageResult::Drop { metadata } => {
+                    tracker.track(metadata.partition, metadata.offset + 1);
+                }
+                StageResult::Skip => {}
+                StageResult::Reject { metadata, .. } => {
+                    tracker.track(metadata.partition, metadata.offset + 1);
+                }
+                StageResult::Fail(err) => {
+                    let _ = tracker.flush();
+                    return Err(err);
+                }
+                StageResult::Exit(reason) => {
+                    tracker.flush()?;
+                    return Ok(reason);
+                }
+            }
+
+            tracker.maybe_commit()?;
+        }
+
+        // Stream ended naturally (no Exit item)
+        tracker.flush()?;
+        Ok(PipelineExit::Complete)
+    }
+}
+
+// Blanket impl: any Stream of StageResult<T> gets these methods.
+impl<T: Send, S> PipelineExt<T> for S where S: Stream<Item = StageResult<T>> {}
diff --git a/rust-arroyo/src/processing/stream/handlers/dlq.rs b/rust-arroyo/src/processing/stream/handlers/dlq.rs
new file mode 100644
index 00000000..52956062
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/handlers/dlq.rs
@@ -0,0 +1,61 @@
+use std::sync::Arc;
+
+use crate::backends::kafka::types::{Headers, KafkaPayload};
+use crate::backends::Producer;
+use crate::types::TopicOrPartition;
+
+use super::rejection::{RejectionHandler, RejectionMetadata};
+
+/// A canned RejectionHandler that produces the original message to a DLQ Kafka topic.
+pub struct DlqHandler {
+    producer: Arc<dyn Producer<KafkaPayload>>,
+    topic: TopicOrPartition,
+}
+
+impl DlqHandler {
+    pub fn new(producer: impl Producer<KafkaPayload> + 'static, topic: TopicOrPartition) -> Self {
+        Self {
+            producer: Arc::new(producer),
+            topic,
+        }
+    }
+}
+
+impl RejectionHandler for DlqHandler {
+    async fn handle(
+        &self,
+        rejected: &RejectionMetadata,
+    ) -> Result<(), Box<dyn std::error::Error + Send>> {
+        tracing::error!(
+            "DLQ: {:?}:{} reason={:?}",
+            rejected.metadata.partition,
+            rejected.metadata.offset,
+            rejected.reason,
+        );
+
+        // Preserve original message headers and append partition/offset metadata.
+        let headers = rejected
+            .raw
+            .headers()
+            .cloned()
+            .unwrap_or_else(Headers::new)
+            .insert(
+                "original_partition",
+                Some(rejected.metadata.partition.index.to_string().into_bytes()),
+            )
+            .insert(
+                "original_offset",
+                Some(rejected.metadata.offset.to_string().into_bytes()),
+            );
+
+        let payload = KafkaPayload::new(
+            rejected.raw.key().cloned(),
+            Some(headers),
+            rejected.raw.payload().cloned(),
+        );
+
+        self.producer
+            .produce(&self.topic, payload)
+            .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>)
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/handlers/kafka_producer.rs b/rust-arroyo/src/processing/stream/handlers/kafka_producer.rs
new file mode 100644
index 00000000..59876fae
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/handlers/kafka_producer.rs
@@ -0,0 +1,34 @@
+use std::sync::Arc;
+
+use crate::backends::kafka::types::KafkaPayload;
+use crate::backends::Producer;
+use crate::types::TopicOrPartition;
+
+use super::super::pipeline_envelope::PipelineEnvelope;
+use super::next::NextHandler;
+
+/// A canned NextHandler that produces the envelope's payload to a Kafka topic.
+pub struct KafkaProducerHandler {
+    producer: Arc<dyn Producer<KafkaPayload>>,
+    topic: TopicOrPartition,
+}
+
+impl KafkaProducerHandler {
+    pub fn new(producer: impl Producer<KafkaPayload> + 'static, topic: TopicOrPartition) -> Self {
+        Self {
+            producer: Arc::new(producer),
+            topic,
+        }
+    }
+}
+
+impl NextHandler<KafkaPayload> for KafkaProducerHandler {
+    async fn handle(
+        &self,
+        envelope: &PipelineEnvelope<KafkaPayload>,
+    ) -> Result<(), Box<dyn std::error::Error + Send>> {
+        self.producer
+            .produce(&self.topic, envelope.payload.clone())
+            .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>)
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/handlers/log.rs b/rust-arroyo/src/processing/stream/handlers/log.rs
new file mode 100644
index 00000000..35fb25db
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/handlers/log.rs
@@ -0,0 +1,19 @@
+use super::rejection::{RejectionHandler, RejectionMetadata};
+
+/// A canned RejectionHandler that logs the rejection and continues.
+pub struct LogHandler;
+
+impl RejectionHandler for LogHandler {
+    async fn handle(
+        &self,
+        rejected: &RejectionMetadata,
+    ) -> Result<(), Box<dyn std::error::Error + Send>> {
+        tracing::error!(
+            "Rejected message at {:?}:{} reason={:?}",
+            rejected.metadata.partition,
+            rejected.metadata.offset,
+            rejected.reason,
+        );
+        Ok(())
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/handlers/mod.rs b/rust-arroyo/src/processing/stream/handlers/mod.rs
new file mode 100644
index 00000000..7a2a4847
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/handlers/mod.rs
@@ -0,0 +1,11 @@
+pub mod dlq;
+pub mod kafka_producer;
+pub mod log;
+pub mod next;
+pub mod rejection;
+
+pub use dlq::DlqHandler;
+pub use kafka_producer::KafkaProducerHandler;
+pub use log::LogHandler;
+pub use next::NextHandler;
+pub use rejection::{RejectionHandler, RejectionMetadata};
diff --git a/rust-arroyo/src/processing/stream/handlers/next.rs b/rust-arroyo/src/processing/stream/handlers/next.rs
new file mode 100644
index 00000000..a6d1938c
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/handlers/next.rs
@@ -0,0 +1,12 @@
+use std::future::Future;
+
+use super::super::pipeline_envelope::PipelineEnvelope;
+
+/// Handler for successfully processed messages.
+/// Called by the pipeline's on_next() combinator for each Emit envelope.
+pub trait NextHandler<T>: Send + Sync {
+    fn handle(
+        &self,
+        envelope: &PipelineEnvelope<T>,
+    ) -> impl Future<Output = Result<(), Box<dyn std::error::Error + Send>>> + Send;
+}
diff --git a/rust-arroyo/src/processing/stream/handlers/rejection.rs b/rust-arroyo/src/processing/stream/handlers/rejection.rs
new file mode 100644
index 00000000..7f9df14d
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/handlers/rejection.rs
@@ -0,0 +1,25 @@
+use std::future::Future;
+use std::sync::Arc;
+
+use crate::backends::kafka::types::KafkaPayload;
+
+use super::super::pipeline_envelope::MessageMetadata;
+use super::super::stage::RejectionReason;
+
+/// A rejected message, passed to the RejectionHandler.
+pub struct RejectionMetadata {
+    pub metadata: MessageMetadata,
+    pub raw: Arc<KafkaPayload>,
+    pub reason: RejectionReason,
+}
+
+/// Handler for rejected messages. Called by the pipeline's on_reject() combinator.
+///
+/// Receives a RejectionMetadata with metadata (for logging/headers),
+/// raw payload (for DLQ produce), and the reason for rejection.
+pub trait RejectionHandler: Send + Sync {
+    fn handle(
+        &self,
+        rejected: &RejectionMetadata,
+    ) -> impl Future<Output = Result<(), Box<dyn std::error::Error + Send>>> + Send;
+}
diff --git a/rust-arroyo/src/processing/stream/mod.rs b/rust-arroyo/src/processing/stream/mod.rs
new file mode 100644
index 00000000..41a63837
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/mod.rs
@@ -0,0 +1,17 @@
+mod ext;
+pub mod handlers;
+pub mod offset_tracker;
+mod pipeline_envelope;
+mod pipeline_runner;
+pub mod source;
+mod stage;
+
+pub use ext::PipelineExt;
+pub use handlers::{
+    DlqHandler, KafkaProducerHandler, LogHandler, NextHandler, RejectionHandler, RejectionMetadata,
+};
+pub use offset_tracker::{OffsetCommitter, OffsetTracker};
+pub use pipeline_envelope::{MessageMetadata, PipelineEnvelope};
+pub use pipeline_runner::PipelineRunner;
+pub use source::{KafkaSource, PullSource};
+pub use stage::{PipelineExit, RejectionReason, Stage, StageResult};
diff --git a/rust-arroyo/src/processing/stream/offset_tracker.rs b/rust-arroyo/src/processing/stream/offset_tracker.rs
new file mode 100644
index 00000000..6a995261
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/offset_tracker.rs
@@ -0,0 +1,94 @@
+use std::collections::HashMap;
+use std::time::Duration;
+
+use chrono::{DateTime, Utc};
+
+use crate::timer;
+use crate::types::Partition;
+
+/// Trait for committing offsets. KafkaSource implements this.
+/// Tests can provide a mock.
+pub trait OffsetCommitter: Send + Sync {
+    fn commit_offsets(
+        &self,
+        positions: &HashMap<Partition, u64>,
+    ) -> Result<(), Box<dyn std::error::Error + Send>>;
+}
+
+/// Tracks offsets per partition and commits them on a time-throttled interval.
+pub struct OffsetTracker<'a> {
+    committer: &'a dyn OffsetCommitter,
+    offsets: HashMap<Partition, u64>,
+    last_commit_time: coarsetime::Instant,
+    last_record_time: coarsetime::Instant,
+    commit_frequency: coarsetime::Duration,
+}
+
+/// Ensure the coarsetime background updater is started exactly once.
+/// Multiple OffsetTracker instances share the same updater thread.
+fn ensure_time_updater() {
+    use std::sync::Once;
+    static INIT: Once = Once::new();
+    INIT.call_once(|| {
+        coarsetime::Updater::new(10)
+            .start()
+            .expect("Failed to start coarsetime updater");
+        // Intentionally leaked — runs for the process lifetime.
+    });
+}
+
+impl<'a> OffsetTracker<'a> {
+    pub fn new(commit_frequency: Duration, committer: &'a dyn OffsetCommitter) -> Self {
+        ensure_time_updater();
+        Self {
+            committer,
+            offsets: Default::default(),
+            last_commit_time: coarsetime::Instant::recent(),
+            last_record_time: coarsetime::Instant::recent(),
+            commit_frequency: commit_frequency.into(),
+        }
+    }
+
+    /// Record the offset for a partition.
+    pub fn track(&mut self, partition: Partition, offset: u64) {
+        self.offsets.insert(partition, offset);
+    }
+
+    /// Record a message timestamp for latency tracking.
+    pub fn record_latency(&mut self, timestamp: DateTime<Utc>) {
+        let now = coarsetime::Instant::recent();
+        if now - self.last_record_time > coarsetime::Duration::from_secs(1) {
+            timer!(
+                "arroyo.consumer.latency",
+                (Utc::now() - timestamp).to_std().unwrap_or_default()
+            );
+            self.last_record_time = now;
+        }
+    }
+
+    /// Commit offsets if the commit frequency has elapsed.
+    pub fn maybe_commit(&mut self) -> Result<(), Box<dyn std::error::Error + Send>> {
+        self.try_commit(false)
+    }
+
+    /// Commit all tracked offsets regardless of timing.
+    pub fn flush(&mut self) -> Result<(), Box<dyn std::error::Error + Send>> {
+        self.try_commit(true)
+    }
+
+    fn try_commit(&mut self, force: bool) -> Result<(), Box<dyn std::error::Error + Send>> {
+        if self.offsets.is_empty() {
+            return Ok(());
+        }
+
+        if !force && coarsetime::Instant::recent() - self.last_commit_time <= self.commit_frequency
+        {
+            return Ok(());
+        }
+
+        self.committer.commit_offsets(&self.offsets)?;
+        self.offsets.clear();
+        self.last_commit_time = coarsetime::Instant::recent();
+        Ok(())
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/pipeline_envelope.rs b/rust-arroyo/src/processing/stream/pipeline_envelope.rs
new file mode 100644
index 00000000..9605bc52
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/pipeline_envelope.rs
@@ -0,0 +1,90 @@
+use std::sync::Arc;
+
+use chrono::{DateTime, Utc};
+use rdkafka::message::{BorrowedMessage, Message as RdkafkaMessage};
+
+use crate::backends::kafka::types::KafkaPayload;
+use crate::types::{Partition, Topic};
+
+/// Metadata about the original Kafka message — used for offset tracking
+/// and latency metrics. Separated from the raw payload to avoid conflation.
+#[derive(Debug, Clone)]
+pub struct MessageMetadata {
+    pub partition: Partition,
+    pub offset: u64,
+    pub timestamp: DateTime<Utc>,
+}
+
+/// A message envelope that carries context through a pull-based pipeline.
+///
+/// Three concerns, three fields:
+///   - `payload` — the current transformed data (changes at each stage)
+///   - `metadata` — partition/offset/timestamp for commit tracking
+///   - `raw` — original Kafka bytes for DLQ
+pub struct PipelineEnvelope<T> {
+    pub payload: T,
+    pub metadata: MessageMetadata,
+    pub raw: Arc<KafkaPayload>,
+}
+
+impl<T> PipelineEnvelope<T> {
+    pub fn new(payload: T, metadata: MessageMetadata, raw: Arc<KafkaPayload>) -> Self {
+        Self {
+            payload,
+            metadata,
+            raw,
+        }
+    }
+
+    /// Transform the payload, preserving metadata and raw.
+    pub fn map_payload<U>(self, f: impl FnOnce(T) -> U) -> PipelineEnvelope<U> {
+        PipelineEnvelope {
+            payload: f(self.payload),
+            metadata: self.metadata,
+            raw: self.raw,
+        }
+    }
+
+    /// Transform the payload with a fallible function.
+    pub fn try_map_payload<U, E>(
+        self,
+        f: impl FnOnce(T) -> Result<U, E>,
+    ) -> Result<PipelineEnvelope<U>, E> {
+        Ok(PipelineEnvelope {
+            payload: f(self.payload)?,
+            metadata: self.metadata,
+            raw: self.raw,
+        })
+    }
+}
+
+impl PipelineEnvelope<KafkaPayload> {
+    /// Create an envelope directly from an rdkafka BorrowedMessage.
+    /// Copies key, headers, payload bytes out of rdkafka's internal buffer
+    /// and extracts the broker timestamp.
+    pub fn from_kafka(msg: &BorrowedMessage<'_>) -> Self {
+        let topic = Topic::new(msg.topic());
+        let partition = Partition::new(topic, msg.partition() as u16);
+        let time_millis = msg.timestamp().to_millis().unwrap_or(0);
+        let timestamp =
+            DateTime::from_timestamp_millis(time_millis).unwrap_or(DateTime::<Utc>::MIN_UTC);
+
+        let kafka_payload = KafkaPayload::new(
+            msg.key().map(|k| k.to_vec()),
+            msg.headers().map(|h| h.into()),
+            msg.payload().map(|p| p.to_vec()),
+        );
+
+        let metadata = MessageMetadata {
+            partition,
+            offset: msg.offset() as u64,
+            timestamp,
+        };
+
+        Self {
+            payload: kafka_payload.clone(),
+            metadata,
+            raw: Arc::new(kafka_payload),
+        }
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/pipeline_runner.rs b/rust-arroyo/src/processing/stream/pipeline_runner.rs
new file mode 100644
index 00000000..bfe68710
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/pipeline_runner.rs
@@ -0,0 +1,246 @@
+use std::future::Future;
+use std::pin::Pin;
+
+use futures::stream::Stream;
+
+use crate::backends::kafka::types::KafkaPayload;
+
+use super::offset_tracker::OffsetCommitter;
+use super::source::PullSource;
+use super::stage::{PipelineExit, StageResult};
+
+/// Runs a pipeline in a loop, restarting on rebalance.
+///
+/// Rebalance flow:
+///   1. rdkafka detects partition revocation
+///   2. `KafkaSource`'s `ConsumerContext` fires, ending the stream
+///   3. Stream yields `StageResult::Exit(Rebalance)`
+///   4. `Exit` passes through all combinators to `commit()`
+///   5. `commit()` flushes offsets and returns `Ok(PipelineExit::Rebalance)`
+///   6. `PipelineRunner` calls the `build` closure again with a fresh stream
+///   7. New stream picks up the new partition assignment from rdkafka
+///
+/// The `build` closure is called once per partition assignment.
+/// It receives a fresh stream and committer, builds the pipeline,
+/// and returns the exit reason. Create all stages, handlers, and
+/// trackers inside the closure so they start fresh each assignment.
+///
+/// ```ignore
+/// PipelineRunner::run(&source, |stream, committer| async move {
+///     let stage = MyStage;
+///     let mut tracker = OffsetTracker::new(Duration::from_secs(1), committer);
+///     stream.apply(&stage).commit(&mut tracker).await
+/// }).await?;
+/// ```
+pub struct PipelineRunner;
+
+impl PipelineRunner {
+    pub async fn run<'s, S, F, Fut>(
+        source: &'s S,
+        mut build: F,
+    ) -> Result<(), Box<dyn std::error::Error + Send>>
+    where
+        S: PullSource,
+        F: FnMut(
+            Pin<Box<dyn Stream<Item = StageResult<KafkaPayload>> + 's>>,
+            &'s dyn OffsetCommitter,
+        ) -> Fut,
+        Fut: Future<Output = Result<PipelineExit, Box<dyn std::error::Error + Send>>> + 's,
+    {
+        loop {
+            match build(source.stream(), source.committer()).await? {
+                PipelineExit::Rebalance => {
+                    tracing::info!("Rebalance detected, restarting pipeline");
+                    continue;
+                }
+                PipelineExit::Shutdown | PipelineExit::Complete => return Ok(()),
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::{HashMap, VecDeque};
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::{Arc, Mutex};
+    use std::time::Duration;
+
+    use super::*;
+    use crate::backends::kafka::types::KafkaPayload;
+    use crate::processing::stream::{
+        MessageMetadata, OffsetTracker, PipelineEnvelope, PipelineExt,
+    };
+    use crate::types::{Partition, Topic};
+
+    struct MockCommitter {
+        committed: Mutex<Vec<HashMap<Partition, u64>>>,
+    }
+
+    impl MockCommitter {
+        fn new() -> Self {
+            Self {
+                committed: Mutex::new(Vec::new()),
+            }
+        }
+
+        fn commit_count(&self) -> usize {
+            self.committed.lock().unwrap().len()
+        }
+    }
+
+    impl OffsetCommitter for MockCommitter {
+        fn commit_offsets(
+            &self,
+            positions: &HashMap<Partition, u64>,
+        ) -> Result<(), Box<dyn std::error::Error + Send>> {
+            self.committed.lock().unwrap().push(positions.clone());
+            Ok(())
+        }
+    }
+
+    /// Test source that yields pre-configured batches of messages.
+    /// Each call to `stream()` pops the next batch and exit reason.
+    struct RebalanceTestSource {
+        batches: Mutex<VecDeque<(Vec<StageResult<KafkaPayload>>, PipelineExit)>>,
+        committer: MockCommitter,
+    }
+
+    impl RebalanceTestSource {
+        fn new(batches: Vec<(Vec<StageResult<KafkaPayload>>, PipelineExit)>) -> Self {
+            Self {
+                batches: Mutex::new(VecDeque::from(batches)),
+                committer: MockCommitter::new(),
+            }
+        }
+
+        fn commit_count(&self) -> usize {
+            self.committer.commit_count()
+        }
+    }
+
+    impl PullSource for RebalanceTestSource {
+        fn stream(&self) -> Pin<Box<dyn Stream<Item = StageResult<KafkaPayload>> + '_>> {
+            let (messages, exit) = self
+                .batches
+                .lock()
+                .unwrap()
+                .pop_front()
+                .expect("RebalanceTestSource: no more batches");
+            Box::pin(async_stream::stream! {
+                for msg in messages {
+                    yield msg;
+                }
+                yield StageResult::Exit(exit);
+            })
+        }
+
+        fn committer(&self) -> &dyn OffsetCommitter {
+            &self.committer
+        }
+
+        fn shutdown(&self) {}
+    }
+
+    fn make_message(payload: &[u8], offset: u64) -> StageResult<KafkaPayload> {
+        let kp = KafkaPayload::new(None, None, Some(payload.to_vec()));
+        let md = MessageMetadata {
+            partition: Partition::new(Topic::new("test"), 0),
+            offset,
+            timestamp: chrono::Utc::now(),
+        };
+        StageResult::Emit(PipelineEnvelope::new(kp.clone(), md, Arc::new(kp)))
+    }
+
+    #[tokio::test]
+    async fn test_pipeline_runner_rebalance() {
+        let source = RebalanceTestSource::new(vec![
+            // First assignment: 2 messages, then rebalance
+            (
+                vec![make_message(b"a", 0), make_message(b"b", 1)],
+                PipelineExit::Rebalance,
+            ),
+            // Second assignment: 2 messages, then shutdown
+            (
+                vec![make_message(b"c", 0), make_message(b"d", 1)],
+                PipelineExit::Shutdown,
+            ),
+        ]);
+
+        let call_count = AtomicUsize::new(0);
+
+        let result = PipelineRunner::run(&source, |stream, committer| {
+            call_count.fetch_add(1, Ordering::SeqCst);
+            async move {
+                let mut tracker = OffsetTracker::new(Duration::from_millis(1), committer);
+                stream.commit(&mut tracker).await
+            }
+        })
+        .await;
+
+        assert!(result.is_ok());
+        assert_eq!(
+            call_count.load(Ordering::SeqCst),
+            2,
+            "Closure should be called twice"
+        );
+        assert!(source.commit_count() > 0, "Offsets should be committed");
+    }
+
+    #[tokio::test]
+    async fn test_pipeline_runner_shutdown() {
+        let source = RebalanceTestSource::new(vec![(
+            vec![make_message(b"a", 0), make_message(b"b", 1)],
+            PipelineExit::Shutdown,
+        )]);
+
+        let call_count = AtomicUsize::new(0);
+
+        let result = PipelineRunner::run(&source, |stream, committer| {
+            call_count.fetch_add(1, Ordering::SeqCst);
+            async move {
+                let mut tracker = OffsetTracker::new(Duration::from_millis(1), committer);
+                stream.commit(&mut tracker).await
+            }
+        })
+        .await;
+
+        assert!(result.is_ok());
+        assert_eq!(
+            call_count.load(Ordering::SeqCst),
+            1,
+            "Closure should be called once"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_pipeline_runner_complete() {
+        // Source that naturally ends (no Exit item)
+        struct FiniteSource {
+            committer: MockCommitter,
+        }
+
+        impl PullSource for FiniteSource {
+            fn stream(&self) -> Pin<Box<dyn Stream<Item = StageResult<KafkaPayload>> + '_>> {
+                let messages = vec![make_message(b"a", 0), make_message(b"b", 1)];
+                Box::pin(futures::stream::iter(messages))
+            }
+            fn committer(&self) -> &dyn OffsetCommitter {
+                &self.committer
+            }
+            fn shutdown(&self) {}
+        }
+
+        let source = FiniteSource {
+            committer: MockCommitter::new(),
+        };
+
+        let result = PipelineRunner::run(&source, |stream, committer| async move {
+            let mut tracker = OffsetTracker::new(Duration::from_millis(1), committer);
+            stream.commit(&mut tracker).await
+        })
+        .await;
+
+        assert!(result.is_ok());
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/source.rs b/rust-arroyo/src/processing/stream/source.rs
new file mode 100644
index 00000000..538eb147
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/source.rs
@@ -0,0 +1,198 @@
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use async_stream::stream;
+use futures::stream::Stream;
+use futures::StreamExt;
+use rdkafka::config::ClientConfig as RdKafkaConfig;
+use rdkafka::consumer::{BaseConsumer, CommitMode, Consumer, ConsumerContext, StreamConsumer};
+use rdkafka::types::RDKafkaRespErr;
+use rdkafka::{ClientContext, TopicPartitionList};
+use tokio::sync::Notify;
+use tokio_util::sync::CancellationToken;
+
+use crate::backends::kafka::config::KafkaConfig;
+use crate::backends::kafka::kafka_poll_error_is_recoverable;
+use crate::backends::kafka::types::KafkaPayload;
+use crate::types::{Partition, Topic};
+
+use super::offset_tracker::OffsetCommitter;
+use super::pipeline_envelope::PipelineEnvelope;
+use super::stage::{PipelineExit, StageResult};
+
+/// Trait for pipeline sources. Provides a stream of raw Kafka payloads,
+/// an offset committer, and graceful shutdown.
+///
+/// Object-safe — can be used as `Box<dyn PullSource>` or `Arc<dyn PullSource>`.
+pub trait PullSource: Send + Sync {
+    /// Returns a stream of Kafka messages. Ends on rebalance, shutdown,
+    /// or when the source is exhausted. Callable multiple times — each
+    /// call returns a fresh stream for the current partition assignment.
+    fn stream(&self) -> Pin<Box<dyn Stream<Item = StageResult<KafkaPayload>> + '_>>;
+
+    /// Returns the offset committer for this source.
+    fn committer(&self) -> &dyn OffsetCommitter;
+
+    /// Initiate graceful shutdown. The stream will yield Exit(Shutdown).
+    fn shutdown(&self);
+}
+
+/// ConsumerContext that signals the pipeline on partition revocation.
+///
+/// NOTE: The rebalance callback runs inline during StreamConsumer::poll_next
+/// (same thread/task as the async stream). We cannot block here to drain
+/// the pipeline — that would deadlock. Instead, we just notify and return.
+/// rdkafka proceeds with unassign immediately. The pipeline drains after
+/// unassign, so offset commits are best-effort.
+///
+/// For drain-before-unassign, we would need to switch to BaseConsumer
+/// with manual polling (like the push model does).
+struct PullRebalanceContext {
+    revoke: Arc<Notify>,
+}
+
+impl ClientContext for PullRebalanceContext {}
+
+impl ConsumerContext for PullRebalanceContext {
+    fn rebalance(
+        &self,
+        base_consumer: &BaseConsumer<Self>,
+        err: RDKafkaRespErr,
+        tpl: &mut TopicPartitionList,
+    ) {
+        if err == RDKafkaRespErr::RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS {
+            tracing::info!("Partition revocation detected");
+            self.revoke.notify_one();
+            base_consumer
+                .unassign()
+                .expect("Failed to unassign partitions");
+        } else if err == RDKafkaRespErr::RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS {
+            tracing::info!("Partition assignment received");
+            // TODO: The push model explicitly fetches committed offsets from
+            // the broker and resolves unset offsets via InitialOffset +
+            // watermarks before assigning. We rely on rdkafka's default
+            // behavior (auto.offset.reset). Add explicit resolution if
+            // edge cases arise.
+            base_consumer
+                .assign(tpl)
+                .expect("Failed to assign partitions");
+        }
+    }
+}
+
+/// A Kafka consumer source that produces a Stream of StageResult<KafkaPayload>.
+///
+/// Handles three lifecycle events:
+/// - Messages: yielded as StageResult::Emit
+/// - Rebalance: yields StageResult::Exit(Rebalance), then ends
+/// - Shutdown: yields StageResult::Exit(Shutdown), then ends
+///
+/// The stream can be called again after a rebalance — the underlying
+/// StreamConsumer is reused with the new partition assignment.
+pub struct KafkaSource {
+    consumer: StreamConsumer<PullRebalanceContext>,
+    shutdown: CancellationToken,
+    revoke: Arc<Notify>,
+}
+
+impl KafkaSource {
+    pub fn new(config: KafkaConfig, topics: &[Topic]) -> Self {
+        let revoke = Arc::new(Notify::new());
+        let context = PullRebalanceContext {
+            revoke: revoke.clone(),
+        };
+
+        let mut rdkafka_config: RdKafkaConfig = config.into();
+        let consumer: StreamConsumer<PullRebalanceContext> = rdkafka_config
+            .set_log_level(rdkafka::config::RDKafkaLogLevel::Warning)
+            .create_with_context(context)
+            .expect("Failed to create consumer");
+
+        let topic_strs: Vec<&str> = topics.iter().map(|t| t.as_str()).collect();
+        consumer
+            .subscribe(&topic_strs)
+            .expect("Failed to subscribe");
+
+        Self {
+            consumer,
+            shutdown: CancellationToken::new(),
+            revoke,
+        }
+    }
+}
+
+impl PullSource for KafkaSource {
+    fn stream(&self) -> Pin<Box<dyn Stream<Item = StageResult<KafkaPayload>> + '_>> {
+        let shutdown = self.shutdown.clone();
+        let revoke = &self.revoke;
+
+        Box::pin(stream! {
+            let mut kafka_stream = self.consumer.stream();
+
+            loop {
+                tokio::select! {
+                    msg = kafka_stream.next() => {
+                        match msg {
+                            Some(Ok(m)) => {
+                                yield StageResult::Emit(PipelineEnvelope::from_kafka(&m));
+                            }
+                            Some(Err(e)) if kafka_poll_error_is_recoverable(&e) => {
+                                tracing::warn!("Recoverable Kafka error, skipping: {}", e);
+                                continue;
+                            }
+                            Some(Err(e)) => {
+                                tracing::error!("Fatal Kafka error: {}", e);
+                                yield StageResult::Fail(Box::new(e));
+                                return;
+                            }
+                            None => {
+                                yield StageResult::Exit(PipelineExit::Complete);
+                                return;
+                            }
+                        }
+                    }
+                    _ = shutdown.cancelled() => {
+                        yield StageResult::Exit(PipelineExit::Shutdown);
+                        return;
+                    }
+                    _ = revoke.notified() => {
+                        yield StageResult::Exit(PipelineExit::Rebalance);
+                        return;
+                    }
+                }
+            }
+        })
+    }
+
+    fn committer(&self) -> &dyn OffsetCommitter {
+        self
+    }
+
+    fn shutdown(&self) {
+        self.shutdown.cancel();
+    }
+}
+
+impl OffsetCommitter for KafkaSource {
+    fn commit_offsets(
+        &self,
+        positions: &HashMap<Partition, u64>,
+    ) -> Result<(), Box<dyn std::error::Error + Send>> {
+        let mut tpl = TopicPartitionList::new();
+        for (partition, offset) in positions {
+            tpl.add_partition_offset(
+                partition.topic.as_str(),
+                partition.index as i32,
+                rdkafka::Offset::Offset(*offset as i64),
+            )
+            .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>)?;
+        }
+        // Async commit matches the existing push model's behavior. The broker
+        // may not ack before we clear tracked offsets, but this is acceptable —
+        // worst case on crash is re-processing already-committed messages.
+        self.consumer
+            .commit(&tpl, CommitMode::Async)
+            .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send>)
+    }
+}
diff --git a/rust-arroyo/src/processing/stream/stage.rs b/rust-arroyo/src/processing/stream/stage.rs
new file mode 100644
index 00000000..94414b09
--- /dev/null
+++ b/rust-arroyo/src/processing/stream/stage.rs
@@ -0,0 +1,94 @@
+use std::future::Future;
+use std::sync::Arc;
+
+use crate::backends::kafka::types::KafkaPayload;
+
+use super::pipeline_envelope::{MessageMetadata, PipelineEnvelope};
+
+/// Why the pipeline stream ended.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PipelineExit {
+    /// Partition revocation — caller should recreate stages and restart.
+    Rebalance,
+    /// Graceful shutdown requested (SIGTERM/SIGINT).
+    Shutdown,
+    /// Source stream naturally ended (finite data, test sources).
+    Complete,
+}
+
+/// The result of a Stage processing one envelope.
+pub enum StageResult<T> {
+    /// Produced output — pass downstream.
+    Emit(PipelineEnvelope<T>),
+
+    /// Evaluated and intentionally dropped (filtered).
+    /// Carries metadata so the offset is still tracked.
+    Drop { metadata: MessageMetadata },
+
+    /// Equivalent to no result emission — offset is not propagated.
+    /// Supports accumulating (batching).
+    Skip,
+
+    /// Rejected message — agnostic about reason.
+    /// Carries metadata + raw for offset tracking and DLQ routing.
+    Reject {
+        metadata: MessageMetadata,
+        raw: Arc<KafkaPayload>,
+        reason: RejectionReason,
+    },
+
+    /// Unrecoverable error — kill the pipeline.
+    Fail(Box<dyn std::error::Error + Send>),
+
+    /// Pipeline termination signal from the source.
+    /// Passes through all combinators untouched until reaching commit().
+    Exit(PipelineExit),
+}
+
+impl<T> StageResult<T> {
+    /// Create a Reject result from an envelope, extracting metadata and raw.
+    pub fn reject(envelope: PipelineEnvelope<T>, reason: RejectionReason) -> Self {
+        StageResult::Reject {
+            metadata: envelope.metadata,
+            raw: envelope.raw,
+            reason,
+        }
+    }
+
+    /// Create a Drop result from an envelope, extracting metadata.
+    pub fn drop(envelope: PipelineEnvelope<T>) -> Self {
+        StageResult::Drop {
+            metadata: envelope.metadata,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum RejectionReason {
+    /// Message is malformed or unparseable.
+    Invalid,
+    /// Message is valid but intentionally dropped (e.g. too old, load shedding).
+    Ignored,
+}
+
+/// A processing stage in a pull-based pipeline.
+///
+/// Unified trait for all stage types:
+///   - 1:1 transforms — return Emit(envelope)
+///   - Filters — return Drop(metadata) to filter with offset tracking
+///   - Batching — return Skip while accumulating, Emit when flushing
+///   - Errors — return Reject or Fail
+///
+/// The framework handles error routing (DLQ), metrics, and offset tracking.
+/// The process method is async to support stages that do I/O.
+pub trait Stage: Send + Sync {
+    type In: Send;
+    type Out: Send;
+
+    fn process(
+        &self,
+        envelope: PipelineEnvelope<Self::In>,
+    ) -> impl Future<Output = StageResult<Self::Out>> + Send;
+
+    fn name(&self) -> &str;
+}