Introduce pipeline messages and checked routing acceptance

bt-daemon/Cargo.toml

@@ -22,7 +22,7 @@ expect_used = "deny"
 
 [dependencies]
 # Keep this pinned exactly, matching the dependency policy used by `bt`.
-braintrust-sdk-rust = { git = "https://github.com/braintrustdata/braintrust-sdk-rust", rev = "e85ab580f52174efa9f90b9069a5fb2f5460a20d" }
+braintrust-sdk-rust = { git = "https://github.com/braintrustdata/braintrust-sdk-rust", rev = "f2216fa0ceee132e02f1392c0e5af89ae762e9c8" }
 base64 = "0.22"
 anyhow = "1"
 parking_lot = "0.12"

bt-daemon/src/dispatch.rs

@@ -9,6 +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::sink::SinkFactory;
 use crate::translate::{Registry, SessionCtx};
 use crate::wire::{Envelope, SessionRoute};
@@ -629,12 +631,13 @@ impl SessionActor {
                         .checkpoint_and_flush(&mut translator, &mut sink, &ctx)
                         .await;
                     if checkpointable {
-                        self.record_delivery_checkpoint(
-                            &mut sink,
-                            pending_through,
-                            &mut acknowledged_through,
-                        )
-                        .await;
+                        checkpointable &= self
+                            .record_delivery_checkpoint(
+                                &mut sink,
+                                pending_through,
+                                &mut acknowledged_through,
+                            )
+                            .await;
                     }
                     let _ = reply.send(self.counters.queued.load(Ordering::Relaxed));
                 }
@@ -648,12 +651,13 @@ impl SessionActor {
                         )
                         .await;
                     if checkpointable {
-                        self.record_delivery_checkpoint(
-                            &mut sink,
-                            pending_through,
-                            &mut acknowledged_through,
-                        )
-                        .await;
+                        checkpointable &= self
+                            .record_delivery_checkpoint(
+                                &mut sink,
+                                pending_through,
+                                &mut acknowledged_through,
+                            )
+                            .await;
                     }
                     let _ = reply.send(self.counters.queued.load(Ordering::Relaxed));
                 }
@@ -725,58 +729,38 @@ impl SessionActor {
                     .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 crate::span_processor::process(
-                        plugin_paths,
-                        op,
-                        &self.source,
-                        &self.session_id,
-                    ) {
-                        Ok(result) => {
-                            if let Some(failure) = result.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
-                                    ));
-                                }
-                            }
-                            if let Some(op) = result.op {
-                                processed.push(op);
-                            }
-                        }
-                        Err(error) => {
+                    match stage.process(op).await {
+                        PluginOutcome::Prepared(op) => processed.push(*op),
+                        PluginOutcome::Filtered => {}
+                        PluginOutcome::Failed(failure) => {
                             delivered = false;
-                            if let Some(plugin) = plugin_paths.first() {
-                                if let Err(diagnostic_error) = crate::plugin_diagnostics::record(
+                            if failure.newly_seen {
+                                if let Err(error) = crate::plugin_diagnostics::record(
                                     &self.data_dir,
                                     &self.source,
-                                    plugin,
-                                    &error.to_string(),
+                                    &failure.path,
+                                    &failure.message,
                                 ) {
-                                    tracing::warn!(
-                                        session_id = %self.session_id,
-                                        "failed to persist span plugin diagnostic: {diagnostic_error}"
-                                    );
+                                    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
+                                ));
                             }
-                            self.set_error(format!(
-                                "span plugin processor failed; span operation discarded: {error}"
-                            ));
+                        }
+                        PluginOutcome::Unavailable(error) => {
+                            delivered = false;
+                            self.set_error(format!("span plugin processor failed: {error}"));
                         }
                     }
                 }
@@ -953,38 +937,36 @@ impl SessionActor {
         }
     }
 
-    async fn flush_sink(&self, sink: &mut Box<dyn crate::sink::Sink>) -> bool {
-        if let Err(e) = sink.flush().await {
-            self.set_error(format!("sink flush failed: {e}"));
-            return false;
-        }
-        self.refresh_permalink(sink.as_ref());
-        !sink.has_pending_delivery()
-    }
-
+    /// False means delivery was rejected and this actor must never advance its
+    /// contiguous event boundary. Deferred delivery and failed checkpoint writes
+    /// leave accepted work eligible for a later checkpoint.
     async fn record_delivery_checkpoint(
         &self,
         sink: &mut Box<dyn crate::sink::Sink>,
         through: u64,
         acknowledged_through: &mut u64,
-    ) {
-        if !self.flush_sink(sink).await {
-            return;
+    ) -> bool {
+        if let Err(error) = sink.flush().await {
+            self.set_error(format!("sink flush failed: {error}"));
+            return false;
         }
-        if through <= *acknowledged_through {
-            return;
+        self.refresh_permalink(sink.as_ref());
+        if sink.has_pending_delivery() || through <= *acknowledged_through {
+            return true;
         }
-        if let Err(error) = self
-            .journal
-            .lock()
-            .await
-            .append_delivery_checkpoint(&self.route, through)
-            .await
-        {
-            self.set_error(format!("delivery checkpoint failed: {error}"));
-        } else {
-            *acknowledged_through = through;
+        let pending = Delivery::new(
+            through,
+            EventCompletion {
+                journal: self.journal.clone(),
+                route: self.route.clone(),
+                through,
+            },
+        );
+        match pending.into_completed().await {
+            Ok(through) => *acknowledged_through = through,
+            Err(failed) => self.set_error(format!("delivery checkpoint failed: {}", failed.error)),
         }
+        true
     }
 
     fn refresh_permalink(&self, sink: &dyn crate::sink::Sink) {
@@ -1015,6 +997,23 @@ pub(crate) fn is_tool_lifecycle_event(event: &str) -> bool {
     )
 }
 
+struct EventCompletion {
+    journal: Arc<tokio::sync::Mutex<JournalWriter>>,
+    route: SessionRoute,
+    through: u64,
+}
+
+#[async_trait::async_trait]
+impl Completion for EventCompletion {
+    async fn complete(&mut self) -> anyhow::Result<()> {
+        self.journal
+            .lock()
+            .await
+            .append_delivery_checkpoint(&self.route, self.through)
+            .await
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1151,16 +1150,8 @@ mod tests {
             .await
             .unwrap();
         let updates = session.join("updates.jsonl");
-        tokio::fs::write(&updates, b"{\"update\":1}\n")
-            .await
-            .unwrap();
-        let writer_path = updates.clone();
-        let writer = tokio::spawn(async move {
-            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
-            tokio::fs::write(&writer_path, b"{\"update\":1}\n{\"update\":2}\n")
-                .await
-                .unwrap();
-        });
+        let initial = b"{\"update\":1}\n";
+        tokio::fs::write(&updates, initial).await.unwrap();
         let mut env = Envelope {
             source: "grok".into(),
             source_version: None,
@@ -1177,8 +1168,31 @@ mod tests {
             config: None,
         };
 
-        hydrate_transcript_reference(tmp.path(), &mut env).await;
-        writer.await.unwrap();
+        let mirror = crate::transcript_mirror::mirror_path(
+            tmp.path(),
+            &crate::ids::session_namespace("grok", "session-1"),
+            updates.to_str().unwrap(),
+        );
+        {
+            let hydration = hydrate_transcript_reference(tmp.path(), &mut env);
+            tokio::pin!(hydration);
+            tokio::select! {
+                biased;
+                () = async {
+                    // Append after the first copy, before polling hydration again.
+                    // A delayed task can miss the entire capture window on busy CI.
+                    while !std::fs::metadata(&mirror).is_ok_and(|meta| meta.len() == initial.len() as u64) {
+                        tokio::task::yield_now().await;
+                    }
+                    std::fs::write(&updates, b"{\"update\":1}\n{\"update\":2}\n").unwrap();
+                } => {},
+                () = &mut hydration => panic!("terminal hydration did not wait for another capture"),
+                _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {
+                    panic!("initial transcript capture did not complete");
+                },
+            }
+            hydration.await;
+        }
 
         let update_mirror = &env.payload["_bt_grok_transcript_mirrors"]["updates"];
         assert_eq!(

bt-daemon/src/lib.rs

@@ -24,6 +24,7 @@ mod hook;
 mod ids;
 mod journal;
 mod managed_run;
+mod pipeline;
 mod plugin_diagnostics;
 pub(crate) mod process;
 mod route;