diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs
index 55896cd9..08d01e05 100644
--- a/bt-daemon/src/dispatch.rs
+++ b/bt-daemon/src/dispatch.rs
@@ -9,8 +9,8 @@
 
 use crate::delivery_ledger::LedgerSink;
 use crate::journal::JournalWriter;
-use crate::pipeline::plugins::{PluginOutcome, Plugins};
-use crate::pipeline::{Completion, Delivery, Stage};
+use crate::pipeline::routing::RoutingSink;
+use crate::pipeline::{Completion, Delivery};
 use crate::sink::SinkFactory;
 use crate::translate::{Registry, SessionCtx};
 use crate::wire::{Envelope, SessionRoute};
@@ -545,7 +545,7 @@ impl SessionActor {
                 return;
             }
         };
-        let mut sink: Box<dyn crate::sink::Sink> = Box::new(
+        let sink: Box<dyn crate::sink::Sink> = Box::new(
             LedgerSink::new(
                 sink,
                 Some(&self.data_dir),
@@ -556,6 +556,13 @@ impl SessionActor {
             )
             .await,
         );
+        let mut sink: Box<dyn crate::sink::Sink> = Box::new(RoutingSink::spawn(
+            sink,
+            self.source.clone(),
+            self.session_id.clone(),
+            self.data_dir.clone(),
+            self.last_error.clone(),
+        ));
         let mut ctx = SessionCtx {
             session_id: self.session_id.clone(),
             config: Some(self.config.clone()),
@@ -724,54 +731,25 @@ impl SessionActor {
                         self.persist_correlation_if_changed(true).await;
                     }
                 }
-                let plugin_paths = ctx
-                    .config
-                    .as_ref()
-                    .map(|config| config.span_plugins.as_slice())
-                    .unwrap_or_default();
-                let mut stage = Plugins {
-                    paths: plugin_paths.to_vec(),
-                    source: self.source.clone(),
-                    session_id: self.session_id.clone(),
-                };
-                let mut processed = Vec::with_capacity(ops.len());
-                for op in &ops {
-                    match stage.process(op).await {
-                        PluginOutcome::Prepared(op) => processed.push(*op),
-                        PluginOutcome::Filtered => {}
-                        PluginOutcome::Failed(failure) => {
-                            delivered = false;
-                            if failure.newly_seen {
-                                if let Err(error) = crate::plugin_diagnostics::record(
-                                    &self.data_dir,
-                                    &self.source,
-                                    &failure.path,
-                                    &failure.message,
-                                ) {
-                                    tracing::warn!(session_id = %self.session_id,
-                                        "failed to persist span plugin diagnostic: {error}");
-                                }
-                                self.set_error(format!(
-                                    "span plugin {} failed; span operations are being discarded: {}",
-                                    failure.path.display(),
-                                    failure.message
-                                ));
-                            }
-                        }
-                        PluginOutcome::Unavailable(error) => {
-                            delivered = false;
-                            self.set_error(format!("span plugin processor failed: {error}"));
-                        }
-                    }
-                }
-                if !processed.is_empty() {
-                    match sink.emit(&processed).await {
+                if !ops.is_empty() {
+                    match sink.emit(&ops).await {
                         Ok(n) => {
                             self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
                         }
                         Err(e) => {
                             delivered = false;
-                            self.set_error(format!("{emit_error}: {e}"));
+                            if let Some(failure) =
+                                e.downcast_ref::<crate::pipeline::routing::EmitFailure>()
+                            {
+                                self.counters
+                                    .spans_emitted
+                                    .fetch_add(failure.emitted, Ordering::Relaxed);
+                                if failure.report {
+                                    self.set_error(format!("{emit_error}: {e}"));
+                                }
+                            } else {
+                                self.set_error(format!("{emit_error}: {e}"));
+                            }
                         }
                     }
                 }
@@ -964,7 +942,11 @@ impl SessionActor {
         );
         match pending.into_completed().await {
             Ok(through) => *acknowledged_through = through,
-            Err(failed) => self.set_error(format!("delivery checkpoint failed: {}", failed.error)),
+            Err(failed) => self.set_error(format!(
+                "delivery checkpoint at {} failed: {}",
+                failed.delivery.value(),
+                failed.error
+            )),
         }
         true
     }
diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs
index d6cd35cb..da9371cf 100644
--- a/bt-daemon/src/pipeline/delivery.rs
+++ b/bt-daemon/src/pipeline/delivery.rs
@@ -1,3 +1,5 @@
+use tokio::sync::mpsc;
+
 /// A message owns its completion obligation until terminal handling succeeds.
 pub(crate) struct Delivery<T> {
     value: T,
@@ -25,7 +27,6 @@ impl<T> Delivery<T> {
         }
     }
 
-    #[cfg_attr(not(test), allow(dead_code))]
     pub fn value(&self) -> &T {
         &self.value
     }
@@ -49,6 +50,40 @@ impl<T> Delivery<T> {
     }
 }
 
+// async_trait adds a redundant must_use annotation to its boxed futures.
+#[allow(clippy::double_must_use)]
+#[async_trait::async_trait]
+pub(crate) trait InputPort: Send {
+    type Item: Send;
+    async fn receive(&mut self) -> Option<Delivery<Self::Item>>;
+}
+
+// async_trait adds a redundant must_use annotation to its boxed futures.
+#[allow(clippy::double_must_use)]
+#[async_trait::async_trait]
+pub(crate) trait OutputPort: Send {
+    type Item: Send;
+    async fn send(&mut self, delivery: Delivery<Self::Item>) -> Result<(), Delivery<Self::Item>>;
+}
+
+#[async_trait::async_trait]
+impl<T: Send + 'static> InputPort for mpsc::Receiver<Delivery<T>> {
+    type Item = T;
+    async fn receive(&mut self) -> Option<Delivery<T>> {
+        self.recv().await
+    }
+}
+
+#[async_trait::async_trait]
+impl<T: Send + 'static> OutputPort for mpsc::Sender<Delivery<T>> {
+    type Item = T;
+    async fn send(&mut self, delivery: Delivery<T>) -> Result<(), Delivery<T>> {
+        mpsc::Sender::send(self, delivery)
+            .await
+            .map_err(|error| error.0)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs
index 6180279e..3adefa05 100644
--- a/bt-daemon/src/pipeline/mod.rs
+++ b/bt-daemon/src/pipeline/mod.rs
@@ -2,6 +2,7 @@
 
 pub(crate) mod delivery;
 pub(crate) mod plugins;
+pub(crate) mod routing;
 
 pub(crate) use delivery::{Completion, Delivery};
 
diff --git a/bt-daemon/src/pipeline/routing.rs b/bt-daemon/src/pipeline/routing.rs
new file mode 100644
index 00000000..ffb65498
--- /dev/null
+++ b/bt-daemon/src/pipeline/routing.rs
@@ -0,0 +1,279 @@
+//! Routing owns its sink behind a bounded typed port. Session dispatch is its facade.
+use super::delivery::{Delivery, InputPort, OutputPort};
+use super::plugins::{PluginOutcome, Plugins};
+use super::{Completion, Stage};
+use crate::{sink::Sink, translate::SpanOp, wire::SessionConfig};
+use parking_lot::Mutex;
+use std::{path::PathBuf, sync::Arc};
+use tokio::sync::{mpsc, oneshot};
+
+struct Request {
+    config: SessionConfig,
+    versions: (Option<String>, Option<String>),
+    operation: Operation,
+    reply: oneshot::Sender<Response>,
+}
+
+enum Operation {
+    Emit(Vec<SpanOp>),
+    Flush,
+}
+struct Response {
+    result: anyhow::Result<u64>,
+    pending: bool,
+    permalink: Option<String>,
+}
+
+/// Rows accepted before a mandatory plugin failed still count as emitted.
+#[derive(Debug)]
+pub(crate) struct EmitFailure {
+    pub emitted: u64,
+    pub report: bool,
+    pub cause: anyhow::Error,
+}
+impl std::fmt::Display for EmitFailure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.cause.fmt(f)
+    }
+}
+impl std::error::Error for EmitFailure {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.cause.as_ref())
+    }
+}
+
+struct Forward;
+#[async_trait::async_trait]
+impl Completion for Forward {
+    async fn complete(&mut self) -> anyhow::Result<()> {
+        Ok(())
+    }
+}
+
+pub(crate) struct RoutingSink {
+    tx: mpsc::Sender<Delivery<Request>>,
+    config: Option<SessionConfig>,
+    versions: (Option<String>, Option<String>),
+    pending: bool,
+    permalink: Option<String>,
+}
+
+impl RoutingSink {
+    pub fn spawn(
+        mut sink: Box<dyn Sink>,
+        source: String,
+        session_id: String,
+        data_dir: PathBuf,
+        last_error: Arc<Mutex<Option<String>>>,
+    ) -> Self {
+        let (tx, mut rx) = mpsc::channel::<Delivery<Request>>(16);
+        tokio::spawn(async move {
+            while let Some(delivery) = InputPort::receive(&mut rx).await {
+                let request = delivery.value();
+                sink.configure(&request.config);
+                sink.set_capture_versions(
+                    request.versions.0.as_deref(),
+                    request.versions.1.as_deref(),
+                );
+                let result = match &request.operation {
+                    Operation::Emit(ops) => {
+                        let mut stage = Plugins {
+                            paths: request.config.span_plugins.clone(),
+                            source: source.clone(),
+                            session_id: session_id.clone(),
+                        };
+                        let mut prepared = Vec::with_capacity(ops.len());
+                        let mut failure = None;
+                        let mut report = false;
+                        for op in ops {
+                            match stage.process(op).await {
+                                PluginOutcome::Prepared(op) => prepared.push(*op),
+                                PluginOutcome::Filtered => {}
+                                PluginOutcome::Failed(error) => {
+                                    report |= error.newly_seen;
+                                    if error.newly_seen {
+                                        if let Err(diagnostic) = crate::plugin_diagnostics::record(
+                                            &data_dir,
+                                            &source,
+                                            &error.path,
+                                            &error.message,
+                                        ) {
+                                            tracing::warn!(%diagnostic, "plugin diagnostic could not be saved");
+                                        }
+                                    }
+                                    failure = Some(anyhow::anyhow!(
+                                        "span plugin {} failed; span operations are being discarded: {}",
+                                        error.path.display(), error.message));
+                                }
+                                PluginOutcome::Unavailable(error) => {
+                                    report = true;
+                                    failure = Some(error);
+                                }
+                            }
+                        }
+                        let emitted = if prepared.is_empty() {
+                            Ok(0)
+                        } else {
+                            sink.emit(&prepared).await
+                        };
+                        match (emitted, failure) {
+                            (Err(error), _) => Err(error),
+                            (Ok(emitted), Some(cause)) => Err(EmitFailure {
+                                emitted,
+                                report,
+                                cause,
+                            }
+                            .into()),
+                            (Ok(emitted), None) => Ok(emitted),
+                        }
+                    }
+                    Operation::Flush => sink.flush().await.map(|()| 0),
+                };
+                if let Err(error) = &result {
+                    *last_error.lock() = Some(error.to_string());
+                }
+                let request = match delivery.into_completed().await {
+                    Ok(request) => request,
+                    Err(failed) => {
+                        *last_error.lock() =
+                            Some(format!("routing completion failed: {}", failed.error));
+                        // Dropping the unfinished request closes its reply; the
+                        // caller receives an error instead of a success response.
+                        continue;
+                    }
+                };
+                let _ = request.reply.send(Response {
+                    result,
+                    pending: sink.has_pending_delivery(),
+                    permalink: sink.permalink(),
+                });
+            }
+        });
+        Self {
+            tx,
+            config: None,
+            versions: (None, None),
+            pending: false,
+            permalink: None,
+        }
+    }
+
+    async fn request(&mut self, operation: Operation) -> anyhow::Result<u64> {
+        let config = self
+            .config
+            .clone()
+            .ok_or_else(|| anyhow::anyhow!("route is not configured"))?;
+        let (reply, receive) = oneshot::channel();
+        let delivery = Delivery::new(operation, Forward).map(|operation| Request {
+            config,
+            versions: self.versions.clone(),
+            operation,
+            reply,
+        });
+        // Retain ownership on a failed send; no upstream completion is implied.
+        if let Err(undelivered) = OutputPort::send(&mut self.tx, delivery).await {
+            let _ = undelivered.value();
+            anyhow::bail!("routing actor is gone");
+        }
+        let response = receive
+            .await
+            .map_err(|_| anyhow::anyhow!("routing actor dropped its reply"))?;
+        self.pending = response.pending;
+        self.permalink = response.permalink;
+        response.result
+    }
+}
+
+#[async_trait::async_trait]
+impl Sink for RoutingSink {
+    fn configure(&mut self, config: &SessionConfig) {
+        self.config = Some(config.clone());
+    }
+    fn set_capture_versions(&mut self, plugin: Option<&str>, source: Option<&str>) {
+        self.versions = (plugin.map(str::to_owned), source.map(str::to_owned));
+    }
+    async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result<u64> {
+        self.request(Operation::Emit(ops.to_vec())).await
+    }
+    async fn flush(&mut self) -> anyhow::Result<()> {
+        self.request(Operation::Flush).await.map(|_| ())
+    }
+    fn has_pending_delivery(&self) -> bool {
+        self.pending
+    }
+    fn permalink(&self) -> Option<String> {
+        self.permalink.clone()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        translate::SpanRow,
+        wire::{BackendAuth, SessionRoute},
+    };
+    use std::sync::atomic::{AtomicU64, Ordering};
+
+    struct CountingSink(Arc<AtomicU64>);
+    #[async_trait::async_trait]
+    impl Sink for CountingSink {
+        async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result<u64> {
+            self.0.fetch_add(ops.len() as u64, Ordering::Relaxed);
+            Ok(ops.len() as u64)
+        }
+        async fn flush(&mut self) -> anyhow::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn partial_plugin_failure_preserves_count_and_reports_only_once() {
+        let tmp = tempfile::tempdir().unwrap();
+        let plugin = tmp.path().join("selective.mjs");
+        std::fs::write(&plugin, "export default span => { if (span.name === 'reject') throw new Error('rejected'); return span; }").unwrap();
+        let emitted = Arc::new(AtomicU64::new(0));
+        let error = Arc::new(Mutex::new(None));
+        let mut sink = RoutingSink::spawn(
+            Box::new(CountingSink(emitted.clone())),
+            "debug".into(),
+            "partial".into(),
+            tmp.path().into(),
+            error.clone(),
+        );
+        sink.configure(
+            &SessionRoute {
+                span_plugins: vec![plugin],
+                ..SessionRoute::default()
+            }
+            .with_auth(BackendAuth {
+                token: "test".into(),
+                api_url: None,
+                app_url: None,
+                org_name: None,
+                org_id: None,
+            }),
+        );
+        let op = |name: &str| {
+            SpanOp::Insert(SpanRow {
+                span_id: name.into(),
+                name: name.into(),
+                ..SpanRow::default()
+            })
+        };
+        let failure = sink.emit(&[op("accept"), op("reject")]).await.unwrap_err();
+        let first = failure.downcast_ref::<EmitFailure>().unwrap();
+        assert_eq!(first.emitted, 1);
+        assert!(first.report);
+        assert_eq!(emitted.load(Ordering::Relaxed), 1);
+        let failure = sink.emit(&[op("accept")]).await.unwrap_err();
+        let cached = failure.downcast_ref::<EmitFailure>().unwrap();
+        assert_eq!(cached.emitted, 0);
+        assert!(
+            !cached.report,
+            "cached failures must not request another warning"
+        );
+        assert!(error.lock().as_ref().unwrap().contains("rejected"));
+        assert_eq!(emitted.load(Ordering::Relaxed), 1);
+    }
+}
diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs
index 06f81e45..c284d3d8 100644
--- a/bt-daemon/tests/pipeline.rs
+++ b/bt-daemon/tests/pipeline.rs
@@ -3292,3 +3292,39 @@ async fn packaged_grok_hook_replays_bounded_transcripts_to_isolated_debug_routes
     shutdown(&socket).await;
     second.await.unwrap();
 }
+
+#[tokio::test]
+async fn status_counts_rows_emitted_before_a_plugin_failure() {
+    let (data_dir, socket, handle, tmp) = start_daemon().await;
+    let plugin = tmp.path().join("partial.mjs");
+    std::fs::write(&plugin, "export default span => { if (span.name === 'SessionStart') throw new Error('rejected event'); return span; }").unwrap();
+    let mut env = envelope("partial-plugin", "SessionStart", 1);
+    env.route.as_mut().unwrap().span_plugins = vec![plugin];
+    forward_envelope(&env, &socket, &dummy_host(), false)
+        .await
+        .unwrap();
+    flush_session("partial-plugin", &socket, 5_000)
+        .await
+        .unwrap();
+    let status = run_status(StatusArgs {
+        socket: Some(socket.clone()),
+        session_id: Some("partial-plugin".into()),
+    })
+    .await
+    .unwrap()
+    .unwrap();
+    let rows = std::fs::read_to_string(data_dir.join("spans/partial-plugin.ndjson")).unwrap();
+    assert_eq!(
+        rows.lines().count(),
+        1,
+        "the root was emitted before the plugin rejected its event"
+    );
+    assert_eq!(status.sessions[0].spans_emitted, 1);
+    assert!(status.sessions[0]
+        .last_error
+        .as_ref()
+        .unwrap()
+        .contains("rejected event"));
+    shutdown(&socket).await;
+    handle.await.unwrap();
+}