Separate translator checkpoints from finalization

bt-daemon/src/dispatch.rs

@@ -34,6 +34,7 @@ enum SessionMsg {
     Event(Box<Envelope>, oneshot::Sender<()>),
     Configure(Box<crate::wire::SessionConfig>, oneshot::Sender<()>),
     Flush(oneshot::Sender<u64>),
+    Finalize(oneshot::Sender<u64>),
     Shutdown(oneshot::Sender<()>),
 }
 
@@ -157,6 +158,20 @@ impl Session {
         }
     }
 
+    /// Finalize an invocation-local session and flush its sink. Unlike a
+    /// delivery checkpoint, a managed-run completion is a terminal boundary.
+    pub async fn finalize(&self, timeout: std::time::Duration) -> (bool, u64) {
+        self.touch();
+        let (reply_tx, reply_rx) = oneshot::channel();
+        if self.tx.send(SessionMsg::Finalize(reply_tx)).await.is_err() {
+            return (false, self.counters.queued.load(Ordering::Relaxed));
+        }
+        match tokio::time::timeout(timeout, reply_rx).await {
+            Ok(Ok(pending)) => (pending == 0, pending),
+            _ => (false, self.counters.queued.load(Ordering::Relaxed)),
+        }
+    }
+
     /// Reconfigure the sink before a refresh-triggered flush. Queue ordering
     /// guarantees that all earlier events are processed first.
     pub async fn configure(&self, config: crate::wire::SessionConfig) -> anyhow::Result<()> {
@@ -249,20 +264,29 @@ struct SessionActor {
 enum BatchMode {
     Live,
     Replay,
-    Flush,
+    Checkpoint,
+    TerminalFinalize,
+    ShutdownFinalize,
 }
 
 impl BatchMode {
     fn errors(self) -> (&'static str, &'static str) {
         match self {
             Self::Live => ("translate failed", "sink emit failed"),
             Self::Replay => ("journal replay failed", "sink replay emit failed"),
-            Self::Flush => ("translate flush failed", "sink emit (flush) failed"),
+            Self::Checkpoint => (
+                "translate checkpoint failed",
+                "sink emit (checkpoint) failed",
+            ),
+            Self::TerminalFinalize | Self::ShutdownFinalize => (
+                "translate finalization failed",
+                "sink emit (finalization) failed",
+            ),
         }
     }
 
     fn observes_correlation(self) -> bool {
-        !matches!(self, Self::Flush)
+        !matches!(self, Self::ShutdownFinalize)
     }
 }
 
@@ -291,6 +315,9 @@ impl SessionActor {
                         SessionMsg::Flush(r) => {
                             let _ = r.send(0);
                         }
+                        SessionMsg::Finalize(r) => {
+                            let _ = r.send(0);
+                        }
                         SessionMsg::Shutdown(r) => {
                             let _ = r.send(());
                             break;
@@ -357,11 +384,28 @@ impl SessionActor {
                     let _ = reply.send(());
                 }
                 SessionMsg::Flush(reply) => {
-                    self.drain_flush(&mut translator, &mut sink, &ctx).await;
+                    self.checkpoint_and_flush(&mut translator, &mut sink, &ctx)
+                        .await;
+                    let _ = reply.send(self.counters.queued.load(Ordering::Relaxed));
+                }
+                SessionMsg::Finalize(reply) => {
+                    self.finalize_and_flush(
+                        &mut translator,
+                        &mut sink,
+                        &ctx,
+                        BatchMode::TerminalFinalize,
+                    )
+                    .await;
                     let _ = reply.send(self.counters.queued.load(Ordering::Relaxed));
                 }
                 SessionMsg::Shutdown(reply) => {
-                    self.drain_flush(&mut translator, &mut sink, &ctx).await;
+                    self.finalize_and_flush(
+                        &mut translator,
+                        &mut sink,
+                        &ctx,
+                        BatchMode::ShutdownFinalize,
+                    )
+                    .await;
                     let _ = reply.send(());
                     break;
                 }
@@ -464,16 +508,53 @@ impl SessionActor {
         }
     }
 
-    async fn drain_flush(
+    async fn checkpoint_and_flush(
         &self,
         translator: &mut Box<dyn crate::translate::AgentTranslator>,
         sink: &mut Box<dyn crate::sink::Sink>,
         ctx: &SessionCtx,
     ) {
-        let translated = translator.flush(ctx);
-        let _ = self
-            .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Flush)
+        let translated = translator.checkpoint(ctx);
+        let correlation_changed = self
+            .emit_translator_batches(translator, sink, ctx, translated, BatchMode::Checkpoint)
+            .await;
+        self.persist_correlation_if_changed(correlation_changed)
             .await;
+        self.flush_sink(sink).await;
+    }
+
+    async fn finalize_and_flush(
+        &self,
+        translator: &mut Box<dyn crate::translate::AgentTranslator>,
+        sink: &mut Box<dyn crate::sink::Sink>,
+        ctx: &SessionCtx,
+        mode: BatchMode,
+    ) {
+        let translated = translator.finalize(ctx);
+        let correlation_changed = self
+            .emit_translator_batches(translator, sink, ctx, translated, mode)
+            .await;
+        self.persist_correlation_if_changed(correlation_changed)
+            .await;
+        self.flush_sink(sink).await;
+    }
+
+    async fn persist_correlation_if_changed(&self, changed: bool) {
+        if !changed {
+            return;
+        }
+        if let Err(error) = crate::server::persist_active_parent_snapshot(
+            &self.data_dir,
+            &self.correlation_key,
+            &self.correlation,
+        )
+        .await
+        {
+            self.set_error(error);
+        }
+    }
+
+    async fn flush_sink(&self, sink: &mut Box<dyn crate::sink::Sink>) {
         if let Err(e) = sink.flush().await {
             self.set_error(format!("sink flush failed: {e}"));
         }

bt-daemon/src/server.rs

@@ -478,7 +478,7 @@ impl Daemon {
             let session = { self.sessions.lock().unwrap().get(&key).cloned() };
             if let Some(session) = session {
                 let (flushed, pending) = session
-                    .flush(Duration::from_millis(params.timeout_ms))
+                    .finalize(Duration::from_millis(params.timeout_ms))
                     .await;
                 result.flushed &= flushed;
                 result.pending = result.pending.saturating_add(pending);

bt-daemon/src/translate/antigravity.rs

@@ -516,7 +516,7 @@ impl AgentTranslator for AntigravityTranslator {
         Ok(ops)
     }
 
-    fn flush(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+    fn finalize(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
         let mut ops = Vec::new();
         self.close_pending(self.last_ts_ms, None, &mut ops);
         self.close_turn(self.last_ts_ms, None, &mut ops);

bt-daemon/src/translate/claude.rs

@@ -97,6 +97,7 @@ struct ClaudeTranslator {
     git: Arc<GitMetadataCache>,
     current_cwd: Option<String>,
     last_turn_cwd: Option<String>,
+    last_ts_ms: i64,
 }
 
 impl ClaudeTranslator {
@@ -126,6 +127,7 @@ impl ClaudeTranslator {
             git,
             current_cwd: None,
             last_turn_cwd: None,
+            last_ts_ms: 0,
         }
     }
 
@@ -700,6 +702,7 @@ impl ClaudeTranslator {
 
 impl AgentTranslator for ClaudeTranslator {
     fn handle(&mut self, event: &Envelope, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+        self.last_ts_ms = self.last_ts_ms.max(event.ts_ms);
         anyhow::ensure!(
             self.pending_emission.is_none(),
             "Claude translator has pending catch-up work; drain it before handling another event"
@@ -778,8 +781,47 @@ impl AgentTranslator for ClaudeTranslator {
         Ok(Some(ops))
     }
 
-    fn flush(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
-        Ok(Vec::new())
+    fn finalize(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+        let end_ms = self.last_ts_ms;
+        let mut ops = Vec::new();
+        for (_, tool) in self.pending_tools.drain() {
+            ops.push(SpanOp::Merge(SpanRow {
+                span_id: tool.span_id,
+                root_span_id: self.root_span_id.clone(),
+                end_ms: Some(end_ms),
+                error: Some("Session ended before tool completion".into()),
+                ..Default::default()
+            }));
+        }
+        if let Some(turn) = self.turn.take() {
+            ops.push(SpanOp::Merge(SpanRow {
+                span_id: turn.id,
+                root_span_id: self.root_span_id.clone(),
+                end_ms: Some(end_ms),
+                error: Some("Session ended before turn completion".into()),
+                ..Default::default()
+            }));
+        }
+        for (_, subagent) in self.subagents.drain() {
+            ops.push(SpanOp::Merge(SpanRow {
+                span_id: subagent.span_id,
+                root_span_id: self.root_span_id.clone(),
+                end_ms: Some(end_ms),
+                error: Some("Session ended before subagent completion".into()),
+                ..Default::default()
+            }));
+        }
+        if self.root_open && !self.root_ended {
+            self.root_ended = true;
+            ops.push(SpanOp::Merge(SpanRow {
+                span_id: self.session_span_id.clone(),
+                root_span_id: self.root_span_id.clone(),
+                end_ms: Some(end_ms),
+                ..Default::default()
+            }));
+        }
+        self.release_terminal_state();
+        Ok(ops)
     }
 }
 

bt-daemon/src/translate/codex.rs

@@ -145,9 +145,10 @@ enum PendingWork {
         through_ms: Option<i64>,
         after: DeferredHook,
     },
-    Flush {
+    CatchUp {
         paths: Vec<String>,
         next_path: usize,
+        finalize: bool,
     },
 }
 
@@ -259,24 +260,30 @@ impl AgentTranslator for CodexTranslator {
                     });
                 }
             }
-            PendingWork::Flush {
+            PendingWork::CatchUp {
                 paths,
                 mut next_path,
+                finalize,
             } => {
                 while next_path < paths.len() {
                     let path = &paths[next_path];
                     if self.catch_up_chunk(path, 0, None, &mut ops) {
-                        if let Some(mut scope) = self.scopes.remove(path) {
-                            self.close_dangling(&mut scope, None, &mut ops);
-                            self.scopes.insert(path.clone(), scope);
+                        if finalize {
+                            if let Some(mut scope) = self.scopes.remove(path) {
+                                self.close_dangling(&mut scope, None, &mut ops);
+                                self.scopes.insert(path.clone(), scope);
+                            }
                         }
                         next_path += 1;
                     }
                     // Return after any completed scope or a bounded partial read.
-                    // This keeps a flush over many scopes bounded as well.
+                    // This keeps catch-up over many scopes bounded as well.
                     if !ops.is_empty() || next_path < paths.len() {
-                        self.pending = (next_path < paths.len())
-                            .then_some(PendingWork::Flush { paths, next_path });
+                        self.pending = (next_path < paths.len()).then_some(PendingWork::CatchUp {
+                            paths,
+                            next_path,
+                            finalize,
+                        });
                         break;
                     }
                 }
@@ -285,25 +292,35 @@ impl AgentTranslator for CodexTranslator {
         Ok(Some(ops))
     }
 
-    fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+    fn checkpoint(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+        self.start_catch_up(ctx, false)
+    }
+
+    fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+        self.start_catch_up(ctx, true)
+    }
+}
+
+impl CodexTranslator {
+    fn start_catch_up(&mut self, ctx: &SessionCtx, finalize: bool) -> anyhow::Result<Vec<SpanOp>> {
         anyhow::ensure!(
             self.pending.is_none(),
-            "Codex translator has pending catch-up work; drain it before flushing"
+            "Codex translator has pending catch-up work; drain it before checkpointing"
         );
-        // Re-read each scope to catch a late task_complete, then close dangling.
+        // Re-read each scope to catch a late task_complete. Finalization also
+        // closes dangling work whose terminal native event never arrived.
         let paths: Vec<String> = self.scopes.keys().cloned().collect();
         if paths.is_empty() {
             return Ok(Vec::new());
         }
-        self.pending = Some(PendingWork::Flush {
+        self.pending = Some(PendingWork::CatchUp {
             paths,
             next_path: 0,
+            finalize,
         });
         Ok(self.drain_pending(ctx)?.unwrap_or_default())
     }
-}
 
-impl CodexTranslator {
     fn ensure_main_scope(&mut self, path: &str) {
         if self.scopes.contains_key(path) {
             return;

bt-daemon/src/translate/debug.rs

@@ -74,7 +74,7 @@ impl AgentTranslator for DebugTranslator {
         Ok(ops)
     }
 
-    fn flush(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+    fn finalize(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
         Ok(Vec::new())
     }
 }

bt-daemon/src/translate/mod.rs

@@ -92,18 +92,34 @@ pub trait AgentTranslator: Send {
     /// Handle one event, returning span ops to emit.
     fn handle(&mut self, event: &Envelope, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>>;
 
-    /// Continue bounded work started by [`Self::handle`] or [`Self::flush`].
+    /// Continue bounded work started by [`Self::handle`], [`Self::checkpoint`],
+    /// or [`Self::finalize`].
     /// `Some` means the caller must emit this batch and call again; `None`
     /// means the translator is fully caught up.
     fn drain_pending(&mut self, _ctx: &SessionCtx) -> anyhow::Result<Option<Vec<SpanOp>>> {
         Ok(None)
     }
 
-    /// Emit any pending spans (e.g. close dangling turns) at flush/shutdown.
-    fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+    /// Catch up externally buffered observations without ending the logical
+    /// agent session. Delivery barriers call this before flushing the sink.
+    fn checkpoint(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+        let _ = ctx;
+        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.
+    fn finalize(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
         let _ = ctx;
         Ok(Vec::new())
     }
+
+    /// Backward-compatible terminal flush used by transcript import callers.
+    /// Live delivery barriers use [`Self::checkpoint`] instead.
+    fn flush(&mut self, ctx: &SessionCtx) -> anyhow::Result<Vec<SpanOp>> {
+        self.finalize(ctx)
+    }
 }
 
 /// Builds translator instances for a given `source`.