From 41d4f21240c371480dc563eae72054581c3f970a Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Fri, 16 Oct 2026 22:16:51 +0800 Subject: [PATCH 1/9] Isolate doctor tests from installed agent settings --- bt-daemon/src/trace_runtime.rs | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/bt-daemon/src/trace_runtime.rs b/bt-daemon/src/trace_runtime.rs index 2e1ad6f0..114f1614 100644 --- a/bt-daemon/src/trace_runtime.rs +++ b/bt-daemon/src/trace_runtime.rs @@ -17,7 +17,7 @@ use crate::{ }; use async_trait::async_trait; use std::ffi::OsString; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; @@ -489,18 +489,26 @@ fn daemon_warnings( } async fn doctor_output(host: &TraceHostContext, args: DoctorArgs) -> DoctorCommandOutput { - doctor_output_at(host, args, &paths::socket_path(None)).await + let source = args.agent.identity().id; + doctor_output_at( + host, + args, + &paths::socket_path(None), + paths::agent_settings_path(source, None), + crate::settings::AgentSettings::load(source), + ) + .await } async fn doctor_output_at( host: &TraceHostContext, args: DoctorArgs, socket: &Path, + settings_path: PathBuf, + settings: crate::settings::AgentSettings, ) -> DoctorCommandOutput { let source = args.agent.identity().id; - let settings_path = paths::agent_settings_path(source, None); let settings_present = settings_path.exists(); - let settings = crate::settings::AgentSettings::load(source); let enabled = settings.tracing_enabled(); let mut warnings = Vec::new(); @@ -1304,7 +1312,16 @@ mod tests { agent: DoctorAgent, ) -> DoctorCommandOutput { let host = test_host(Arc::new(shell)); - doctor_output_at(&host, DoctorArgs { agent }, &self.socket).await + // Use fixture settings rather than the developer's installed route + // or a managed run's invocation settings. + doctor_output_at( + &host, + DoctorArgs { agent }, + &self.socket, + self._dir.path().join("braintrust.json"), + crate::settings::AgentSettings::default(), + ) + .await } async fn forward_event(&self, source: &str, session_id: &str, route: SessionRoute) { @@ -1512,6 +1529,8 @@ mod tests { agent: DoctorAgent::Codex, }, &socket, + temp.path().join("braintrust.json"), + crate::settings::AgentSettings::default(), ) .await; @@ -1535,6 +1554,8 @@ mod tests { agent: DoctorAgent::Codex, }, &test_endpoint(temp.path()), + temp.path().join("braintrust.json"), + crate::settings::AgentSettings::default(), ) .await; assert_eq!(output.daemon.status, DaemonStatus::NotRunning); From b2c24beceb342e06ddbf8af0a3b75223c6f99e2b Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 18:42:16 +0800 Subject: [PATCH 2/9] Synchronize Grok transcript capture regression test --- bt-daemon/src/dispatch.rs | 39 +++++++++++++++++++++++++++------------ 1 file changed, 27 insertions(+), 12 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 61fc8a5a..97de96fe 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -1151,16 +1151,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 +1169,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!( From 7667cded543a3a83002c815b6b5164ca64d1207a Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 13 Oct 2026 23:12:55 +0800 Subject: [PATCH 3/9] Introduce completion-preserving messages and checked routing acceptance --- bt-daemon/Cargo.lock | 2 +- bt-daemon/Cargo.toml | 2 +- bt-daemon/docs/recoverable-pipeline-plan.md | 251 ++++++++++++++++++++ bt-daemon/src/dispatch.rs | 108 ++++----- bt-daemon/src/lib.rs | 1 + bt-daemon/src/pipeline/delivery.rs | 153 ++++++++++++ bt-daemon/src/pipeline/mod.rs | 14 ++ bt-daemon/src/pipeline/plugins.rs | 34 +++ bt-daemon/src/sink/braintrust.rs | 25 +- bt-daemon/tests/braintrust_sink.rs | 88 +++++++ 10 files changed, 607 insertions(+), 71 deletions(-) create mode 100644 bt-daemon/docs/recoverable-pipeline-plan.md create mode 100644 bt-daemon/src/pipeline/delivery.rs create mode 100644 bt-daemon/src/pipeline/mod.rs create mode 100644 bt-daemon/src/pipeline/plugins.rs diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index 05fbde0f..84cf1efe 100644 --- a/bt-daemon/Cargo.lock +++ b/bt-daemon/Cargo.lock @@ -239,7 +239,7 @@ dependencies = [ [[package]] name = "braintrust-sdk-rust" version = "0.1.0-alpha.2" -source = "git+https://github.com/braintrustdata/braintrust-sdk-rust?rev=e85ab580f52174efa9f90b9069a5fb2f5460a20d#e85ab580f52174efa9f90b9069a5fb2f5460a20d" +source = "git+https://github.com/braintrustdata/braintrust-sdk-rust?rev=74e568dd30eb1654039dcc792f3327a8399deaba#74e568dd30eb1654039dcc792f3327a8399deaba" dependencies = [ "anyhow", "arc-swap", diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index 6d8ed9a5..94b6b979 100644 --- a/bt-daemon/Cargo.toml +++ b/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 = "74e568dd30eb1654039dcc792f3327a8399deaba" } base64 = "0.22" anyhow = "1" parking_lot = "0.12" diff --git a/bt-daemon/docs/recoverable-pipeline-plan.md b/bt-daemon/docs/recoverable-pipeline-plan.md new file mode 100644 index 00000000..a120b5c7 --- /dev/null +++ b/bt-daemon/docs/recoverable-pipeline-plan.md @@ -0,0 +1,251 @@ +# Recoverable pipeline implementation plan + +Status: proposal. Supersedes the recoverable-stream pipeline design and implementation +plan and the event WAL recovery proposal. Start from current `main`, not PRs #130, +#132, or #171. This plan authorizes no implementation by itself. + +## 1. Goal and scope + +Build a concurrent pipeline of self-contained actors with typed input/output ports: + +```text +capture -> event journal -> translation -> SpanOp journal + | | + head A head B + | | + route A route B + | | + sink A sink B +``` + +The requirements are failure reasons with retry criteria, persisted SpanOps, and +isolated stages composed through channels. Each route owns an independent resumable +head in the shared SpanOp journal. Downstream recovery starts at its earliest +unresolved SpanOp; it never replays translation. + +Preserve trace shape, routing identity, attachment, imports, and existing ledger +semantics. Per-plugin durable checkpoints, translator snapshots, historical backfill, +prefix collection, and a new scheduler or workflow platform are outside scope. + +## 2. Composition contract + +Actors own mutable processing state. Stages accept typed inputs and return typed +outcome enums; assembly maps each variant to a compatible downstream input port. +Stages do not name successors, access another actor's state, or receive `Daemon`. +A small shared stage trait with associated `Input` and `Outcome` types and typed +input/output port traits is sufficient. Wiring must cover every outcome variant +and check connected types. Ordinary enum matches and adapters are sufficient. + +For example, a plugin returns `Prepared`, `Filtered`, or `Failed`. Assembly routes +those outcomes to submission, terminal completion, or recovery. The actor runner +preserves the message envelope across that mapping. Adding a validation stage +changes wiring rather than the plugin implementation. + +Both bounded Tokio channels and journal-backed channels expose send/receive ports +carrying the same envelope. Normal stages need not know which transport they use. +Storage adapters own durability, subscriptions, and completion; a successful volatile +send means queued, not consumed. A failed send retains or returns the message. + +### Message ownership and completion + +```rust +// Illustrative API; fields and completion plumbing are private. +struct Delivery { + value: T, + completion: Completion, +} + +impl Delivery { + fn value(&self) -> &T; + fn map(self, f: impl FnOnce(T) -> U) -> Delivery; + async fn into_completed(self) -> Result>; +} +``` + +`value()` borrows for processing. `map()` changes the value type while moving its +completion obligation. `into_completed()` claims the value has been handled, +confirms upstream completion, and returns ownership of the final value. The sink +calls it after checked acceptance; an intentional filter can also complete work. +If completion fails, `Unfinished` retains the value and context for retry. + +Volatile handoffs preserve the obligation. A no-op completion is valid only when +there is no upstream obligation. Dropping a message, cancelling a task, or releasing +an `Arc` does not complete it. An immutable payload may be shared with `Arc`, but +each route's completion capability is unique and non-cloneable. + +A bounded owner mailbox and one-shot reply can implement completion. The reply +confirms the durable head write, not merely receipt of the request. Completion +requests are idempotent and cannot skip unresolved positions; late or duplicate +requests cannot advance a different subscription or overwrite newer progress. +Only one live consumer owns a given head. Restart reconstructs envelopes from +persisted positions, not from channels or reference counts. + +For one input producing several outputs, keep a bounded batch under one obligation; +never clone the token onto each output. Terminal completion covers the whole batch. +Translation may produce no output: its empty commit still consumes the input. +General-purpose splitting/joining of obligations is unnecessary for this plan. + +### Concurrency and ordering + +The graph is shared; current stage and recovery state belong to pending executions, +not the pipeline globally. Use one ordered translation consumer per source/session +and one ordered routing consumer per route head. Different consumers run concurrently; +initially each processes one item or bounded batch at a time. Stateful translators +and route state belong to their consumers, not a global registered handler instance. + +Use ordinary Tokio tasks and bounded queues; retain worker-local plugin runtimes. +Small synchronous stages may share an actor. A recovering consumer stops later data +on its head while still servicing configuration, status, and shutdown controls. +Journal-backed edges let upstream work continue persisting while consumers wait. +Notifications are wake hints; subscribe/recheck against storage to avoid lost wakes. + +## 3. Durable channels + +Keep the event journal. Use a concrete SQLite adapter for SpanOps, heads, and recovery +records, with blocking database work outside Tokio workers. Keep SQL and codecs +inside storage. Capture acknowledges only its durable event append, independently +of downstream progress. Storage read/corruption errors hold work and report failure; +they never silently skip retained records. + +### Handoff and restart + +The SpanOp adapter atomically commits ordered output and its consumed event position. +That position is the authoritative translation boundary. Stable commit identities +make retry idempotent, including empty output. Support bounded output chunks without +exposing partial commits to readers or collecting a whole transcript into memory. + +At this durable handoff, the adapter resolves the preceding event obligation and +creates new obligations for downstream subscriptions. If output committed but +upstream confirmation was interrupted, reconcile the committed input position; +do not regenerate or append that output. Keep candidate output while retrying a +failed local commit before reading more input. + +Existing translators rebuild by replaying the committed event prefix with output +suppressed, then process the uncommitted suffix. Retain that prefix and required +transcript references. This rebuild is for translation state, never routing retry. +Output-producing checkpoint/finalize controls need stable ordered event identities +and captured transcript boundaries so restart reproduces the same input. + +Persist every field needed by routing, including creation provenance and operation +identity. Current `SpanRow` serialization skips fields such as `late_merge_key`; +use a storage codec that preserves them. Persist no credentials or auth leases. +Committed SpanOps are route-neutral; route preparation applies attachment, tags, +metadata, and plugin configuration to a private value. Version the storage codec +and hold unsupported records with a diagnostic rather than defaulting or skipping fields. + +### Subscription split and completion + +The producer sends one output batch without knowing route counts or heads. Assembly +registers subscriptions and pairs their receivers with routing actors. The durable +adapter supplies independent deliveries to each route; it need not eagerly push +copies or add a separate fan-out stage. + +Persist the nonsecret subscription identity/settings needed to resume without a +new event. Preserve existing route identity rules: credential refresh keeps its +head, and configuration changes must not redirect an old head to another destination. + +Each head advances monotonically over a contiguous consumed prefix. Reading and +forwarding never advance it. Sink enqueue is insufficient: require checked flush/ +acceptance for every covered operation, including when SDK clients are shared. +Preserve terminal-span and late-merge ledger behavior without recording acceptance +prematurely. + +Deferred roots and batching may read ahead while retaining the earliest unfinished +position. They must allow reading the child that releases a deferred root; later +success cannot skip the unresolved prefix. Keep pending memory bounded and preserve +existing deferral behavior. + +After backend acceptance, retry a failed local head write without rerunning delivery +in process. If confirmation is lost, reload the durable head. A crash before durable +completion may require idempotent resubmission with existing stable span IDs; the +pipeline provides at-least-once delivery, not exactly-once external effects. + +Retirement preserves subscriptions, heads, and pending work. Retain completed SpanOps +under the existing session retention policy so late routes can read history without +backfill. Delete a session only when retention permits, translation is caught up, +all heads are complete, and nothing is recovering. Protect event/transcript references +needed for reconstruction. Empty subscriber sets do not authorize accidental deletion. + +## 4. Recovery as a pipeline state + +A failure outcome carries a typed cause, local diagnostic, and retry condition. +Wiring maps it to `Recovering`, which retains the pending position, stage, boundary, +and condition evidence. It waits asynchronously, then returns an eligible outcome +mapped back into processing with updated conditions. No external supervisor restarts +stage clusters. Successful processing and completion, not eligibility alone, resolve +the failure. + +| Cause | Retry criterion | +| --- | --- | +| Plugin | Failed file's bytes/existence or relevant configured chain changes | +| Translation | Applicable translator revision changes | +| Credentials | Credential notification or bounded fresh resolution | +| Transport/server | Capped retry delay | +| Rate limit | Retry-After or capped fallback delay | +| Permission/destination | Relevant configuration change or bounded recheck | +| Permanent rejection | Relevant processor/configuration change | +| Local persistence | Bounded retry of the failed storage operation | + +Record exact plugin identity/digest and nonsecret configuration evidence. Unrelated +file edits do not qualify. Classify typed errors at their source rather than parsing +error strings. Restart re-evaluates persisted criteria; it is not itself permission +to retry unchanged plugins. Reload changed plugins or fresh credentials on retry. + +Routing retries reload unresolved SpanOps and may repeat downstream transformations. +Translation failures use the event boundary. Persistence failures retry the local +commit/completion where possible. Persist recovery evidence and expose stage, pending +position, reason, and eligibility through doctor. A failure to persist diagnostics +must retain the original reason in memory and never release work. + +## 5. Incremental PRs + +Each PR has one reviewable responsibility, includes its checks, and leaves `main` +usable. Small preparatory extractions are acceptable; no line-count target applies. + +1. **Contracts inside the current actor.** Extract stage handlers, typed outcomes, + `Delivery`, and queue port adapters without changing task ownership or replay. + Establish checked sink acceptance and `into_completed()`. Test mapping, filtering, + no completion on drop, and rejected delivery retaining its checkpoint. +2. **Actor/channel separation.** Split translation and routing ownership within each + existing route, using bounded ports and the existing session handle as a facade. + Carry event identity through completion. Test ordering, backpressure, cancellation, + and controls during blocked sends. Preserve current journals/checkpoints. +3. **Route-neutral output.** Move attachment/tags/metadata into route preparation and + carry provenance with SpanOps. Remove translator route-config reads and sink version + setters. Test canonical output across routes and equivalent final trace behavior. +4. **Durable adapter.** Add SQLite commits, full SpanOp codec, heads, and subscription + envelopes behind the port contract. Test empty/multiple output, chunked commit, + interruption before/after commit, independent heads, and confirmed completion. + Keep production wiring unchanged until these contracts pass. +5. **Activate the journal boundary.** Replace the volatile edge with the durable port; + share translation per source/session and register a subscription per route. Include + ordered controls and necessary lifecycle wiring. Migrate released checkpoints + restartably; preserve data when a completion mapping cannot be proved. Test restart, + late routes, zero translator calls during routing resume, and healthy sibling progress. +6. **Recovery transitions.** Persist reasons/criteria, wire waiting and retry outcomes, + and restore pending work without new native events. Test exact plugin edits, + credential refresh, transport retries, unchanged-condition holds, and pending retention. + Plugin/auth/delivery retries must never enter translation. +7. **Lifecycle consolidation and cleanup.** Remove replaced per-route translation and + active event delivery checkpoints; retain upgrade readers. Validate flush/finalize, + managed runs, startup, retirement, shutdown, correlation, and import/live ledger + compatibility. Existing import entry points remain unchanged. + +Activation includes required lifecycle correctness; the last PR consolidates rather +than repairs an unshippable intermediate state. Flush targets ordered controls and +reports completion only when the covered route heads reach their output boundary. +Timeout or shutdown leaves unresolved work durable. Correlation progress follows +translation, not downstream delivery; adjust existing barriers narrowly to preserve +independence without redesigning correlation. + +## 6. Validation + +Test handlers separately from actor wiring. Fault-inject commit and head-write gaps, +including lost confirmation, completion cancellation, and actor failure between +volatile stages. Prove a blocked route does not block capture, translation, or its +sibling, and blocked translation leaves committed SpanOps deliverable. + +Use the actual SDK path to establish rejection/acceptance semantics, alongside fake +sinks for pipeline tests. Run targeted checks for each PR and the locked all-feature +daemon suite, formatting, and CI lints before presenting it. Persistence changes +need completed Linux, macOS, and Windows CI. Report local and remote results separately. diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 97de96fe..37a6ed4e 100644 --- a/bt-daemon/src/dispatch.rs +++ b/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}; @@ -725,58 +727,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: {}", + 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}")); } } } @@ -974,16 +956,17 @@ impl SessionActor { if through <= *acknowledged_through { return; } - 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)), } } @@ -1015,6 +998,23 @@ pub(crate) fn is_tool_lifecycle_event(event: &str) -> bool { ) } +struct EventCompletion { + journal: Arc>, + 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::*; diff --git a/bt-daemon/src/lib.rs b/bt-daemon/src/lib.rs index 2df4e679..61a3f58f 100644 --- a/bt-daemon/src/lib.rs +++ b/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; diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs new file mode 100644 index 00000000..2d4b7464 --- /dev/null +++ b/bt-daemon/src/pipeline/delivery.rs @@ -0,0 +1,153 @@ +// Queue port consumers are introduced in the next stack layer. +#![allow(dead_code)] + +use tokio::sync::mpsc; + +/// A message owns its completion obligation until terminal handling succeeds. +pub(crate) struct Delivery { + value: T, + completion: Box, +} + +#[async_trait::async_trait] +pub(crate) trait Completion: Send { + async fn complete(&mut self) -> anyhow::Result<()>; +} + +pub(crate) struct Unfinished { + pub delivery: Delivery, + pub error: anyhow::Error, +} + +impl Delivery { + pub fn new(value: T, completion: impl Completion + 'static) -> Self { + Self { + value, + completion: Box::new(completion), + } + } + + pub fn value(&self) -> &T { + &self.value + } + + pub fn map(self, map: impl FnOnce(T) -> U) -> Delivery { + Delivery { + value: map(self.value), + completion: self.completion, + } + } + + pub async fn into_completed(mut self) -> Result> { + match self.completion.complete().await { + Ok(()) => Ok(self.value), + Err(error) => Err(Unfinished { + delivery: self, + error, + }), + } + } +} + +#[async_trait::async_trait] +pub(crate) trait InputPort: Send { + type Item: Send; + async fn receive(&mut self) -> Option>; +} + +#[async_trait::async_trait] +pub(crate) trait OutputPort: Send { + type Item: Send; + async fn send(&mut self, delivery: Delivery) -> Result<(), Delivery>; +} + +#[async_trait::async_trait] +impl InputPort for mpsc::Receiver> { + type Item = T; + async fn receive(&mut self) -> Option> { + self.recv().await + } +} + +#[async_trait::async_trait] +impl OutputPort for mpsc::Sender> { + type Item = T; + async fn send(&mut self, delivery: Delivery) -> Result<(), Delivery> { + mpsc::Sender::send(self, delivery) + .await + .map_err(|error| error.0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + + struct Count(Arc); + #[async_trait::async_trait] + impl Completion for Count { + async fn complete(&mut self) -> anyhow::Result<()> { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + } + + #[tokio::test] + async fn mapping_and_channel_forwarding_preserve_completion() { + let count = Arc::new(AtomicUsize::new(0)); + let (mut tx, mut rx) = mpsc::channel(1); + OutputPort::send( + &mut tx, + Delivery::new(7, Count(count.clone())).map(|n| n.to_string()), + ) + .await + .unwrap_or_else(|_| panic!("channel closed")); + let message = InputPort::receive(&mut rx).await.unwrap(); + assert_eq!(message.value(), "7"); + assert_eq!(count.load(Ordering::SeqCst), 0); + assert_eq!(message.into_completed().await.ok().unwrap(), "7"); + assert_eq!(count.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn failed_send_returns_the_obligation_and_drop_does_not_complete() { + let count = Arc::new(AtomicUsize::new(0)); + let (mut tx, rx) = mpsc::channel(1); + drop(rx); + let pending = OutputPort::send(&mut tx, Delivery::new(7, Count(count.clone()))) + .await + .err() + .unwrap(); + drop(pending); + assert_eq!(count.load(Ordering::SeqCst), 0); + } + + struct Retry { + first: bool, + } + #[async_trait::async_trait] + impl Completion for Retry { + async fn complete(&mut self) -> anyhow::Result<()> { + if std::mem::take(&mut self.first) { + anyhow::bail!("head write failed"); + } + Ok(()) + } + } + + #[tokio::test] + async fn failed_completion_returns_the_value_and_can_retry() { + let failed = Delivery::new(7, Retry { first: true }) + .into_completed() + .await + .err() + .unwrap(); + assert_eq!(failed.error.to_string(), "head write failed"); + assert_eq!(*failed.delivery.value(), 7); + assert_eq!(failed.delivery.into_completed().await.ok().unwrap(), 7); + } +} diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs new file mode 100644 index 00000000..07c5a9b0 --- /dev/null +++ b/bt-daemon/src/pipeline/mod.rs @@ -0,0 +1,14 @@ +//! Typed stage contracts and completion-preserving channel messages. + +pub(crate) mod delivery; +pub(crate) mod plugins; + +pub(crate) use delivery::{Completion, Delivery}; + +#[async_trait::async_trait] +pub(crate) trait Stage: Send { + type Input: Sync; + type Outcome: Send; + + async fn process(&mut self, input: &Self::Input) -> Self::Outcome; +} diff --git a/bt-daemon/src/pipeline/plugins.rs b/bt-daemon/src/pipeline/plugins.rs new file mode 100644 index 00000000..ddce96b9 --- /dev/null +++ b/bt-daemon/src/pipeline/plugins.rs @@ -0,0 +1,34 @@ +use super::Stage; +use crate::{span_processor, translate::SpanOp}; +use std::path::PathBuf; + +/// The outcome says what happened; the caller chooses the next stage. +pub(crate) enum PluginOutcome { + Prepared(Box), + Filtered, + Failed(span_processor::PluginFailure), + Unavailable(anyhow::Error), +} + +pub(crate) struct Plugins { + pub paths: Vec, + pub source: String, + pub session_id: String, +} + +#[async_trait::async_trait] +impl Stage for Plugins { + type Input = SpanOp; + type Outcome = PluginOutcome; + + async fn process(&mut self, input: &SpanOp) -> PluginOutcome { + match span_processor::process(&self.paths, input, &self.source, &self.session_id) { + Ok(result) => match (result.op, result.failure) { + (_, Some(failure)) => PluginOutcome::Failed(failure), + (Some(op), None) => PluginOutcome::Prepared(Box::new(op)), + (None, None) => PluginOutcome::Filtered, + }, + Err(error) => PluginOutcome::Unavailable(error), + } + } +} diff --git a/bt-daemon/src/sink/braintrust.rs b/bt-daemon/src/sink/braintrust.rs index 7bca1248..100a1ecd 100644 --- a/bt-daemon/src/sink/braintrust.rs +++ b/bt-daemon/src/sink/braintrust.rs @@ -1,9 +1,8 @@ //! The Braintrust sink: maps sink-neutral [`SpanOp`]s onto `braintrust-sdk-rust`. //! -//! Multi-profile: a session's backend URLs come from its own config (bt -//! resolves them per profile), so clients are built lazily and cached by -//! `(api_url, app_url)` — sessions on the same instance share a client, -//! sessions on different instances get their own. Within a client, each +//! Each routing branch owns its checked SDK queue. Clients are built lazily +//! using its configured backend URLs; flush results cannot include another +//! branch's rejected operations. Within a client, each //! session's token/org travel per span (`span_builder_with_credentials`) and //! never leak across sessions. Span ids are the translator's deterministic //! UUIDv5 strings, reused as the SDK `row_id` merge key so journal replay @@ -57,20 +56,14 @@ impl CachedClient { if *queued_bytes < QUEUE_FLUSH_BYTE_BUDGET { return Ok(()); } - self.client - .flush() - .await - .map_err(|error| anyhow::anyhow!("braintrust flush failed: {error}"))?; + self.client.flush().await.map_err(anyhow::Error::new)?; *queued_bytes = 0; Ok(()) } async fn flush(&self) -> anyhow::Result<()> { let mut queued_bytes = self.queued_bytes.lock().await; - self.client - .flush() - .await - .map_err(|error| anyhow::anyhow!("braintrust flush failed: {error}"))?; + self.client.flush().await.map_err(anyhow::Error::new)?; *queued_bytes = 0; Ok(()) } @@ -122,6 +115,7 @@ impl ClientCache { } let client = BraintrustClient::builder() .skip_login(true) + .checked_delivery(true) .api_url(api_url.to_string()) .app_url(app_url.to_string()) .build() @@ -140,7 +134,6 @@ impl ClientCache { /// Hands out a per-session sink over the shared client pool. pub struct BraintrustSinkFactory { - cache: Arc, default_api_url: Option, default_app_url: Option, } @@ -148,7 +141,6 @@ pub struct BraintrustSinkFactory { impl BraintrustSinkFactory { pub fn new(cfg: BraintrustSinkConfig) -> Self { Self { - cache: Arc::new(ClientCache::new()), default_api_url: cfg.api_url, default_app_url: cfg.app_url, } @@ -158,7 +150,9 @@ impl BraintrustSinkFactory { impl SinkFactory for BraintrustSinkFactory { fn create(&self, _session_id: &str, source: &str) -> anyhow::Result> { Ok(Box::new(BraintrustSink { - cache: self.cache.clone(), + // Acceptance belongs to this routing branch. A shared SDK queue can + // attribute another branch's rejected batch to this branch's flush. + cache: Arc::new(ClientCache::new()), default_api_url: self.default_api_url.clone(), default_app_url: self.default_app_url.clone(), source: source.to_string(), @@ -732,6 +726,7 @@ mod tests { .await; let client = BraintrustClient::builder() .skip_login(true) + .checked_delivery(true) .api_url(server.uri()) .app_url(server.uri()) .environment("test", Some("sink-test")) diff --git a/bt-daemon/tests/braintrust_sink.rs b/bt-daemon/tests/braintrust_sink.rs index 46b843a5..98c24903 100644 --- a/bt-daemon/tests/braintrust_sink.rs +++ b/bt-daemon/tests/braintrust_sink.rs @@ -815,3 +815,91 @@ async fn experiment_sessions_use_experiment_object_type_and_id() { "experiment spans were routed as project logs: {bodies}" ); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rejected_backend_batch_is_not_a_successful_flush() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/logs3")) + .respond_with(ResponseTemplate::new(400).set_body_string("invalid row")) + .mount(&server) + .await; + let factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default()); + let mut sink = factory.create("rejected", "debug").unwrap(); + let mut config = session_config(&server.uri()); + config.destination = Some(TraceDestination::ProjectLogs { + project_id: Some("project".into()), + project_name: None, + }); + sink.configure(&config); + sink.emit(&[SpanOp::Insert(row( + "rejected", + "rejected", + &[], + "rejected", + SpanType::Task, + 1, + Some(2), + ))]) + .await + .unwrap(); + let flushed = sink.flush().await; + assert!( + flushed.is_err(), + "a rejected SDK batch must hold completion" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn rejected_route_does_not_poison_a_sibling_flush_on_the_same_backend() { + use wiremock::matchers::header; + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/logs3")) + .and(header("authorization", "Bearer sk-bad")) + .respond_with(ResponseTemplate::new(400)) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/logs3")) + .and(header("authorization", "Bearer sk-good")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + let factory = BraintrustSinkFactory::new(BraintrustSinkConfig::default()); + let mut bad = factory.create("bad", "debug").unwrap(); + let mut good = factory.create("good", "debug").unwrap(); + for (sink, token) in [(&mut bad, "sk-bad"), (&mut good, "sk-good")] { + let mut config = session_config(&server.uri()); + config.auth.token = token.into(); + config.destination = Some(TraceDestination::ProjectLogs { + project_id: Some("project".into()), + project_name: None, + }); + sink.configure(&config); + sink.emit(&[SpanOp::Insert(row( + token, + token, + &[], + token, + SpanType::Task, + 1, + Some(2), + ))]) + .await + .unwrap(); + } + let (bad_result, good_result) = tokio::join!(bad.flush(), good.flush()); + assert!(bad_result.is_err()); + good_result.unwrap(); +} From 44b27a99184c7e5ac659c367202ffe131ddcf896 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 13 Oct 2026 23:14:02 +0800 Subject: [PATCH 4/9] Preserve existing plugin diagnostic text during stage extraction --- bt-daemon/src/dispatch.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 37a6ed4e..820d50c1 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -750,7 +750,7 @@ impl SessionActor { "failed to persist span plugin diagnostic: {error}"); } self.set_error(format!( - "span plugin {} failed: {}", + "span plugin {} failed; span operations are being discarded: {}", failure.path.display(), failure.message )); From 96036995dfdafa10602e0a2568b20501d0c4da72 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 00:03:11 +0800 Subject: [PATCH 5/9] Fix async trait lint compatibility with Clippy 1.99 --- bt-daemon/src/pipeline/delivery.rs | 6 ++++++ bt-daemon/src/pipeline/mod.rs | 2 ++ 2 files changed, 8 insertions(+) diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs index 2d4b7464..776e12b8 100644 --- a/bt-daemon/src/pipeline/delivery.rs +++ b/bt-daemon/src/pipeline/delivery.rs @@ -9,6 +9,8 @@ pub(crate) struct Delivery { completion: Box, } +// async_trait adds a redundant must_use annotation to its boxed futures. +#[allow(clippy::double_must_use)] #[async_trait::async_trait] pub(crate) trait Completion: Send { async fn complete(&mut self) -> anyhow::Result<()>; @@ -49,12 +51,16 @@ impl Delivery { } } +// 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>; } +// 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; diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index 07c5a9b0..6180279e 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/bt-daemon/src/pipeline/mod.rs @@ -5,6 +5,8 @@ pub(crate) mod plugins; pub(crate) use delivery::{Completion, Delivery}; +// async_trait adds a redundant must_use annotation to its boxed futures. +#[allow(clippy::double_must_use)] #[async_trait::async_trait] pub(crate) trait Stage: Send { type Input: Sync; From e3b63f8bb1255b33be2b33bfa2e2ce0a7c5b2af9 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 00:12:46 +0800 Subject: [PATCH 6/9] Keep pipeline planning document outside the repository --- bt-daemon/docs/recoverable-pipeline-plan.md | 251 -------------------- 1 file changed, 251 deletions(-) delete mode 100644 bt-daemon/docs/recoverable-pipeline-plan.md diff --git a/bt-daemon/docs/recoverable-pipeline-plan.md b/bt-daemon/docs/recoverable-pipeline-plan.md deleted file mode 100644 index a120b5c7..00000000 --- a/bt-daemon/docs/recoverable-pipeline-plan.md +++ /dev/null @@ -1,251 +0,0 @@ -# Recoverable pipeline implementation plan - -Status: proposal. Supersedes the recoverable-stream pipeline design and implementation -plan and the event WAL recovery proposal. Start from current `main`, not PRs #130, -#132, or #171. This plan authorizes no implementation by itself. - -## 1. Goal and scope - -Build a concurrent pipeline of self-contained actors with typed input/output ports: - -```text -capture -> event journal -> translation -> SpanOp journal - | | - head A head B - | | - route A route B - | | - sink A sink B -``` - -The requirements are failure reasons with retry criteria, persisted SpanOps, and -isolated stages composed through channels. Each route owns an independent resumable -head in the shared SpanOp journal. Downstream recovery starts at its earliest -unresolved SpanOp; it never replays translation. - -Preserve trace shape, routing identity, attachment, imports, and existing ledger -semantics. Per-plugin durable checkpoints, translator snapshots, historical backfill, -prefix collection, and a new scheduler or workflow platform are outside scope. - -## 2. Composition contract - -Actors own mutable processing state. Stages accept typed inputs and return typed -outcome enums; assembly maps each variant to a compatible downstream input port. -Stages do not name successors, access another actor's state, or receive `Daemon`. -A small shared stage trait with associated `Input` and `Outcome` types and typed -input/output port traits is sufficient. Wiring must cover every outcome variant -and check connected types. Ordinary enum matches and adapters are sufficient. - -For example, a plugin returns `Prepared`, `Filtered`, or `Failed`. Assembly routes -those outcomes to submission, terminal completion, or recovery. The actor runner -preserves the message envelope across that mapping. Adding a validation stage -changes wiring rather than the plugin implementation. - -Both bounded Tokio channels and journal-backed channels expose send/receive ports -carrying the same envelope. Normal stages need not know which transport they use. -Storage adapters own durability, subscriptions, and completion; a successful volatile -send means queued, not consumed. A failed send retains or returns the message. - -### Message ownership and completion - -```rust -// Illustrative API; fields and completion plumbing are private. -struct Delivery { - value: T, - completion: Completion, -} - -impl Delivery { - fn value(&self) -> &T; - fn map(self, f: impl FnOnce(T) -> U) -> Delivery; - async fn into_completed(self) -> Result>; -} -``` - -`value()` borrows for processing. `map()` changes the value type while moving its -completion obligation. `into_completed()` claims the value has been handled, -confirms upstream completion, and returns ownership of the final value. The sink -calls it after checked acceptance; an intentional filter can also complete work. -If completion fails, `Unfinished` retains the value and context for retry. - -Volatile handoffs preserve the obligation. A no-op completion is valid only when -there is no upstream obligation. Dropping a message, cancelling a task, or releasing -an `Arc` does not complete it. An immutable payload may be shared with `Arc`, but -each route's completion capability is unique and non-cloneable. - -A bounded owner mailbox and one-shot reply can implement completion. The reply -confirms the durable head write, not merely receipt of the request. Completion -requests are idempotent and cannot skip unresolved positions; late or duplicate -requests cannot advance a different subscription or overwrite newer progress. -Only one live consumer owns a given head. Restart reconstructs envelopes from -persisted positions, not from channels or reference counts. - -For one input producing several outputs, keep a bounded batch under one obligation; -never clone the token onto each output. Terminal completion covers the whole batch. -Translation may produce no output: its empty commit still consumes the input. -General-purpose splitting/joining of obligations is unnecessary for this plan. - -### Concurrency and ordering - -The graph is shared; current stage and recovery state belong to pending executions, -not the pipeline globally. Use one ordered translation consumer per source/session -and one ordered routing consumer per route head. Different consumers run concurrently; -initially each processes one item or bounded batch at a time. Stateful translators -and route state belong to their consumers, not a global registered handler instance. - -Use ordinary Tokio tasks and bounded queues; retain worker-local plugin runtimes. -Small synchronous stages may share an actor. A recovering consumer stops later data -on its head while still servicing configuration, status, and shutdown controls. -Journal-backed edges let upstream work continue persisting while consumers wait. -Notifications are wake hints; subscribe/recheck against storage to avoid lost wakes. - -## 3. Durable channels - -Keep the event journal. Use a concrete SQLite adapter for SpanOps, heads, and recovery -records, with blocking database work outside Tokio workers. Keep SQL and codecs -inside storage. Capture acknowledges only its durable event append, independently -of downstream progress. Storage read/corruption errors hold work and report failure; -they never silently skip retained records. - -### Handoff and restart - -The SpanOp adapter atomically commits ordered output and its consumed event position. -That position is the authoritative translation boundary. Stable commit identities -make retry idempotent, including empty output. Support bounded output chunks without -exposing partial commits to readers or collecting a whole transcript into memory. - -At this durable handoff, the adapter resolves the preceding event obligation and -creates new obligations for downstream subscriptions. If output committed but -upstream confirmation was interrupted, reconcile the committed input position; -do not regenerate or append that output. Keep candidate output while retrying a -failed local commit before reading more input. - -Existing translators rebuild by replaying the committed event prefix with output -suppressed, then process the uncommitted suffix. Retain that prefix and required -transcript references. This rebuild is for translation state, never routing retry. -Output-producing checkpoint/finalize controls need stable ordered event identities -and captured transcript boundaries so restart reproduces the same input. - -Persist every field needed by routing, including creation provenance and operation -identity. Current `SpanRow` serialization skips fields such as `late_merge_key`; -use a storage codec that preserves them. Persist no credentials or auth leases. -Committed SpanOps are route-neutral; route preparation applies attachment, tags, -metadata, and plugin configuration to a private value. Version the storage codec -and hold unsupported records with a diagnostic rather than defaulting or skipping fields. - -### Subscription split and completion - -The producer sends one output batch without knowing route counts or heads. Assembly -registers subscriptions and pairs their receivers with routing actors. The durable -adapter supplies independent deliveries to each route; it need not eagerly push -copies or add a separate fan-out stage. - -Persist the nonsecret subscription identity/settings needed to resume without a -new event. Preserve existing route identity rules: credential refresh keeps its -head, and configuration changes must not redirect an old head to another destination. - -Each head advances monotonically over a contiguous consumed prefix. Reading and -forwarding never advance it. Sink enqueue is insufficient: require checked flush/ -acceptance for every covered operation, including when SDK clients are shared. -Preserve terminal-span and late-merge ledger behavior without recording acceptance -prematurely. - -Deferred roots and batching may read ahead while retaining the earliest unfinished -position. They must allow reading the child that releases a deferred root; later -success cannot skip the unresolved prefix. Keep pending memory bounded and preserve -existing deferral behavior. - -After backend acceptance, retry a failed local head write without rerunning delivery -in process. If confirmation is lost, reload the durable head. A crash before durable -completion may require idempotent resubmission with existing stable span IDs; the -pipeline provides at-least-once delivery, not exactly-once external effects. - -Retirement preserves subscriptions, heads, and pending work. Retain completed SpanOps -under the existing session retention policy so late routes can read history without -backfill. Delete a session only when retention permits, translation is caught up, -all heads are complete, and nothing is recovering. Protect event/transcript references -needed for reconstruction. Empty subscriber sets do not authorize accidental deletion. - -## 4. Recovery as a pipeline state - -A failure outcome carries a typed cause, local diagnostic, and retry condition. -Wiring maps it to `Recovering`, which retains the pending position, stage, boundary, -and condition evidence. It waits asynchronously, then returns an eligible outcome -mapped back into processing with updated conditions. No external supervisor restarts -stage clusters. Successful processing and completion, not eligibility alone, resolve -the failure. - -| Cause | Retry criterion | -| --- | --- | -| Plugin | Failed file's bytes/existence or relevant configured chain changes | -| Translation | Applicable translator revision changes | -| Credentials | Credential notification or bounded fresh resolution | -| Transport/server | Capped retry delay | -| Rate limit | Retry-After or capped fallback delay | -| Permission/destination | Relevant configuration change or bounded recheck | -| Permanent rejection | Relevant processor/configuration change | -| Local persistence | Bounded retry of the failed storage operation | - -Record exact plugin identity/digest and nonsecret configuration evidence. Unrelated -file edits do not qualify. Classify typed errors at their source rather than parsing -error strings. Restart re-evaluates persisted criteria; it is not itself permission -to retry unchanged plugins. Reload changed plugins or fresh credentials on retry. - -Routing retries reload unresolved SpanOps and may repeat downstream transformations. -Translation failures use the event boundary. Persistence failures retry the local -commit/completion where possible. Persist recovery evidence and expose stage, pending -position, reason, and eligibility through doctor. A failure to persist diagnostics -must retain the original reason in memory and never release work. - -## 5. Incremental PRs - -Each PR has one reviewable responsibility, includes its checks, and leaves `main` -usable. Small preparatory extractions are acceptable; no line-count target applies. - -1. **Contracts inside the current actor.** Extract stage handlers, typed outcomes, - `Delivery`, and queue port adapters without changing task ownership or replay. - Establish checked sink acceptance and `into_completed()`. Test mapping, filtering, - no completion on drop, and rejected delivery retaining its checkpoint. -2. **Actor/channel separation.** Split translation and routing ownership within each - existing route, using bounded ports and the existing session handle as a facade. - Carry event identity through completion. Test ordering, backpressure, cancellation, - and controls during blocked sends. Preserve current journals/checkpoints. -3. **Route-neutral output.** Move attachment/tags/metadata into route preparation and - carry provenance with SpanOps. Remove translator route-config reads and sink version - setters. Test canonical output across routes and equivalent final trace behavior. -4. **Durable adapter.** Add SQLite commits, full SpanOp codec, heads, and subscription - envelopes behind the port contract. Test empty/multiple output, chunked commit, - interruption before/after commit, independent heads, and confirmed completion. - Keep production wiring unchanged until these contracts pass. -5. **Activate the journal boundary.** Replace the volatile edge with the durable port; - share translation per source/session and register a subscription per route. Include - ordered controls and necessary lifecycle wiring. Migrate released checkpoints - restartably; preserve data when a completion mapping cannot be proved. Test restart, - late routes, zero translator calls during routing resume, and healthy sibling progress. -6. **Recovery transitions.** Persist reasons/criteria, wire waiting and retry outcomes, - and restore pending work without new native events. Test exact plugin edits, - credential refresh, transport retries, unchanged-condition holds, and pending retention. - Plugin/auth/delivery retries must never enter translation. -7. **Lifecycle consolidation and cleanup.** Remove replaced per-route translation and - active event delivery checkpoints; retain upgrade readers. Validate flush/finalize, - managed runs, startup, retirement, shutdown, correlation, and import/live ledger - compatibility. Existing import entry points remain unchanged. - -Activation includes required lifecycle correctness; the last PR consolidates rather -than repairs an unshippable intermediate state. Flush targets ordered controls and -reports completion only when the covered route heads reach their output boundary. -Timeout or shutdown leaves unresolved work durable. Correlation progress follows -translation, not downstream delivery; adjust existing barriers narrowly to preserve -independence without redesigning correlation. - -## 6. Validation - -Test handlers separately from actor wiring. Fault-inject commit and head-write gaps, -including lost confirmation, completion cancellation, and actor failure between -volatile stages. Prove a blocked route does not block capture, translation, or its -sibling, and blocked translation leaves committed SpanOps deliverable. - -Use the actual SDK path to establish rejection/acceptance semantics, alongside fake -sinks for pipeline tests. Run targeted checks for each PR and the locked all-feature -daemon suite, formatting, and CI lints before presenting it. Persistence changes -need completed Linux, macOS, and Windows CI. Report local and remote results separately. From a509efaa18cdf7d294619608774eed5f993f7997 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 00:23:42 +0800 Subject: [PATCH 7/9] Keep rejected delivery uncheckpointed and defer unused channel ports --- bt-daemon/src/dispatch.rs | 51 ++++++++--------- bt-daemon/src/pipeline/delivery.rs | 61 ++++---------------- bt-daemon/tests/pipeline.rs | 92 ++++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+), 75 deletions(-) diff --git a/bt-daemon/src/dispatch.rs b/bt-daemon/src/dispatch.rs index 820d50c1..55896cd9 100644 --- a/bt-daemon/src/dispatch.rs +++ b/bt-daemon/src/dispatch.rs @@ -631,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)); } @@ -650,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)); } @@ -935,26 +937,22 @@ impl SessionActor { } } - async fn flush_sink(&self, sink: &mut Box) -> 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, 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; } let pending = Delivery::new( through, @@ -968,6 +966,7 @@ impl SessionActor { 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) { diff --git a/bt-daemon/src/pipeline/delivery.rs b/bt-daemon/src/pipeline/delivery.rs index 776e12b8..d6cd35cb 100644 --- a/bt-daemon/src/pipeline/delivery.rs +++ b/bt-daemon/src/pipeline/delivery.rs @@ -1,8 +1,3 @@ -// Queue port consumers are introduced in the next stack layer. -#![allow(dead_code)] - -use tokio::sync::mpsc; - /// A message owns its completion obligation until terminal handling succeeds. pub(crate) struct Delivery { value: T, @@ -17,6 +12,7 @@ pub(crate) trait Completion: Send { } pub(crate) struct Unfinished { + #[cfg_attr(not(test), allow(dead_code))] pub delivery: Delivery, pub error: anyhow::Error, } @@ -29,10 +25,12 @@ impl Delivery { } } + #[cfg_attr(not(test), allow(dead_code))] pub fn value(&self) -> &T { &self.value } + #[cfg_attr(not(test), allow(dead_code))] pub fn map(self, map: impl FnOnce(T) -> U) -> Delivery { Delivery { value: map(self.value), @@ -51,40 +49,6 @@ impl Delivery { } } -// 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>; -} - -// 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) -> Result<(), Delivery>; -} - -#[async_trait::async_trait] -impl InputPort for mpsc::Receiver> { - type Item = T; - async fn receive(&mut self) -> Option> { - self.recv().await - } -} - -#[async_trait::async_trait] -impl OutputPort for mpsc::Sender> { - type Item = T; - async fn send(&mut self, delivery: Delivery) -> Result<(), Delivery> { - mpsc::Sender::send(self, delivery) - .await - .map_err(|error| error.0) - } -} - #[cfg(test)] mod tests { use super::*; @@ -92,6 +56,7 @@ mod tests { atomic::{AtomicUsize, Ordering}, Arc, }; + use tokio::sync::mpsc; struct Count(Arc); #[async_trait::async_trait] @@ -105,14 +70,11 @@ mod tests { #[tokio::test] async fn mapping_and_channel_forwarding_preserve_completion() { let count = Arc::new(AtomicUsize::new(0)); - let (mut tx, mut rx) = mpsc::channel(1); - OutputPort::send( - &mut tx, - Delivery::new(7, Count(count.clone())).map(|n| n.to_string()), - ) - .await - .unwrap_or_else(|_| panic!("channel closed")); - let message = InputPort::receive(&mut rx).await.unwrap(); + let (tx, mut rx) = mpsc::channel(1); + tx.send(Delivery::new(7, Count(count.clone())).map(|n| n.to_string())) + .await + .unwrap_or_else(|_| panic!("channel closed")); + let message = rx.recv().await.unwrap(); assert_eq!(message.value(), "7"); assert_eq!(count.load(Ordering::SeqCst), 0); assert_eq!(message.into_completed().await.ok().unwrap(), "7"); @@ -122,9 +84,10 @@ mod tests { #[tokio::test] async fn failed_send_returns_the_obligation_and_drop_does_not_complete() { let count = Arc::new(AtomicUsize::new(0)); - let (mut tx, rx) = mpsc::channel(1); + let (tx, rx) = mpsc::channel(1); drop(rx); - let pending = OutputPort::send(&mut tx, Delivery::new(7, Count(count.clone()))) + let pending = tx + .send(Delivery::new(7, Count(count.clone()))) .await .err() .unwrap(); diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 2a285244..7a437b38 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -569,6 +569,98 @@ async fn shutdown(socket: &Path) { shutdown_daemon(socket).await.unwrap(); } +/// A failed flush consumes its batch, just like checked SDK delivery. An empty +/// subsequent flush must not acknowledge that lost batch's source events. +#[tokio::test] +async fn rejected_flush_blocks_checkpoints_until_journal_replay() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[derive(Default)] + struct RejectOnce { + flushes: AtomicUsize, + emitted: AtomicUsize, + } + struct RejectOnceSink(Arc); + struct RejectOnceFactory(Arc); + impl SinkFactory for RejectOnceFactory { + fn create(&self, _: &str, _: &str) -> anyhow::Result> { + Ok(Box::new(RejectOnceSink(self.0.clone()))) + } + } + #[async_trait] + impl Sink for RejectOnceSink { + async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + self.0.emitted.fetch_add(ops.len(), Ordering::SeqCst); + Ok(ops.len() as u64) + } + async fn flush(&mut self) -> anyhow::Result<()> { + if self.0.flushes.fetch_add(1, Ordering::SeqCst) == 0 { + anyhow::bail!("backend rejected and consumed the batch"); + } + Ok(()) + } + } + + let tmp = tempfile::tempdir().unwrap(); + let data_dir = tmp.path().join("data"); + let socket = tmp.path().join("daemon.sock"); + let sink = Arc::new(RejectOnce::default()); + let start = || { + let args = ServeArgs { + socket: Some(socket.clone()), + data_dir: Some(data_dir.clone()), + idle_timeout_secs: 0, + session_idle_timeout_secs: 0, + }; + let opts = ServeOptions { + version: "test".into(), + translators: Arc::new(Registry::default_agents()), + sink_factory: Arc::new(RejectOnceFactory(sink.clone())), + auth_provider: Some(Arc::new(TestAuthProvider { + calls: Mutex::new(Vec::new()), + fail: false, + first_lease_expired: false, + })), + }; + tokio::spawn(async move { run_serve(args, opts).await.unwrap() }) + }; + let handle = start(); + wait_for(&socket).await; + let session = "rejected-flush"; + forward_envelope( + &envelope(session, "SessionStart", 1), + &socket, + &dummy_host(), + false, + ) + .await + .unwrap(); + flush_session(session, &socket, 5_000).await.unwrap(); + assert_eq!(sink.flushes.load(Ordering::SeqCst), 1); + forward_envelope(&envelope(session, "Stop", 2), &socket, &dummy_host(), false) + .await + .unwrap(); + flush_session(session, &socket, 5_000).await.unwrap(); + shutdown(&socket).await; + handle.await.unwrap(); + + let journal_path = source_journal_path(&data_dir, "debug", session); + let journal = std::fs::read_to_string(&journal_path).unwrap(); + assert!(!journal.contains("delivery_checkpoint"), "{journal}"); + assert_eq!(sink.flushes.load(Ordering::SeqCst), 1); + let emitted_before_restart = sink.emitted.load(Ordering::SeqCst); + + let handle = start(); + wait_for(&socket).await; + flush_session(session, &socket, 5_000).await.unwrap(); + assert!(sink.emitted.load(Ordering::SeqCst) > emitted_before_restart); + assert!(std::fs::read_to_string(journal_path) + .unwrap() + .contains("delivery_checkpoint")); + shutdown(&socket).await; + handle.await.unwrap(); +} + #[tokio::test] async fn routed_sessions_resolve_multiple_profiles_without_journaling_credentials() { let provider = Arc::new(TestAuthProvider { From 3514528194271ecdc58c44022cb5aafdd52f34c0 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 14 Oct 2026 00:48:03 +0800 Subject: [PATCH 8/9] Use the platform endpoint helper in rejected flush recovery test --- bt-daemon/tests/pipeline.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 7a437b38..06f81e45 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -603,7 +603,7 @@ async fn rejected_flush_blocks_checkpoints_until_journal_replay() { let tmp = tempfile::tempdir().unwrap(); let data_dir = tmp.path().join("data"); - let socket = tmp.path().join("daemon.sock"); + let socket = test_endpoint(tmp.path()); let sink = Arc::new(RejectOnce::default()); let start = || { let args = ServeArgs { From 2c4e7be8ab91ab4a36632e1d4b3fa7a9d3dc1dce Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 20 Oct 2026 01:25:05 +0800 Subject: [PATCH 9/9] Pin checked delivery to the merged Rust SDK commit --- bt-daemon/Cargo.lock | 2 +- bt-daemon/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/bt-daemon/Cargo.lock b/bt-daemon/Cargo.lock index 84cf1efe..fe6c4914 100644 --- a/bt-daemon/Cargo.lock +++ b/bt-daemon/Cargo.lock @@ -239,7 +239,7 @@ dependencies = [ [[package]] name = "braintrust-sdk-rust" version = "0.1.0-alpha.2" -source = "git+https://github.com/braintrustdata/braintrust-sdk-rust?rev=74e568dd30eb1654039dcc792f3327a8399deaba#74e568dd30eb1654039dcc792f3327a8399deaba" +source = "git+https://github.com/braintrustdata/braintrust-sdk-rust?rev=f2216fa0ceee132e02f1392c0e5af89ae762e9c8#f2216fa0ceee132e02f1392c0e5af89ae762e9c8" dependencies = [ "anyhow", "arc-swap", diff --git a/bt-daemon/Cargo.toml b/bt-daemon/Cargo.toml index 94b6b979..b058b2bf 100644 --- a/bt-daemon/Cargo.toml +++ b/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 = "74e568dd30eb1654039dcc792f3327a8399deaba" } +braintrust-sdk-rust = { git = "https://github.com/braintrustdata/braintrust-sdk-rust", rev = "f2216fa0ceee132e02f1392c0e5af89ae762e9c8" } base64 = "0.22" anyhow = "1" parking_lot = "0.12"