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..52327dea 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 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) 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" }));