Keep provisional roots out of turn delivery barriers

bt-daemon/README.md

@@ -304,6 +304,11 @@ or intentional filtering. Plugin, authentication, and delivery recovery resumes
 from unresolved SpanOps without replaying translation. Retries use deterministic
 span IDs, so repeated delivery updates the same spans.
 
+Provisional session roots stay hidden until their first child arrives. Root
+candidates are saved with translation's input checkpoint, independently of route
+delivery, so an unused root does not hold completed turn batches open. When work
+arrives, the root's saved operations are published before its child.
+
 Failures retain their reason and retry condition, including across restarts.
 The daemon waits for the relevant plugin/configuration change, fresh credentials,
 or retry delay before trying again. `bt trace doctor` reports the affected stage,

bt-daemon/src/pipeline/journal.rs

@@ -12,6 +12,7 @@ use parking_lot::Mutex;
 use rusqlite::{params, Connection, OptionalExtension};
 use serde::{Deserialize, Serialize};
 use std::{
+    collections::BTreeMap,
     path::Path,
     sync::{
         atomic::{AtomicBool, Ordering},
@@ -82,7 +83,7 @@ impl SpanJournal {
             let db = Connection::open(path)?;
             db.busy_timeout(std::time::Duration::from_secs(5))?;
             let version: u32 = db.query_row("PRAGMA user_version", [], |row| row.get(0))?;
-            anyhow::ensure!(version <= 3, "unsupported SpanOp journal schema {version}");
+            anyhow::ensure!(version <= 4, "unsupported SpanOp journal schema {version}");
             db.execute_batch("PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;")?;
             if version < 3 {
                 db.execute_batch(
@@ -108,6 +109,9 @@ impl SpanJournal {
                 db.execute_batch("ALTER TABLE heads ADD COLUMN binding TEXT")?;
                 db.execute_batch("PRAGMA user_version=3; COMMIT")?;
             }
+            if version < 4 {
+                db.execute_batch("BEGIN IMMEDIATE; CREATE TABLE root_candidates (span_id TEXT PRIMARY KEY, value BLOB NOT NULL); PRAGMA user_version=4; COMMIT")?;
+            }
             let tail = db.query_row("SELECT coalesce(max(position),0) FROM ops JOIN batches USING(input) WHERE committed=1", [], |row| row.get(0))?;
             Ok((db, tail))
         })
@@ -216,6 +220,56 @@ impl SpanJournal {
     }
 
     pub async fn commit(&self, input: u64) -> anyhow::Result<u64> {
+        self.commit_output(input, None).await
+    }
+
+    /// Root candidates and translated output share the consumed-input boundary.
+    pub async fn commit_with_roots(
+        &self,
+        input: u64,
+        roots: &BTreeMap<String, Vec<SpanOp>>,
+    ) -> anyhow::Result<u64> {
+        let roots = roots
+            .iter()
+            .map(|(id, ops)| {
+                let ops = ops
+                    .iter()
+                    .map(StoredOp::encode)
+                    .collect::<anyhow::Result<Vec<_>>>()?;
+                Ok((id.clone(), serde_json::to_vec(&ops)?))
+            })
+            .collect::<anyhow::Result<Vec<_>>>()?;
+        self.commit_output(input, Some(roots)).await
+    }
+
+    pub async fn root_candidates(&self) -> anyhow::Result<BTreeMap<String, Vec<SpanOp>>> {
+        self.with_db(|db| {
+            let mut query = db.prepare("SELECT span_id,value FROM root_candidates")?;
+            let roots = query
+                .query_map([], |row| {
+                    Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
+                })?
+                .map(|row| {
+                    let (id, bytes) = row?;
+                    let ops: Vec<Vec<u8>> = serde_json::from_slice(&bytes)?;
+                    Ok((
+                        id,
+                        ops.iter()
+                            .map(|bytes| StoredOp::decode(bytes))
+                            .collect::<anyhow::Result<Vec<_>>>()?,
+                    ))
+                })
+                .collect();
+            roots
+        })
+        .await
+    }
+
+    async fn commit_output(
+        &self,
+        input: u64,
+        roots: Option<Vec<(String, Vec<u8>)>>,
+    ) -> anyhow::Result<u64> {
         let tail = self
             .with_db(move |db| {
                 let tx = db.transaction()?;
@@ -238,6 +292,15 @@ impl SpanJournal {
                     tx.execute("UPDATE batches SET committed=1 WHERE input=?", [input])? == 1,
                     "missing translation batch"
                 );
+                if let Some(roots) = roots {
+                    tx.execute("DELETE FROM root_candidates", [])?;
+                    for (id, value) in roots {
+                        tx.execute(
+                            "INSERT INTO root_candidates(span_id,value) VALUES (?,?)",
+                            params![id, value],
+                        )?;
+                    }
+                }
                 let tail = tx.query_row(
                     "SELECT coalesce(max(position),0) FROM ops WHERE input<=?",
                     [input],
@@ -780,6 +843,67 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn root_candidates_commit_and_release_atomically() {
+        let tmp = tempfile::tempdir().unwrap();
+        let path = tmp.path().join("roots.sqlite");
+        let journal = SpanJournal::open(&path).await.unwrap();
+        let mut roots = crate::pipeline::roots::Roots::default();
+        let mut root = op("root");
+        let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &mut root;
+        row.metadata = Some(serde_json::json!({"_bt_defer_root": true}));
+        assert!(roots.process(&vec![root]).await.is_empty());
+        journal.begin(10, true).await.unwrap();
+        let db = Connection::open(&path).unwrap();
+        db.execute_batch("CREATE TRIGGER fail_roots BEFORE INSERT ON root_candidates BEGIN SELECT RAISE(FAIL,'root write'); END;").unwrap();
+        assert!(journal
+            .commit_with_roots(10, roots.pending())
+            .await
+            .is_err());
+        assert_eq!(journal.translated_through().await.unwrap(), 0);
+        assert!(journal.root_candidates().await.unwrap().is_empty());
+        db.execute_batch("DROP TRIGGER fail_roots").unwrap();
+        journal
+            .commit_with_roots(10, roots.pending())
+            .await
+            .unwrap();
+        journal.close().await.unwrap();
+
+        let journal = SpanJournal::open(&path).await.unwrap();
+        let candidates = journal.root_candidates().await.unwrap();
+        let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &candidates["root"][0];
+        assert_eq!(row.late_merge_key.as_deref(), Some("late"));
+        assert_eq!(
+            row.origin.as_ref().unwrap().plugin_version.as_deref(),
+            Some("capture")
+        );
+        let mut roots = crate::pipeline::roots::Roots::restore(candidates);
+        let child = SpanOp::Insert(SpanRow {
+            span_id: "turn".into(),
+            parent_span_ids: vec!["root".into()],
+            ..Default::default()
+        });
+        let ready = roots.process(&vec![child]).await;
+        assert_eq!(ready.len(), 2);
+        journal.begin(20, true).await.unwrap();
+        journal.append(20, &ready).await.unwrap();
+        db.execute_batch("CREATE TRIGGER fail_release BEFORE DELETE ON root_candidates BEGIN SELECT RAISE(FAIL,'release'); END;").unwrap();
+        assert!(journal
+            .commit_with_roots(20, roots.pending())
+            .await
+            .is_err());
+        assert_eq!(journal.translated_through().await.unwrap(), 10);
+        assert_eq!(journal.tail().await.unwrap(), 0);
+        assert_eq!(journal.root_candidates().await.unwrap().len(), 1);
+        db.execute_batch("DROP TRIGGER fail_release").unwrap();
+        journal
+            .commit_with_roots(20, roots.pending())
+            .await
+            .unwrap();
+        assert_eq!(journal.tail().await.unwrap(), 2);
+        assert!(journal.root_candidates().await.unwrap().is_empty());
+    }
+
     #[tokio::test]
     async fn input_port_replays_abandoned_delivery_without_cursor_control() {
         use crate::pipeline::delivery::InputPort;

bt-daemon/src/pipeline/mod.rs

@@ -3,6 +3,7 @@
 pub(crate) mod delivery;
 pub(crate) mod plugins;
 pub(crate) mod prepare;
+pub(crate) mod roots;
 pub(crate) mod routing;
 pub(crate) mod submission;
 

bt-daemon/src/pipeline/roots.rs

@@ -0,0 +1,134 @@
+//! Root candidates are translation state, not unfinished sink deliveries.
+use super::Stage;
+use crate::translate::SpanOp;
+use std::collections::BTreeMap;
+
+#[derive(Default)]
+pub(crate) struct Roots {
+    pending: BTreeMap<String, Vec<SpanOp>>,
+}
+
+impl Roots {
+    pub fn restore(pending: BTreeMap<String, Vec<SpanOp>>) -> Self {
+        Self { pending }
+    }
+
+    pub fn pending(&self) -> &BTreeMap<String, Vec<SpanOp>> {
+        &self.pending
+    }
+}
+
+#[async_trait::async_trait]
+impl Stage for Roots {
+    type Input = Vec<SpanOp>;
+    type Outcome = Vec<SpanOp>;
+
+    async fn process(&mut self, ops: &Vec<SpanOp>) -> Vec<SpanOp> {
+        let mut ready = Vec::new();
+        for op in ops {
+            let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op;
+            if let Some(root) = self.pending.get_mut(&row.span_id) {
+                root.push(op.clone());
+                continue;
+            }
+            for parent in &row.parent_span_ids {
+                if let Some(root) = self.pending.remove(parent) {
+                    for mut op in root {
+                        let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &mut op;
+                        if let Some(metadata) =
+                            row.metadata.as_mut().and_then(|v| v.as_object_mut())
+                        {
+                            metadata.remove("_bt_defer_root");
+                        }
+                        ready.push(op);
+                    }
+                }
+            }
+            if matches!(op, SpanOp::Insert(_))
+                && row.metadata.as_ref().and_then(|v| v.get("_bt_defer_root"))
+                    == Some(&serde_json::Value::Bool(true))
+            {
+                self.pending.insert(row.span_id.clone(), vec![op.clone()]);
+            } else {
+                ready.push(op.clone());
+            }
+        }
+        ready
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::translate::SpanRow;
+    use serde_json::json;
+
+    fn root(id: &str) -> SpanOp {
+        SpanOp::Insert(SpanRow {
+            span_id: id.into(),
+            root_span_id: id.into(),
+            metadata: Some(json!({"_bt_defer_root": true})),
+            ..Default::default()
+        })
+    }
+
+    fn child(id: &str, parent: &str) -> SpanOp {
+        SpanOp::Insert(SpanRow {
+            span_id: id.into(),
+            root_span_id: parent.into(),
+            parent_span_ids: vec![parent.into()],
+            ..Default::default()
+        })
+    }
+
+    #[tokio::test]
+    async fn unused_root_does_not_hold_other_roots_turns() {
+        let mut roots = Roots::default();
+        assert!(roots
+            .process(&vec![root("unused"), root("active")])
+            .await
+            .is_empty());
+        let output = roots.process(&vec![child("turn", "active")]).await;
+        assert_eq!(output.len(), 2);
+        let (SpanOp::Insert(row) | SpanOp::Merge(row)) = &output[0];
+        assert_eq!(row.span_id, "active");
+        assert!(row
+            .metadata
+            .as_ref()
+            .unwrap()
+            .get("_bt_defer_root")
+            .is_none());
+        assert!(roots.pending.contains_key("unused"));
+        // An arbitrary number of later turns leaves just the unrelated root,
+        // without retaining their values or completion obligations.
+        for index in 0..1000 {
+            assert_eq!(
+                roots
+                    .process(&vec![child(&format!("turn-{index}"), "active")])
+                    .await
+                    .len(),
+                1
+            );
+        }
+        assert_eq!(roots.pending.len(), 1);
+        assert_eq!(roots.pending["unused"].len(), 1);
+    }
+
+    #[tokio::test]
+    async fn restored_root_updates_are_reported_before_its_first_child() {
+        let mut roots = Roots::default();
+        let closed = SpanOp::Merge(SpanRow {
+            span_id: "root".into(),
+            end_ms: Some(2),
+            ..Default::default()
+        });
+        assert!(roots.process(&vec![root("root"), closed]).await.is_empty());
+        let mut restored = Roots::restore(roots.pending);
+        let output = restored.process(&vec![child("turn", "root")]).await;
+        assert_eq!(output.len(), 3);
+        assert!(matches!(&output[0], SpanOp::Insert(row) if row.span_id == "root"));
+        assert!(matches!(&output[1], SpanOp::Merge(row) if row.end_ms == Some(2)));
+        assert!(matches!(&output[2], SpanOp::Insert(row) if row.span_id == "turn"));
+        assert!(restored.pending.is_empty());
+    }
+}