From 71edc71ab94fd9c4ccc0dba4ea9e2c269ca13207 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 20 Oct 2026 23:00:16 +0800 Subject: [PATCH 1/4] fix(claude): keep lifecycle termination out of error --- bt-daemon/src/pipeline/translation.rs | 14 ++++++++------ bt-daemon/src/translate/claude.rs | 16 ++++++++++++---- bt-daemon/src/translate/mod.rs | 6 +++--- 3 files changed, 23 insertions(+), 13 deletions(-) diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index 042bc6d9..e54f8c1d 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -549,8 +549,9 @@ impl Producer { ) } JournalRecord::Control(control) => { - // Retirement exports defensive terminal updates but does not - // close native state when rebuilding for later captured work. + // Retirement releases daemon resources; it does not prove + // that the native agent session ended. Keep translator + // state resumable and only finalize explicit terminal runs. if input <= self.committed && matches!(control.kind, ControlKind::Retire) { self.position = input; continue; @@ -560,10 +561,11 @@ impl Producer { .translator .as_mut() .ok_or_else(|| anyhow::anyhow!("translator is missing"))?; - let ops = if !matches!(control.kind, ControlKind::Checkpoint) { - translator.finalize_at(&ctx, &control.bounds)? - } else { - translator.checkpoint_at(&ctx, &control.bounds)? + let ops = match control.kind { + ControlKind::Finalize => translator.finalize_at(&ctx, &control.bounds)?, + ControlKind::Checkpoint | ControlKind::Retire => { + translator.checkpoint_at(&ctx, &control.bounds)? + } }; (ops, self.capture.clone()) } diff --git a/bt-daemon/src/translate/claude.rs b/bt-daemon/src/translate/claude.rs index 8cd65b9a..5b58f8a5 100644 --- a/bt-daemon/src/translate/claude.rs +++ b/bt-daemon/src/translate/claude.rs @@ -258,6 +258,7 @@ struct ClaudeTranslator { session_parent_span_ids: Vec, root_open: bool, root_end_ms: Option, + root_generation: u32, turn: Option, last_turn_id: Option, turn_count: u32, @@ -307,6 +308,7 @@ impl ClaudeTranslator { session_parent_span_ids: Vec::new(), root_open: false, root_end_ms: None, + root_generation: 0, turn: None, last_turn_id: None, turn_count: 0, @@ -356,6 +358,14 @@ impl ClaudeTranslator { if self.root_open { return; } + if self.root_end_ms.is_some() { + self.root_generation = self.root_generation.saturating_add(1); + let root = ids::span_id(&self.session_id, &format!("root:{}", self.root_generation)); + self.session_span_id = root.clone(); + self.root_span_id = root; + self.root_end_ms = None; + self.claude_version_logged = false; + } self.root_open = true; let cwd = hook.cwd.clone().unwrap_or_default(); let workspace = basename(&cwd); @@ -887,7 +897,6 @@ impl ClaudeTranslator { { self.extend_user_turn(&owner, event.ts_ms, ops); } - self.end_root(event.ts_ms, ops); } fn emit_main(&mut self, parent: &str, ops: &mut Vec) { @@ -1198,7 +1207,6 @@ impl ClaudeTranslator { self.last_user_turn_id = turn.user_turn_id; self.last_turn_id = Some(turn_id); self.pending_skills.clear(); - self.end_root(event.ts_ms, ops); } fn close_pending_tools( @@ -1288,6 +1296,7 @@ impl ClaudeTranslator { self.last_user_turn_id = turn.user_turn_id; } self.end_root(event.ts_ms, ops); + self.root_open = false; if let Some(pending) = &mut self.pending_emission { pending.clear_after = true; } else { @@ -1450,7 +1459,6 @@ impl AgentTranslator for ClaudeTranslator { root_span_id: self.root_span_id.clone(), parent_span_ids: vec![tool.parent_id], end_ms: Some(end_ms), - error: Some("Session ended before tool completion".into()), ..Default::default() })); } @@ -1483,11 +1491,11 @@ impl AgentTranslator for ClaudeTranslator { root_span_id: self.root_span_id.clone(), parent_span_ids: vec![subagent.parent_span_id], end_ms: Some(end_ms), - error: Some("Session ended before subagent completion".into()), ..Default::default() })); } self.end_root(end_ms, &mut ops); + self.root_open = false; if let Some(pending) = &mut self.pending_emission { pending.clear_after = true; } else { diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 848e6ee8..80236952 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -147,9 +147,9 @@ pub trait AgentTranslator: Send { Ok(Vec::new()) } - /// Finish the logical agent session and defensively close work whose - /// terminal native event never arrived. Called only when the actor itself - /// is shutting down or being retired. + /// Finish a terminal agent invocation and defensively close work whose + /// terminal native event never arrived. Resource retirement uses + /// `checkpoint`, since inactivity alone does not end the native session. fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result> { let _ = ctx; Ok(Vec::new()) From 01da31c51988096d6b57218faab3c740a01de1af Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 00:08:11 +0800 Subject: [PATCH 2/4] Update Claude root reopen regression expectation --- bt-daemon/src/translate/mod.rs | 4 ++-- bt-daemon/tests/claude_translator.rs | 26 +++++++++++++++----------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/bt-daemon/src/translate/mod.rs b/bt-daemon/src/translate/mod.rs index 80236952..52327dea 100644 --- a/bt-daemon/src/translate/mod.rs +++ b/bt-daemon/src/translate/mod.rs @@ -148,8 +148,8 @@ pub trait AgentTranslator: Send { } /// Finish a terminal agent invocation and defensively close work whose - /// terminal native event never arrived. Resource retirement uses - /// `checkpoint`, since inactivity alone does not end the native session. + /// terminal native event never arrived. Resource retirement may use this + /// for agents whose sessions do not resume after an idle period. fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result> { let _ = ctx; Ok(Vec::new()) diff --git a/bt-daemon/tests/claude_translator.rs b/bt-daemon/tests/claude_translator.rs index fd328950..82c1f6e2 100644 --- a/bt-daemon/tests/claude_translator.rs +++ b/bt-daemon/tests/claude_translator.rs @@ -1872,7 +1872,7 @@ fn claude_background_child_completion_extends_the_originating_human_turn() { } #[test] -fn claude_later_turns_extend_an_already_closed_session_root() { +fn claude_activity_after_session_end_starts_a_new_root_generation() { let session = "resumed-root"; let registry = Registry::default_agents(); let mut translator = registry.create("claude-code", session); @@ -1900,16 +1900,20 @@ fn claude_later_turns_extend_an_already_closed_session_root() { ); } let rows = reduce(ops.clone()); - let root = rows - .values() - .find(|row| row.parent_span_ids.is_empty()) - .unwrap(); - assert_eq!(root.start_ms, Some(10)); - assert_eq!(root.end_ms, Some(50)); - assert!(rows + let roots: Vec<_> = rows .values() - .filter(|row| row.name.starts_with("Turn ")) - .all(|row| row.end_ms <= root.end_ms)); + .filter(|row| row.parent_span_ids.is_empty()) + .collect(); + assert_eq!(roots.len(), 2); + let original_root = roots.iter().find(|row| row.start_ms == Some(10)).unwrap(); + let resumed_root = roots.iter().find(|row| row.start_ms == Some(40)).unwrap(); + assert_ne!(original_root.span_id, resumed_root.span_id); + assert_eq!(original_root.end_ms, Some(25)); + assert_eq!(resumed_root.end_ms, None); + let first_turn = rows.values().find(|row| row.name == "Turn 1").unwrap(); + let second_turn = rows.values().find(|row| row.name == "Turn 2").unwrap(); + assert_eq!(first_turn.root_span_id, original_root.span_id); + assert_eq!(second_turn.root_span_id, resumed_root.span_id); // A delayed terminal hook must not shrink the root behind completed work. ops.extend( translator @@ -1919,7 +1923,7 @@ fn claude_later_turns_extend_an_already_closed_session_root() { let rows = reduce(ops); assert_eq!( rows.values() - .find(|row| row.parent_span_ids.is_empty()) + .find(|row| row.span_id == resumed_root.span_id) .unwrap() .end_ms, Some(50) From cc6315d79997de069e822dbb2b443dcd2ad6d09f Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 00:27:49 +0800 Subject: [PATCH 3/4] Keep Claude roots open through daemon shutdown --- bt-daemon/tests/pipeline.rs | 42 +++++++++++++++---------------------- 1 file changed, 17 insertions(+), 25 deletions(-) diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 8a38f31d..96e4686e 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -3681,12 +3681,10 @@ async fn startup_restores_a_durable_head_absent_from_captured_routes() { } #[tokio::test] -async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { - use std::sync::atomic::{AtomicUsize, Ordering}; +async fn clean_shutdown_keeps_claude_roots_resumable_for_every_route() { struct Shared { - started: AtomicUsize, - release: tokio::sync::Semaphore, - delivered: Mutex>, + roots: Mutex>, + terminal_roots: Mutex>, } struct Factory(Arc); struct RouteSink { @@ -3707,15 +3705,18 @@ async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { self.tag = config.tags[0].clone(); } async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { + if ops.iter().any(|op| match op { + SpanOp::Insert(row) | SpanOp::Merge(row) => row.parent_span_ids.is_empty(), + }) { + self.shared.roots.lock().unwrap().insert(self.tag.clone()); + } if ops.iter().any(|op| match op { SpanOp::Insert(row) | SpanOp::Merge(row) => { row.parent_span_ids.is_empty() && row.end_ms.is_some() } }) { - self.shared.started.fetch_add(1, Ordering::SeqCst); - self.shared.release.acquire().await.unwrap().forget(); self.shared - .delivered + .terminal_roots .lock() .unwrap() .insert(self.tag.clone()); @@ -3730,9 +3731,8 @@ async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { let data_dir = tmp.path().join("data"); let socket = test_endpoint(tmp.path()); let shared = Arc::new(Shared { - started: AtomicUsize::new(0), - release: tokio::sync::Semaphore::new(0), - delivered: Mutex::new(HashSet::new()), + roots: Mutex::new(HashSet::new()), + terminal_roots: Mutex::new(HashSet::new()), }); let mut opts = debug_serve_options("test", &data_dir); opts.auth_provider = Some(Arc::new(TestAuthProvider { @@ -3764,28 +3764,20 @@ async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { .unwrap() .flushed ); - let gate = shared.clone(); - let release = tokio::spawn(async move { - let all_started = tokio::time::timeout(Duration::from_secs(1), async { - while gate.started.load(Ordering::SeqCst) < 2 { - tokio::time::sleep(Duration::from_millis(10)).await; - } - }) - .await; - gate.release.add_permits(2); - all_started.unwrap(); - }); shutdown(&socket).await; handle.await.unwrap(); - release.await.unwrap(); assert_eq!( - *shared.delivered.lock().unwrap(), + *shared.roots.lock().unwrap(), HashSet::from(["left".into(), "right".into()]) ); + assert!( + shared.terminal_roots.lock().unwrap().is_empty(), + "daemon shutdown must not mark a still-open Claude session as ended" + ); let (head, tail) = head_and_tail(&data_dir); assert_eq!( head, tail, - "every route durably completed the retirement output" + "every route durably completed the retirement boundary" ); } From 98dfd676d9871b3f8b017716fd1041d0cd3fc186 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 00:42:53 +0800 Subject: [PATCH 4/4] Restore retirement finalization semantics --- bt-daemon/src/pipeline/translation.rs | 14 ++++----- bt-daemon/tests/pipeline.rs | 42 ++++++++++++++++----------- bt-daemon/tests/root_provenance.rs | 33 ++++++++++++--------- 3 files changed, 50 insertions(+), 39 deletions(-) diff --git a/bt-daemon/src/pipeline/translation.rs b/bt-daemon/src/pipeline/translation.rs index e54f8c1d..042bc6d9 100644 --- a/bt-daemon/src/pipeline/translation.rs +++ b/bt-daemon/src/pipeline/translation.rs @@ -549,9 +549,8 @@ impl Producer { ) } JournalRecord::Control(control) => { - // Retirement releases daemon resources; it does not prove - // that the native agent session ended. Keep translator - // state resumable and only finalize explicit terminal runs. + // Retirement exports defensive terminal updates but does not + // close native state when rebuilding for later captured work. if input <= self.committed && matches!(control.kind, ControlKind::Retire) { self.position = input; continue; @@ -561,11 +560,10 @@ impl Producer { .translator .as_mut() .ok_or_else(|| anyhow::anyhow!("translator is missing"))?; - let ops = match control.kind { - ControlKind::Finalize => translator.finalize_at(&ctx, &control.bounds)?, - ControlKind::Checkpoint | ControlKind::Retire => { - translator.checkpoint_at(&ctx, &control.bounds)? - } + let ops = if !matches!(control.kind, ControlKind::Checkpoint) { + translator.finalize_at(&ctx, &control.bounds)? + } else { + translator.checkpoint_at(&ctx, &control.bounds)? }; (ops, self.capture.clone()) } diff --git a/bt-daemon/tests/pipeline.rs b/bt-daemon/tests/pipeline.rs index 96e4686e..8a38f31d 100644 --- a/bt-daemon/tests/pipeline.rs +++ b/bt-daemon/tests/pipeline.rs @@ -3681,10 +3681,12 @@ async fn startup_restores_a_durable_head_absent_from_captured_routes() { } #[tokio::test] -async fn clean_shutdown_keeps_claude_roots_resumable_for_every_route() { +async fn clean_shutdown_delivers_the_shared_retirement_tail_to_every_route() { + use std::sync::atomic::{AtomicUsize, Ordering}; struct Shared { - roots: Mutex>, - terminal_roots: Mutex>, + started: AtomicUsize, + release: tokio::sync::Semaphore, + delivered: Mutex>, } struct Factory(Arc); struct RouteSink { @@ -3705,18 +3707,15 @@ async fn clean_shutdown_keeps_claude_roots_resumable_for_every_route() { self.tag = config.tags[0].clone(); } async fn emit(&mut self, ops: &[SpanOp]) -> anyhow::Result { - if ops.iter().any(|op| match op { - SpanOp::Insert(row) | SpanOp::Merge(row) => row.parent_span_ids.is_empty(), - }) { - self.shared.roots.lock().unwrap().insert(self.tag.clone()); - } if ops.iter().any(|op| match op { SpanOp::Insert(row) | SpanOp::Merge(row) => { row.parent_span_ids.is_empty() && row.end_ms.is_some() } }) { + self.shared.started.fetch_add(1, Ordering::SeqCst); + self.shared.release.acquire().await.unwrap().forget(); self.shared - .terminal_roots + .delivered .lock() .unwrap() .insert(self.tag.clone()); @@ -3731,8 +3730,9 @@ async fn clean_shutdown_keeps_claude_roots_resumable_for_every_route() { let data_dir = tmp.path().join("data"); let socket = test_endpoint(tmp.path()); let shared = Arc::new(Shared { - roots: Mutex::new(HashSet::new()), - terminal_roots: Mutex::new(HashSet::new()), + started: AtomicUsize::new(0), + release: tokio::sync::Semaphore::new(0), + delivered: Mutex::new(HashSet::new()), }); let mut opts = debug_serve_options("test", &data_dir); opts.auth_provider = Some(Arc::new(TestAuthProvider { @@ -3764,20 +3764,28 @@ async fn clean_shutdown_keeps_claude_roots_resumable_for_every_route() { .unwrap() .flushed ); + let gate = shared.clone(); + let release = tokio::spawn(async move { + let all_started = tokio::time::timeout(Duration::from_secs(1), async { + while gate.started.load(Ordering::SeqCst) < 2 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + gate.release.add_permits(2); + all_started.unwrap(); + }); shutdown(&socket).await; handle.await.unwrap(); + release.await.unwrap(); assert_eq!( - *shared.roots.lock().unwrap(), + *shared.delivered.lock().unwrap(), HashSet::from(["left".into(), "right".into()]) ); - assert!( - shared.terminal_roots.lock().unwrap().is_empty(), - "daemon shutdown must not mark a still-open Claude session as ended" - ); let (head, tail) = head_and_tail(&data_dir); assert_eq!( head, tail, - "every route durably completed the retirement boundary" + "every route durably completed the retirement output" ); } diff --git a/bt-daemon/tests/root_provenance.rs b/bt-daemon/tests/root_provenance.rs index c95d7bc0..3613c675 100644 --- a/bt-daemon/tests/root_provenance.rs +++ b/bt-daemon/tests/root_provenance.rs @@ -533,21 +533,26 @@ async fn legacy_acknowledged_spans_receive_late_updates_without_inventing_origin second.flush(session).await; let rows = logs3_rows(&server).await; let late_rows = &rows[initial_rows.len()..]; - for span in [&root, &old_turn] { - let updates: Vec<_> = late_rows - .iter() - .filter(|row| row["span_id"] == *span) - .collect(); - assert!(!updates.is_empty(), "missing late legacy update for {span}"); - assert!(updates.iter().any(|row| row["metrics"]["end"] == 2.0)); - for row in updates { - assert_eq!(row["_is_merge"], true, "historical insert replayed: {row}"); - assert!( - row["context"].get("span_origin").is_none(), - "legacy update must omit provenance entirely, not set null: {row}" - ); - } + let turn_updates: Vec<_> = late_rows + .iter() + .filter(|row| row["span_id"] == old_turn) + .collect(); + assert!( + !turn_updates.is_empty(), + "missing late legacy update for {old_turn}" + ); + assert!(turn_updates.iter().any(|row| row["metrics"]["end"] == 2.0)); + for row in turn_updates { + assert_eq!(row["_is_merge"], true, "historical insert replayed: {row}"); + assert!( + row["context"].get("span_origin").is_none(), + "legacy update must omit provenance entirely, not set null: {row}" + ); } + assert!( + !late_rows.iter().any(|row| row["span_id"] == root), + "a completed turn must leave the session root open for later turns" + ); assert!(late_rows .iter() .any(|row| { row["span_id"] == old_turn && row["output"] == "late legacy answer" }));