fix: bound daemon translator and sink memory

bt-daemon/src/dispatch.rs

@@ -266,15 +266,16 @@ impl SessionActor {
                         ctx.config = Some(cfg.clone());
                         self.refresh_permalink(sink.as_ref());
                     }
-                    match translator.handle(&env, &ctx) {
-                        Ok(ops) => match sink.emit(&ops).await {
-                            Ok(n) => {
-                                self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
-                            }
-                            Err(e) => self.set_error(format!("sink emit failed: {e}")),
-                        },
-                        Err(e) => self.set_error(format!("translate failed: {e}")),
-                    }
+                    let translated = translator.handle(&env, &ctx);
+                    self.emit_translator_batches(
+                        &mut translator,
+                        &mut sink,
+                        &ctx,
+                        translated,
+                        "translate failed",
+                        "sink emit failed",
+                    )
+                    .await;
                     self.counters.queued.fetch_sub(1, Ordering::Relaxed);
                 }
                 SessionMsg::Configure(config, reply) => {
@@ -296,6 +297,44 @@ impl SessionActor {
         }
     }
 
+    /// Emit a translator result and every bounded continuation it schedules.
+    /// A continuation may be empty (for example, irrelevant rollout rows), so
+    /// only `None` signals that the translator is fully caught up.
+    async fn emit_translator_batches(
+        &self,
+        translator: &mut Box<dyn crate::translate::AgentTranslator>,
+        sink: &mut Box<dyn crate::sink::Sink>,
+        ctx: &SessionCtx,
+        first: anyhow::Result<Vec<crate::translate::SpanOp>>,
+        translate_error: &str,
+        emit_error: &str,
+    ) {
+        let mut next = match first {
+            Ok(ops) => Some(ops),
+            Err(e) => {
+                self.set_error(format!("{translate_error}: {e}"));
+                return;
+            }
+        };
+        while let Some(ops) = next {
+            if !ops.is_empty() {
+                match sink.emit(&ops).await {
+                    Ok(n) => {
+                        self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
+                    }
+                    Err(e) => self.set_error(format!("{emit_error}: {e}")),
+                }
+            }
+            next = match translator.drain_pending(ctx) {
+                Ok(next) => next,
+                Err(e) => {
+                    self.set_error(format!("{translate_error}: {e}"));
+                    None
+                }
+            };
+        }
+    }
+
     /// Stream the journal through the translator, emitting each entry's spans
     /// as they are produced. Nothing is accumulated across entries: peak
     /// memory is one journal entry and the ops it yields, so recovering a
@@ -336,22 +375,16 @@ impl SessionActor {
                 continue;
             }
             let env = crate::journal::envelope_from_redacted(entry);
-            let ops = match translator.handle(&env, ctx) {
-                Ok(ops) => ops,
-                Err(e) => {
-                    self.set_error(format!("journal replay failed: {e}"));
-                    continue;
-                }
-            };
-            if ops.is_empty() {
-                continue;
-            }
-            match sink.emit(&ops).await {
-                Ok(n) => {
-                    self.counters.spans_emitted.fetch_add(n, Ordering::Relaxed);
-                }
-                Err(e) => self.set_error(format!("sink replay emit failed: {e}")),
-            }
+            let translated = translator.handle(&env, ctx);
+            self.emit_translator_batches(
+                translator,
+                sink,
+                ctx,
+                translated,
+                "journal replay failed",
+                "sink replay emit failed",
+            )
+            .await;
         }
     }
 
@@ -361,14 +394,16 @@ impl SessionActor {
         sink: &mut Box<dyn crate::sink::Sink>,
         ctx: &SessionCtx,
     ) {
-        match translator.flush(ctx) {
-            Ok(ops) => {
-                if let Err(e) = sink.emit(&ops).await {
-                    self.set_error(format!("sink emit (flush) failed: {e}"));
-                }
-            }
-            Err(e) => self.set_error(format!("translate flush failed: {e}")),
-        }
+        let translated = translator.flush(ctx);
+        self.emit_translator_batches(
+            translator,
+            sink,
+            ctx,
+            translated,
+            "translate flush failed",
+            "sink emit (flush) failed",
+        )
+        .await;
         if let Err(e) = sink.flush().await {
             self.set_error(format!("sink flush failed: {e}"));
         }

bt-daemon/src/lib.rs

@@ -73,7 +73,7 @@ pub struct ServeArgs {
     /// Retire a session's in-memory state after this many seconds without
     /// traffic. A later event rebuilds it from the journal. 0 disables
     /// retirement.
-    #[arg(long, default_value_t = 300)]
+    #[arg(long, default_value_t = 30)]
     pub session_idle_timeout_secs: u64,
 }
 
@@ -762,7 +762,14 @@ pub async fn import_transcripts(
         let entries = tail
             .poll(true)
             .with_context(|| format!("import transcript {}", file.display()))?;
+        let session_ids = entries
+            .iter()
+            .map(|entry| entry.session_id.clone())
+            .collect::<std::collections::HashSet<_>>();
         processor.process(entries).await?;
+        for session_id in session_ids {
+            processor.finish_session(&session_id).await?;
+        }
     }
     processor.finish().await
 }
@@ -822,6 +829,17 @@ impl ImportProcessor {
                 live.ctx.config = Some(cfg.clone());
             }
             let ops = live.translator.handle(&env, &live.ctx)?;
+            Self::emit_translator_batches(live, ops).await?;
+        }
+        Ok(())
+    }
+
+    async fn emit_translator_batches(
+        live: &mut ImportLive,
+        first: Vec<SpanOp>,
+    ) -> anyhow::Result<()> {
+        let mut next = Some(first);
+        while let Some(ops) = next {
             // Imports can contain tens of thousands of SDK log commands. Bound the
             // number queued between drains without serializing one network flush
             // for every native turn boundary.
@@ -834,18 +852,28 @@ impl ImportProcessor {
                     live.pending_ops = 0;
                 }
             }
+            next = live.translator.drain_pending(&live.ctx)?;
         }
         Ok(())
     }
 
     async fn finish(self) -> anyhow::Result<()> {
         for (_sid, mut live) in self.sessions {
             let ops = live.translator.flush(&live.ctx)?;
-            live.sink.emit(&ops).await?;
+            Self::emit_translator_batches(&mut live, ops).await?;
             live.sink.flush().await?;
         }
         Ok(())
     }
+
+    async fn finish_session(&mut self, session_id: &str) -> anyhow::Result<()> {
+        let Some(mut live) = self.sessions.remove(session_id) else {
+            return Ok(());
+        };
+        let ops = live.translator.flush(&live.ctx)?;
+        Self::emit_translator_batches(&mut live, ops).await?;
+        live.sink.flush().await
+    }
 }
 
 /// Build a Phase-1 debug [`ServeOptions`]: debug translator registry + a debug
@@ -916,6 +944,18 @@ mod tests {
         args: ImportArgs,
     }
 
+    #[derive(Debug, Parser)]
+    struct ServeCli {
+        #[command(flatten)]
+        args: ServeArgs,
+    }
+
+    #[test]
+    fn serve_defaults_to_short_journal_backed_session_retirement() {
+        let args = ServeCli::try_parse_from(["test"]).unwrap().args;
+        assert_eq!(args.session_idle_timeout_secs, 30);
+    }
+
     #[test]
     fn import_args_accept_multiple_sessions_or_all() {
         let explicit = ImportCli::try_parse_from([