From e3d50f2963b8bb7981bcbbc67c07aed993dbf4fe Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 02:50:51 +0800 Subject: [PATCH 1/3] Stage provisional roots independently of turn delivery --- bt-daemon/README.md | 5 + bt-daemon/src/pipeline/journal.rs | 126 ++++++++++++++++++++++- bt-daemon/src/pipeline/mod.rs | 1 + bt-daemon/src/pipeline/roots.rs | 134 +++++++++++++++++++++++++ bt-daemon/src/pipeline/translation.rs | 96 ++++++++++++------ bt-daemon/tests/distributed_tracing.rs | 21 +++- bt-daemon/tests/pipeline.rs | 61 +++++++++++ bt-daemon/tests/root_provenance.rs | 4 +- 8 files changed, 416 insertions(+), 32 deletions(-) create mode 100644 bt-daemon/src/pipeline/roots.rs diff --git a/bt-daemon/README.md b/bt-daemon/README.md index 941b2ac5..78a4e86e 100644 --- a/bt-daemon/README.md +++ b/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, diff --git a/bt-daemon/src/pipeline/journal.rs b/bt-daemon/src/pipeline/journal.rs index 3c336c08..30df4dca 100644 --- a/bt-daemon/src/pipeline/journal.rs +++ b/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 { + 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>, + ) -> anyhow::Result { + let roots = roots + .iter() + .map(|(id, ops)| { + let ops = ops + .iter() + .map(StoredOp::encode) + .collect::>>()?; + Ok((id.clone(), serde_json::to_vec(&ops)?)) + }) + .collect::>>()?; + self.commit_output(input, Some(roots)).await + } + + pub async fn root_candidates(&self) -> anyhow::Result>> { + 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>(1)?)) + })? + .map(|row| { + let (id, bytes) = row?; + let ops: Vec> = serde_json::from_slice(&bytes)?; + Ok(( + id, + ops.iter() + .map(|bytes| StoredOp::decode(bytes)) + .collect::>>()?, + )) + }) + .collect(); + roots + }) + .await + } + + async fn commit_output( + &self, + input: u64, + roots: Option)>>, + ) -> anyhow::Result { 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; diff --git a/bt-daemon/src/pipeline/mod.rs b/bt-daemon/src/pipeline/mod.rs index 371bd9bb..d2a9a7af 100644 --- a/bt-daemon/src/pipeline/mod.rs +++ b/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; diff --git a/bt-daemon/src/pipeline/roots.rs b/bt-daemon/src/pipeline/roots.rs new file mode 100644 index 00000000..a0c9f41e --- /dev/null +++ b/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>, +} + +impl Roots { + pub fn restore(pending: BTreeMap>) -> Self { + Self { pending } + } + + pub fn pending(&self) -> &BTreeMap> { + &self.pending + } +} + +#[async_trait::async_trait] +impl Stage for Roots { + type Input = Vec; + type Outcome = Vec; + + async fn process(&mut self, ops: &Vec) -> Vec { + 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()); + } +} diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index a60437a0..9bb4a105 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -1,6 +1,7 @@ //! One ordered source consumer. Output publication is independent of route delivery. use super::journal::SpanJournal; use super::prepare::Prepare; +use super::roots::Roots; use super::{ recovery::{Boundary, Failure, Recovery, RecoveryOutcome}, Stage, @@ -102,6 +103,7 @@ impl Translation { reader: None, observers: Vec::new(), origins: HashMap::new(), + roots: Roots::default(), capture: None, pending_snapshots: HashSet::new(), bounds: HashMap::new(), @@ -191,6 +193,7 @@ struct Publication { capture: Option, ops: Option>, begun: bool, + prepared: bool, complete: bool, } @@ -204,6 +207,7 @@ struct Producer { reader: Option, observers: Vec, origins: HashMap, + roots: Roots, capture: Option, pending_snapshots: HashSet, bounds: HashMap, @@ -421,6 +425,17 @@ impl Producer { return Ok(()); } self.committed = self.journal.translated_through().await?; + self.roots = Roots::restore(self.journal.root_candidates().await?); + for ops in self.roots.pending().values() { + for op in ops { + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + if let Some(origin) = &row.origin { + self.origins + .entry(row.span_id.clone()) + .or_insert_with(|| origin.clone()); + } + } + } // Load creation provenance so an upgrade cannot restamp old spans. let mut after = 0; loop { @@ -557,6 +572,7 @@ impl Producer { capture, ops: Some(first), begun: false, + prepared: false, complete: false, }); self.publish().await?; @@ -588,31 +604,37 @@ impl Producer { } while !pending.complete { if let Some(ops) = pending.ops.as_mut() { - for op in ops.iter_mut() { - let insert = matches!(op, SpanOp::Insert(_)); - let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; - if insert { - self.origins.entry(row.span_id.clone()).or_insert_with(|| { - pending.capture.clone().unwrap_or_else(|| OriginSnapshot { - plugin_version: None, - source_version: None, - bt_version: self.options.version.clone(), - }) - }); + if !pending.prepared { + for op in ops.iter_mut() { + let insert = matches!(op, SpanOp::Insert(_)); + let (SpanOp::Insert(row) | SpanOp::Merge(row)) = op; + if insert { + self.origins.entry(row.span_id.clone()).or_insert_with(|| { + pending.capture.clone().unwrap_or_else(|| OriginSnapshot { + plugin_version: None, + source_version: None, + bt_version: self.options.version.clone(), + }) + }); + } + row.origin = self.origins.get(&row.span_id).cloned(); } - row.origin = self.origins.get(&row.span_id).cloned(); - } - self.phase = Boundary::Journal; - if pending.observation { - for observer in &self.observers { - Self::observe( - &self.options, - &mut self.pending_snapshots, - ops, - observer, - ) - .await?; + self.phase = Boundary::Journal; + if pending.observation { + for observer in &self.observers { + Self::observe( + &self.options, + &mut self.pending_snapshots, + ops, + observer, + ) + .await?; + } + } + if publish { + *ops = self.roots.process(ops).await; } + pending.prepared = true; } if publish { self.journal.append(pending.input, ops).await?; @@ -626,10 +648,13 @@ impl Producer { .ok_or_else(|| anyhow::anyhow!("translator is missing"))? .drain_pending(&ctx)?; pending.complete = pending.ops.is_none(); + pending.prepared = false; } if publish { self.phase = Boundary::Journal; - self.journal.commit(pending.input).await?; + self.journal + .commit_with_roots(pending.input, self.roots.pending()) + .await?; self.committed = pending.input; self.progress .committed @@ -799,6 +824,7 @@ mod tests { reader: None, observers: Vec::new(), origins: HashMap::new(), + roots: Roots::default(), capture: None, pending_snapshots: HashSet::new(), bounds: HashMap::new(), @@ -810,11 +836,19 @@ mod tests { input: 10, observation: true, capture: None, - ops: Some(vec![SpanOp::Insert(SpanRow { - span_id: "built".into(), - ..Default::default() - })]), + ops: Some(vec![ + SpanOp::Insert(SpanRow { + span_id: "unused-root".into(), + metadata: Some(serde_json::json!({"_bt_defer_root": true})), + ..Default::default() + }), + SpanOp::Insert(SpanRow { + span_id: "built".into(), + ..Default::default() + }), + ]), begun: false, + prepared: false, complete: false, }), }; @@ -831,5 +865,11 @@ mod tests { assert_eq!(drains.load(Ordering::Relaxed), 1); assert_eq!(journal.translated_through().await.unwrap(), 10); assert_eq!(journal.tail().await.unwrap(), 1); + let roots = journal.root_candidates().await.unwrap(); + assert_eq!( + roots["unused-root"].len(), + 1, + "a local retry must not stage the root twice" + ); } } diff --git a/bt-daemon/tests/distributed_tracing.rs b/bt-daemon/tests/distributed_tracing.rs index fb7c9c6b..6a666a5d 100644 --- a/bt-daemon/tests/distributed_tracing.rs +++ b/bt-daemon/tests/distributed_tracing.rs @@ -965,7 +965,26 @@ async fn session_fallback_survives_daemon_restart() { ); forward(child_start.remove(0), &socket, &host).await; flush("pending-child", &socket).await; - assert_attached_to_session(first_recording.as_ref(), "pending-parent", "pending-child"); + // Correlation is decided before work arrives, but the empty child's root + // is still a candidate rather than a sink delivery. + assert!(!first_recording + .sessions + .lock() + .unwrap() + .contains_key("pending-child")); + let status = run_status(StatusArgs { + socket: Some(socket.clone()), + session_id: Some("pending-child".into()), + }) + .await + .unwrap() + .unwrap(); + assert!(status + .sessions + .iter() + .any(|session| session.route.as_ref().is_some_and(|route| { + matches!(route.destination, Some(TraceDestination::ParentSpan { .. })) + }))); shutdown_daemon(&socket).await.unwrap(); first_daemon.await.unwrap(); diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index f75b8ff3..10981b1e 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -1308,6 +1308,7 @@ async fn unknown_sources_are_rejected_before_journaling() { #[tokio::test] async fn identical_native_ids_from_different_sources_are_isolated() { + use std::io::Write; let (data_dir, socket, handle, tmp) = start_daemon().await; let host = dummy_host(); let codex_transcript = tmp.path().join("codex-rollout.jsonl"); @@ -1327,6 +1328,21 @@ async fn identical_native_ids_from_different_sources_are_isolated() { ) .unwrap(); let mut codex = envelope("shared-native-id", "SessionStart", 1); + // Both sources need actual work; an empty Codex session stays hidden. + let mut transcript = std::fs::OpenOptions::new() + .append(true) + .open(&codex_transcript) + .unwrap(); + writeln!( + transcript, + "{}", + serde_json::json!({ + "timestamp": "2026-01-01T00:00:02Z", + "type": "event_msg", + "payload": {"type": "task_started", "turn_id": "turn"} + }) + ) + .unwrap(); codex.source = "codex".into(); codex.payload = serde_json::json!({ "session_id": "shared-native-id", @@ -1493,6 +1509,41 @@ async fn opencode_lifecycle_flushes_without_an_explicit_client_flush() { .await .unwrap(); + let mut work = start.clone(); + work.event = "chat.message".into(); + work.payload = serde_json::json!({ + "input": {"sessionID": native_session_id}, + "output": {"parts": [{"type": "text", "text": "hello"}]} + }); + forward_envelope(&work, &socket, &host, false) + .await + .unwrap(); + + // Observe the work flush separately, so it cannot satisfy the terminal + // event's assertion below. + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if flushes + .lock() + .unwrap() + .get(&session_id) + .copied() + .unwrap_or_default() + > 0 + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + // Disposal adds no translated rows; its work was already flushed. + // The other lifecycle events must also deliver their terminal updates. + if event != "server.instance.disposed" { + flushes.lock().unwrap().insert(session_id.clone(), 0); + } + let mut terminal = envelope(&session_id, event, 2); terminal.source = "opencode".into(); terminal.payload = serde_json::json!({ @@ -2590,6 +2641,16 @@ async fn managed_run_completion_finalizes_an_open_agent_root() { .await .unwrap(); + let mut work = env.clone(); + work.event = "chat.message".into(); + work.payload = serde_json::json!({ + "input": {"sessionID": "native"}, + "output": {"parts": [{"type": "text", "text": "hello"}]} + }); + forward_envelope(&work, &socket, &dummy_host(), false) + .await + .unwrap(); + let checkpoint = flush_session("managed-opencode", &socket, 5_000) .await .unwrap(); diff --git a/bt-daemon/tests/root_provenance.rs b/bt-daemon/tests/root_provenance.rs index e0f09230..a72b33a3 100644 --- a/bt-daemon/tests/root_provenance.rs +++ b/bt-daemon/tests/root_provenance.rs @@ -703,8 +703,8 @@ async fn attached_provisional_root_keeps_birth_snapshot_until_work_after_restart first.send(&birth).await; let pending = flush_session(session, &first.socket, 250).await.unwrap(); assert!( - !pending.flushed && pending.pending > 0, - "a deferred root has not been delivered" + pending.flushed && pending.pending == 0, + "a provisional root must not hold the delivery boundary open" ); assert!( logs3_rows(&server).await.is_empty(), From cdff8ff95ae08d8bd6f99b6b779ca76b88f23723 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 02:57:24 +0800 Subject: [PATCH 2/3] Restore staged root correlation when observers are recreated --- bt-daemon/src/pipeline/translation.rs | 86 +++++++++++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index 9bb4a105..61e1b7ea 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -671,6 +671,22 @@ impl Producer { } async fn observe_history(&mut self, observer: &Observation) -> anyhow::Result<()> { + // Candidates have no delivery position yet, but their birth records + // identify the session for correlation. Defensive retirement updates + // are not native observations and must not close a recreated observer. + for ops in self.journal.root_candidates().await?.into_values() { + let births = ops + .into_iter() + .filter(|op| matches!(op, SpanOp::Insert(_))) + .collect::>(); + Self::observe( + &self.options, + &mut self.pending_snapshots, + &births, + observer, + ) + .await?; + } let mut after = 0; loop { let rows = self.journal.read_after(after, 128).await?; @@ -717,6 +733,76 @@ mod tests { use super::*; use crate::{pipeline::delivery::InputPort, translate::SpanRow}; + #[tokio::test] + async fn recreated_observer_restores_a_staged_session_root() { + let tmp = tempfile::tempdir().unwrap(); + let event_path = tmp.path().join("events.ndjson"); + let correlation = Arc::new(crate::correlation::CorrelationRegistry::default()); + let translation = Translation::spawn(TranslationOptions { + source: "debug".into(), + session_id: "staged".into(), + namespace: "staged".into(), + version: "test".into(), + data_dir: tmp.path().into(), + events: Arc::new(tokio::sync::Mutex::new( + JournalWriter::open_path(&event_path).await.unwrap(), + )), + event_path, + registry: Arc::new(Registry::default_agents()), + correlation: correlation.clone(), + correlation_changed: Arc::new(Notify::new()), + }) + .await + .unwrap(); + let mut roots = Roots::default(); + roots + .process(&vec![ + SpanOp::Insert(SpanRow { + span_id: "root".into(), + root_span_id: "root".into(), + metadata: Some(serde_json::json!({"_bt_defer_root":true})), + ..Default::default() + }), + SpanOp::Merge(SpanRow { + span_id: "root".into(), + end_ms: Some(1), + ..Default::default() + }), + ]) + .await; + translation.journal.begin(10, false).await.unwrap(); + translation + .journal + .commit_with_roots(10, roots.pending()) + .await + .unwrap(); + let route = SessionRoute::default(); + let observer = Observation { + key: "observer".into(), + config: route.with_auth(crate::wire::BackendAuth { + token: "test".into(), + api_url: None, + app_url: None, + org_name: Some("org".into()), + org_id: None, + }), + route, + }; + for _ in 0..2 { + translation.observe(observer.clone()).await.unwrap(); + translation.barrier().await.unwrap(); + // A retired producer's defensive terminal update must not close + // the newly restored session fallback. + assert!( + correlation.active_parent_snapshot(&observer.key).is_some(), + "staged root must restore the session fallback" + ); + translation.unobserve(observer.key.clone()).await; + correlation.remove_session(&observer.key); + } + translation.shutdown().await; + } + #[tokio::test] async fn routing_resume_does_not_read_or_translate_the_committed_event_prefix() { let tmp = tempfile::tempdir().unwrap(); From a9055bb7d83a8d2b93455f955cad48d4d3a82127 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Sat, 17 Oct 2026 03:01:10 +0800 Subject: [PATCH 3/3] Use the normal flush deadline for root completion coverage --- bt-daemon/tests/root_provenance.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bt-daemon/tests/root_provenance.rs b/bt-daemon/tests/root_provenance.rs index a72b33a3..c95d7bc0 100644 --- a/bt-daemon/tests/root_provenance.rs +++ b/bt-daemon/tests/root_provenance.rs @@ -701,7 +701,7 @@ async fn attached_provisional_root_keeps_birth_snapshot_until_work_after_restart let first = Daemon::start(tmp.path(), &server, "bt1").await; first.send(&birth).await; - let pending = flush_session(session, &first.socket, 250).await.unwrap(); + let pending = flush_session(session, &first.socket, 5_000).await.unwrap(); assert!( pending.flushed && pending.pending == 0, "a provisional root must not hold the delivery boundary open"