diff --git a/bt-daemon/src/server/ingress.rs b/bt-daemon/src/server/ingress.rs
index d2ae453d..47615a32 100644
--- a/bt-daemon/src/server/ingress.rs
+++ b/bt-daemon/src/server/ingress.rs
@@ -3,6 +3,169 @@
 //! the journals as overflow when the queue is full.
 
 use super::*;
+use std::collections::VecDeque;
+
+/// Live startup captures remain ordered and bounded. Evicted entries are
+/// already durable; their source journal's high-water mark is replayed first.
+pub(super) struct StartupBuffer {
+    pub(super) events: VecDeque<PendingEvent>,
+    pub(super) evicted: HashMap<PathBuf, EvictedRange>,
+}
+
+#[derive(Clone, Copy, Debug)]
+pub(super) struct EvictedRange {
+    pub(super) from: u64,
+    pub(super) through: u64,
+}
+
+impl StartupBuffer {
+    pub(super) fn new() -> Self {
+        Self {
+            events: VecDeque::new(),
+            evicted: HashMap::new(),
+        }
+    }
+
+    fn push(&mut self, data_dir: &std::path::Path, event: PendingEvent) {
+        if self.events.len() == INGRESS_QUEUE_CAPACITY {
+            if let Some(oldest) = self.events.pop_front() {
+                let path = journal::source_journal_path(
+                    data_dir,
+                    &oldest.env.source,
+                    &oldest.env.session_id,
+                );
+                self.evicted
+                    .entry(path)
+                    .and_modify(|range| range.through = range.through.max(oldest.journal_through))
+                    .or_insert(EvictedRange {
+                        from: oldest.replay_through,
+                        through: oldest.journal_through,
+                    });
+            }
+        }
+        self.events.push_back(event);
+    }
+}
+
+pub(super) enum StartupWork {
+    Evicted(HashMap<PathBuf, EvictedRange>),
+    Event(Box<PendingEvent>),
+    Finished,
+}
+
+impl Daemon {
+    pub(super) fn next_startup_work(&self) -> StartupWork {
+        let mut state = self.startup_buffer.lock();
+        let Some(buffer) = state.as_mut() else {
+            return StartupWork::Finished;
+        };
+        if !buffer.evicted.is_empty() {
+            return StartupWork::Evicted(std::mem::take(&mut buffer.evicted));
+        }
+        if let Some(event) = buffer.events.pop_front() {
+            return StartupWork::Event(Box::new(event));
+        }
+        *state = None;
+        StartupWork::Finished
+    }
+}
+
+pub(super) async fn drain_startup_buffer(daemon: &Arc<Daemon>) -> anyhow::Result<()> {
+    let mut cursors = HashMap::new();
+    loop {
+        match daemon.next_startup_work() {
+            StartupWork::Evicted(bounds) => {
+                replay_evicted_startup(daemon, &mut cursors, &bounds).await?
+            }
+            StartupWork::Event(event) => dispatch_ingress_event(daemon, *event).await,
+            StartupWork::Finished => return Ok(()),
+        }
+    }
+}
+
+pub(super) struct StartupCursor {
+    reader: journal::JournalReader,
+    position: u64,
+}
+
+/// Advance each evicted journal range from its previous position, retaining
+/// the reader's Pi delta context between advances. Older records are decoded
+/// only to initialize that context; they are not dispatched again.
+pub(super) async fn replay_evicted_startup(
+    daemon: &Arc<Daemon>,
+    cursors: &mut HashMap<PathBuf, StartupCursor>,
+    bounds: &HashMap<PathBuf, EvictedRange>,
+) -> anyhow::Result<()> {
+    for (path, range) in bounds {
+        anyhow::ensure!(
+            journal::JournalReader::recorded_len(path).await >= range.through,
+            "evicted startup journal is truncated"
+        );
+        if !cursors.contains_key(path) {
+            let reader = journal::JournalReader::open(path, range.from)
+                .await?
+                .ok_or_else(|| anyhow::anyhow!("evicted startup journal disappeared"))?;
+            cursors.insert(
+                path.clone(),
+                StartupCursor {
+                    reader,
+                    position: 0,
+                },
+            );
+        }
+        let cursor = cursors
+            .get_mut(path)
+            .ok_or_else(|| anyhow::anyhow!("startup cursor missing"))?;
+        anyhow::ensure!(
+            cursor.position <= range.from,
+            "evicted startup ranges overlap"
+        );
+        cursor.reader.extend(range.from);
+        while cursor.position < range.from {
+            let entry =
+                cursor.reader.next_record().await?.ok_or_else(|| {
+                    anyhow::anyhow!("evicted startup range begins inside a record")
+                })?;
+            cursor.position = entry.through;
+        }
+        anyhow::ensure!(
+            cursor.position == range.from,
+            "invalid evicted startup boundary"
+        );
+        cursor.reader.extend(range.through);
+        while cursor.position < range.through {
+            let before = cursor.position;
+            let entry = cursor
+                .reader
+                .next_record()
+                .await?
+                .ok_or_else(|| anyhow::anyhow!("evicted startup range ends inside a record"))?;
+            cursor.position = entry.through;
+            if let journal::JournalRecord::Event(redacted) = entry.record {
+                let mut env = journal::envelope_from_redacted(redacted);
+                let Some(source) = daemon.translators.canonical_source(&env.source) else {
+                    continue;
+                };
+                env.source = source.to_owned();
+                dispatch_ingress_event(
+                    daemon,
+                    PendingEvent {
+                        env,
+                        replay_through: before,
+                        journal_through: entry.through,
+                        settings_through: None,
+                    },
+                )
+                .await;
+            }
+        }
+        anyhow::ensure!(
+            cursor.position == range.through,
+            "invalid evicted startup end"
+        );
+    }
+    Ok(())
+}
 
 pub(super) enum IngressMsg {
     Event(Box<PendingEvent>),
@@ -14,8 +177,19 @@ pub(super) enum IngressMsg {
 /// journals become the overflow queue and the worker catches them up in place.
 pub(super) const INGRESS_QUEUE_CAPACITY: usize = 64;
 
-pub(super) fn spawn_ingress_worker(daemon: Arc<Daemon>, mut rx: mpsc::Receiver<IngressMsg>) {
+pub(super) fn spawn_ingress_worker(
+    daemon: Arc<Daemon>,
+    mut rx: mpsc::Receiver<IngressMsg>,
+    startup: Option<oneshot::Receiver<()>>,
+) {
     tokio::spawn(async move {
+        if let Some(startup) = startup {
+            if startup.await.is_err() {
+                tracing::error!("startup recovery failed; stopping ingress");
+                daemon.trigger_shutdown();
+                return;
+            }
+        }
         loop {
             if daemon.ingress_overflow.swap(false, Ordering::AcqRel) {
                 recover_ingress_overflow(&daemon).await;
@@ -63,6 +237,10 @@ pub(super) async fn dispatch_ingress_event(daemon: &Arc<Daemon>, event: PendingE
         return;
     }
 
+    // Startup discovery rebuilds these observations before this worker is
+    // released, so a historical event cannot overwrite a newer capture.
+    daemon.remember_claude_transcripts(&event.env);
+
     let journal_through = event.journal_through;
     // A child may arrive immediately after a parent's tool hook. Catch prior
     // session actors up before resolving a new session, entirely in the daemon.
@@ -323,15 +501,21 @@ impl Daemon {
             .append_to_journal(&mut env)
             .await
             .map_err(|error| format!("journal failed: {error}"))?;
-        self.remember_claude_transcripts(&env);
-        match self
-            .ingress_tx
-            .try_send(IngressMsg::Event(Box::new(PendingEvent {
-                env,
-                replay_through,
-                journal_through,
-                settings_through: None,
-            }))) {
+        let event = PendingEvent {
+            env,
+            replay_through,
+            journal_through,
+            settings_through: None,
+        };
+        let event = {
+            let mut state = self.startup_buffer.lock();
+            if let Some(buffer) = state.as_mut() {
+                buffer.push(&self.data_dir, event);
+                return Ok(true);
+            }
+            event
+        };
+        match self.ingress_tx.try_send(IngressMsg::Event(Box::new(event))) {
             Ok(()) => {}
             Err(mpsc::error::TrySendError::Full(_)) => {
                 self.ingress_overflow.store(true, Ordering::Release);
diff --git a/bt-daemon/src/server/maintenance.rs b/bt-daemon/src/server/maintenance.rs
index 61e9d45c..16107ef2 100644
--- a/bt-daemon/src/server/maintenance.rs
+++ b/bt-daemon/src/server/maintenance.rs
@@ -70,6 +70,17 @@ async fn collect_partitions(daemon: &Daemon, age: Duration) {
         if tokio::fs::try_exists(&wal).await.unwrap_or(true) && !old(&wal, age).await {
             continue;
         }
+        // Scanning a large journal must not hold the capture lock. The length
+        // and route set are revalidated under that lock before deletion.
+        let Ok(partition) = crate::pipeline::journal::SpanJournal::open(&storage).await else {
+            continue;
+        };
+        let Ok(routes) = partition.routes().await else {
+            continue;
+        };
+        if !captured_routes_registered(&daemon.data_dir, &path, recorded, &routes).await {
+            continue;
+        }
         let lock = daemon.session_lock(&first.source, &first.session_id);
         let _guard = lock.lock().await;
         let namespace = crate::ids::session_namespace(&first.source, &first.session_id);
@@ -81,19 +92,24 @@ async fn collect_partitions(daemon: &Daemon, age: Duration) {
         if journal::JournalReader::recorded_len(&path).await != recorded {
             continue;
         }
-        let Ok(partition) = crate::pipeline::journal::SpanJournal::open(&storage).await else {
+        let Ok(current_routes) = partition.routes().await else {
             continue;
         };
-        let Ok(routes) = partition.routes().await else {
+        let (Ok(current_routes), Ok(scanned_routes)) = (
+            serde_json::to_value(&current_routes),
+            serde_json::to_value(&routes),
+        ) else {
             continue;
         };
+        if current_routes != scanned_routes {
+            continue;
+        }
         let pending = daemon.pending_sessions.lock().values().any(|state| {
             state.events.iter().any(|event| {
                 event.env.source == first.source && event.env.session_id == first.session_id
             })
         });
-        if pending || !captured_routes_registered(&daemon.data_dir, &path, recorded, &routes).await
-        {
+        if pending {
             continue;
         }
 
@@ -421,6 +437,316 @@ mod tests {
             source_version: None,
         }
     }
+
+    #[tokio::test]
+    async fn capture_is_durable_while_startup_dispatch_is_blocked() {
+        let tmp = tempfile::tempdir().unwrap();
+        let (ready, startup) = oneshot::channel();
+        let daemon = Daemon::new_with_startup(
+            debug_serve_options("test", tmp.path()),
+            tmp.path().into(),
+            Some(startup),
+        );
+        let env = event("during-startup", SessionRoute::default());
+        assert!(
+            tokio::time::timeout(Duration::from_secs(1), daemon.capture_event(env.clone()))
+                .await
+                .unwrap()
+                .unwrap()
+        );
+        let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id);
+        assert!(tokio::fs::read_to_string(path)
+            .await
+            .unwrap()
+            .contains("during-startup"));
+        assert!(daemon.translations.lock().is_empty());
+        drain_startup_buffer(&daemon).await.unwrap();
+        ready.send(()).unwrap();
+        tokio::time::timeout(Duration::from_secs(3), daemon.ingress_barrier())
+            .await
+            .unwrap();
+        assert!(!daemon.translations.lock().is_empty());
+        drain_all(&daemon).await;
+    }
+
+    #[tokio::test]
+    async fn startup_recovery_dispatches_old_session_before_new_capture() {
+        let tmp = tempfile::tempdir().unwrap();
+        let env = event("resuming", SessionRoute::default());
+        let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id);
+        let mut writer = JournalWriter::open_path(&path).await.unwrap();
+        let old_end = writer.append(&env).await.unwrap();
+        drop(writer);
+        let (ready, startup) = oneshot::channel();
+        let daemon = Daemon::new_with_startup(
+            debug_serve_options("test", tmp.path()),
+            tmp.path().into(),
+            Some(startup),
+        );
+        let mut next = env.clone();
+        next.event = "next".into();
+        next.ts_ms = 2;
+        assert!(daemon.capture_event(next).await.unwrap());
+        let recovered = discover_unprocessed_journals(&daemon).await;
+        assert_eq!(recovered.len(), 1);
+        assert_eq!(recovered[0].journal_through, old_end);
+        for event in recovered {
+            dispatch_ingress_event(&daemon, event).await;
+        }
+        drain_startup_buffer(&daemon).await.unwrap();
+        ready.send(()).unwrap();
+        tokio::time::timeout(Duration::from_secs(3), daemon.ingress_barrier())
+            .await
+            .unwrap();
+        let source = daemon.translations.lock().values().next().unwrap().clone();
+        source.barrier().await.unwrap();
+        let storage = tmp.path().join("pipeline").join(format!(
+            "{}.sqlite",
+            crate::ids::session_storage_id(&env.source, &env.session_id)
+        ));
+        let partition = crate::pipeline::journal::SpanJournal::open(&storage)
+            .await
+            .unwrap();
+        let through = partition.translated_through().await.unwrap();
+        assert!(through > old_end);
+        assert_eq!(through, journal::JournalReader::recorded_len(&path).await);
+        drain_all(&daemon).await;
+    }
+
+    #[tokio::test]
+    async fn startup_overflow_replays_oldest_events_before_the_buffered_tail() {
+        let tmp = tempfile::tempdir().unwrap();
+        let (ready, startup) = oneshot::channel();
+        let daemon = Daemon::new_with_startup(
+            debug_serve_options("test", tmp.path()),
+            tmp.path().into(),
+            Some(startup),
+        );
+        let env = event("overflow", SessionRoute::default());
+        let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id);
+        let mut evicted_end = 0;
+        for i in 0..(INGRESS_QUEUE_CAPACITY + 6) {
+            let mut next = env.clone();
+            next.ts_ms = i as i64;
+            next.payload = serde_json::json!({"index": i});
+            assert!(daemon.capture_event(next).await.unwrap());
+            if i == 5 {
+                evicted_end = journal::JournalReader::recorded_len(&path).await;
+            }
+        }
+        {
+            let state = daemon.startup_buffer.lock();
+            let buffer = state.as_ref().unwrap();
+            assert_eq!(buffer.events.len(), INGRESS_QUEUE_CAPACITY);
+            assert_eq!(
+                buffer.evicted.get(&path).map(|range| range.through),
+                Some(evicted_end)
+            );
+            assert_eq!(buffer.events.front().unwrap().env.payload["index"], 6);
+        }
+        drain_startup_buffer(&daemon).await.unwrap();
+        ready.send(()).unwrap();
+        daemon.ingress_barrier().await;
+        let source = daemon.translations.lock().values().next().unwrap().clone();
+        source.barrier().await.unwrap();
+        let storage = tmp.path().join("pipeline").join(format!(
+            "{}.sqlite",
+            crate::ids::session_storage_id(&env.source, &env.session_id)
+        ));
+        let db = rusqlite::Connection::open(storage).unwrap();
+        let count: u64 = db
+            .query_row(
+                "SELECT count(*) FROM batches WHERE committed=1",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(count, (INGRESS_QUEUE_CAPACITY + 6) as u64);
+        let partition = crate::pipeline::journal::SpanJournal::open(
+            &tmp.path().join("pipeline").join(format!(
+                "{}.sqlite",
+                crate::ids::session_storage_id(&env.source, &env.session_id)
+            )),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            partition.translated_through().await.unwrap(),
+            journal::JournalReader::recorded_len(&path).await
+        );
+        drop(db);
+        drain_all(&daemon).await;
+    }
+
+    #[tokio::test]
+    async fn startup_overflow_advances_again_while_evicted_records_are_catching_up() {
+        let tmp = tempfile::tempdir().unwrap();
+        let (ready, startup) = oneshot::channel();
+        let daemon = Daemon::new_with_startup(
+            debug_serve_options("test", tmp.path()),
+            tmp.path().into(),
+            Some(startup),
+        );
+        let env = event("chasing", SessionRoute::default());
+        let path = journal::source_journal_path(tmp.path(), &env.source, &env.session_id);
+        for i in 0..(INGRESS_QUEUE_CAPACITY + 2) {
+            let mut next = env.clone();
+            next.ts_ms = i as i64;
+            next.payload = serde_json::json!({"index": i});
+            assert!(daemon.capture_event(next).await.unwrap());
+        }
+        let StartupWork::Evicted(first) = daemon.next_startup_work() else {
+            panic!("the oldest events were not evicted");
+        };
+        let first_end = first[&path].through;
+        for i in INGRESS_QUEUE_CAPACITY + 2..INGRESS_QUEUE_CAPACITY + 7 {
+            let mut next = env.clone();
+            next.ts_ms = i as i64;
+            next.payload = serde_json::json!({"index": i});
+            assert!(daemon.capture_event(next).await.unwrap());
+        }
+        {
+            let state = daemon.startup_buffer.lock();
+            let buffer = state.as_ref().unwrap();
+            assert_eq!(buffer.events.len(), INGRESS_QUEUE_CAPACITY);
+            assert!(buffer.evicted[&path].through > first_end);
+            assert_eq!(buffer.events.front().unwrap().env.payload["index"], 7);
+        }
+        let mut cursors = HashMap::new();
+        replay_evicted_startup(&daemon, &mut cursors, &first)
+            .await
+            .unwrap();
+        let StartupWork::Evicted(second) = daemon.next_startup_work() else {
+            panic!("new evictions did not advance recovery");
+        };
+        replay_evicted_startup(&daemon, &mut cursors, &second)
+            .await
+            .unwrap();
+        drain_startup_buffer(&daemon).await.unwrap();
+        ready.send(()).unwrap();
+        daemon.ingress_barrier().await;
+        let source = daemon.translations.lock().values().next().unwrap().clone();
+        source.barrier().await.unwrap();
+        let storage = tmp.path().join("pipeline").join(format!(
+            "{}.sqlite",
+            crate::ids::session_storage_id(&env.source, &env.session_id)
+        ));
+        let db = rusqlite::Connection::open(storage).unwrap();
+        let count: u64 = db
+            .query_row(
+                "SELECT count(*) FROM batches WHERE committed=1",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(count, (INGRESS_QUEUE_CAPACITY + 7) as u64);
+        drop(db);
+        drain_all(&daemon).await;
+    }
+
+    #[tokio::test]
+    async fn startup_overflow_tracks_each_source_journal_independently() {
+        let tmp = tempfile::tempdir().unwrap();
+        let (ready, startup) = oneshot::channel();
+        let daemon = Daemon::new_with_startup(
+            debug_serve_options("test", tmp.path()),
+            tmp.path().into(),
+            Some(startup),
+        );
+        for i in 0..(INGRESS_QUEUE_CAPACITY + 6) {
+            let session = if i % 2 == 0 { "first" } else { "second" };
+            let mut env = event(session, SessionRoute::default());
+            env.ts_ms = i as i64;
+            env.payload = serde_json::json!({"index": i});
+            assert!(daemon.capture_event(env).await.unwrap());
+        }
+        {
+            let state = daemon.startup_buffer.lock();
+            let buffer = state.as_ref().unwrap();
+            assert_eq!(buffer.evicted.len(), 2);
+            assert_eq!(buffer.events.len(), INGRESS_QUEUE_CAPACITY);
+        }
+        drain_startup_buffer(&daemon).await.unwrap();
+        ready.send(()).unwrap();
+        daemon.ingress_barrier().await;
+        let sources: Vec<_> = daemon.translations.lock().values().cloned().collect();
+        for source in sources {
+            source.barrier().await.unwrap();
+        }
+        for session in ["first", "second"] {
+            let storage = tmp.path().join("pipeline").join(format!(
+                "{}.sqlite",
+                crate::ids::session_storage_id("debug", session)
+            ));
+            let db = rusqlite::Connection::open(storage).unwrap();
+            let count: u64 = db
+                .query_row(
+                    "SELECT count(*) FROM batches WHERE committed=1",
+                    [],
+                    |row| row.get(0),
+                )
+                .unwrap();
+            assert_eq!(count, ((INGRESS_QUEUE_CAPACITY + 6) / 2) as u64);
+        }
+        drain_all(&daemon).await;
+    }
+
+    #[tokio::test]
+    async fn startup_overflow_keeps_pi_delta_context_across_recovery_advances() {
+        let tmp = tempfile::tempdir().unwrap();
+        let (ready, startup) = oneshot::channel();
+        let daemon = Daemon::new_with_startup(
+            debug_serve_options("test", tmp.path()),
+            tmp.path().into(),
+            Some(startup),
+        );
+        let mut env = event("pi-overflow", SessionRoute::default());
+        env.source = "pi".into();
+        env.event = "context".into();
+        for i in 0..(INGRESS_QUEUE_CAPACITY + 2) {
+            env.ts_ms = i as i64;
+            env.payload = serde_json::json!({"event":{"messages":[{"role":"user","content":"shared"},{"role":"assistant","content":i}]}});
+            assert!(daemon.capture_event(env.clone()).await.unwrap());
+        }
+        let StartupWork::Evicted(first) = daemon.next_startup_work() else {
+            panic!("missing first overflow")
+        };
+        let mut cursors = HashMap::new();
+        replay_evicted_startup(&daemon, &mut cursors, &first)
+            .await
+            .unwrap();
+        for i in INGRESS_QUEUE_CAPACITY + 2..INGRESS_QUEUE_CAPACITY + 4 {
+            env.ts_ms = i as i64;
+            env.payload = serde_json::json!({"event":{"messages":[{"role":"user","content":"shared"},{"role":"assistant","content":i}]}});
+            assert!(daemon.capture_event(env.clone()).await.unwrap());
+        }
+        let StartupWork::Evicted(second) = daemon.next_startup_work() else {
+            panic!("missing second overflow")
+        };
+        replay_evicted_startup(&daemon, &mut cursors, &second)
+            .await
+            .unwrap();
+        drain_startup_buffer(&daemon).await.unwrap();
+        ready.send(()).unwrap();
+        daemon.ingress_barrier().await;
+        let source = daemon.translations.lock().values().next().unwrap().clone();
+        source.barrier().await.unwrap();
+        let storage = tmp.path().join("pipeline").join(format!(
+            "{}.sqlite",
+            crate::ids::session_storage_id(&env.source, &env.session_id)
+        ));
+        let db = rusqlite::Connection::open(storage).unwrap();
+        let count: u64 = db
+            .query_row(
+                "SELECT count(*) FROM batches WHERE committed=1",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(count, (INGRESS_QUEUE_CAPACITY + 4) as u64);
+        drop(db);
+        drain_all(&daemon).await;
+    }
     #[tokio::test]
     async fn retention_requires_complete_inactive_partitions_and_nonempty_heads() {
         for scenario in [
diff --git a/bt-daemon/src/server/mod.rs b/bt-daemon/src/server/mod.rs
index fba1ef20..6320a630 100644
--- a/bt-daemon/src/server/mod.rs
+++ b/bt-daemon/src/server/mod.rs
@@ -131,6 +131,10 @@ pub struct Daemon {
     ingress_tx: mpsc::Sender<IngressMsg>,
     ingress_overflow: AtomicBool,
     ingress_dispatched: Mutex<HashMap<DeliveryKey, u64>>,
+    /// Earliest live append during startup, keyed by source journal path.
+    /// Recovery must not absorb queued live events into its historical head.
+    startup_cutoffs: Mutex<Option<HashMap<PathBuf, u64>>>,
+    startup_buffer: Mutex<Option<StartupBuffer>>,
     started: Instant,
     last_activity: Mutex<Instant>,
     /// Prevents a shutdown drain from racing capture that has passed the
@@ -143,8 +147,19 @@ pub struct Daemon {
 }
 
 impl Daemon {
+    #[cfg(test)]
     fn new(opts: ServeOptions, data_dir: PathBuf) -> Arc<Self> {
+        Self::new_with_startup(opts, data_dir, None)
+    }
+
+    fn new_with_startup(
+        opts: ServeOptions,
+        data_dir: PathBuf,
+        startup: Option<oneshot::Receiver<()>>,
+    ) -> Arc<Self> {
         let (ingress_tx, ingress_rx) = mpsc::channel(INGRESS_QUEUE_CAPACITY);
+        let startup_cutoffs = startup.as_ref().map(|_| HashMap::new());
+        let startup_buffer = startup.as_ref().map(|_| StartupBuffer::new());
         let daemon = Arc::new(Daemon {
             version: opts.version,
             data_dir,
@@ -171,6 +186,8 @@ impl Daemon {
             ingress_tx,
             ingress_overflow: AtomicBool::new(false),
             ingress_dispatched: Mutex::new(HashMap::new()),
+            startup_cutoffs: Mutex::new(startup_cutoffs),
+            startup_buffer: Mutex::new(startup_buffer),
             started: Instant::now(),
             last_activity: Mutex::new(Instant::now()),
             capture_gate: tokio::sync::RwLock::new(()),
@@ -179,7 +196,7 @@ impl Daemon {
             shutting_down: AtomicBool::new(false),
             shutdown: Notify::new(),
         });
-        spawn_ingress_worker(daemon.clone(), ingress_rx);
+        spawn_ingress_worker(daemon.clone(), ingress_rx, startup);
         daemon
     }
 
@@ -228,6 +245,12 @@ pub async fn run(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()> {
         paths::ensure_private_dir(parent)?;
     }
 
+    #[cfg(unix)]
+    let Some(ownership) = transport::wait_for_endpoint_lock(&socket).await?
+    else {
+        tracing::info!("another daemon owns {}; exiting", socket.display());
+        return Ok(());
+    };
     let listener = match transport::claim(&socket, || probe_alive(&socket)).await? {
         Some(l) => l,
         None => {
@@ -240,25 +263,50 @@ pub async fn run(args: ServeArgs, opts: ServeOptions) -> anyhow::Result<()> {
     };
     tracing::info!(socket = %socket.display(), "bt-daemon listening");
 
-    let daemon = Daemon::new(opts, data_dir);
-    restore_active_parent_snapshots(&daemon.data_dir, &daemon.correlation).await;
-    restore_pending_sessions(&daemon).await;
-    collect_garbage(&daemon).await;
-    recover_unprocessed_journals(&daemon).await;
-    spawn_pending_reconciler(daemon.clone());
+    let (ready, startup) = oneshot::channel();
+    let daemon = Daemon::new_with_startup(opts, data_dir, Some(startup));
     let idle_timeout = Duration::from_secs(args.idle_timeout_secs);
-    spawn_idle_watchdog(daemon.clone(), idle_timeout);
-    spawn_session_reaper(
-        daemon.clone(),
-        Duration::from_secs(args.session_idle_timeout_secs),
-    );
-    spawn_gc(daemon.clone());
+    let session_idle_timeout = Duration::from_secs(args.session_idle_timeout_secs);
+    let recovery_daemon = daemon.clone();
+    let recovery = tokio::spawn(async move {
+        restore_active_parent_snapshots(&recovery_daemon.data_dir, &recovery_daemon.correlation)
+            .await;
+        restore_pending_sessions(&recovery_daemon).await;
+        collect_garbage(&recovery_daemon).await;
+        let candidates = discover_unprocessed_journals(&recovery_daemon).await;
+        *recovery_daemon.startup_cutoffs.lock() = None;
+        for event in candidates {
+            dispatch_ingress_event(&recovery_daemon, event).await;
+        }
+        recovery_daemon.settle_session_actors().await;
+        let _ = retry_pending_sessions(&recovery_daemon).await;
+        if let Err(error) = drain_startup_buffer(&recovery_daemon).await {
+            tracing::error!(%error, "startup ingress recovery failed");
+            recovery_daemon.trigger_shutdown();
+            return;
+        }
+        let _ = ready.send(());
+        if !recovery_daemon.shutting_down.load(Ordering::SeqCst) {
+            spawn_pending_reconciler(recovery_daemon.clone());
+            spawn_idle_watchdog(recovery_daemon.clone(), idle_timeout);
+            spawn_session_reaper(recovery_daemon.clone(), session_idle_timeout);
+            spawn_gc(recovery_daemon);
+        }
+    });
 
     let accept_result = accept_loop(daemon.clone(), listener).await;
 
     // Graceful drain regardless of why we stopped.
+    daemon.trigger_shutdown();
+    if let Err(error) = recovery.await {
+        tracing::error!(%error, "startup recovery task failed");
+    }
     drain_all(&daemon).await;
     transport::cleanup(&socket);
+    #[cfg(unix)]
+    if let Err(error) = fs2::FileExt::unlock(&ownership) {
+        tracing::warn!(%error, "daemon endpoint lock release failed");
+    }
     accept_result
 }
 
diff --git a/bt-daemon/src/server/recovery.rs b/bt-daemon/src/server/recovery.rs
index 3a453bac..11c3b0e5 100644
--- a/bt-daemon/src/server/recovery.rs
+++ b/bt-daemon/src/server/recovery.rs
@@ -2,7 +2,7 @@
 
 use super::*;
 
-pub(super) async fn recover_unprocessed_journals(daemon: &Arc<Daemon>) {
+pub(super) async fn discover_unprocessed_journals(daemon: &Arc<Daemon>) -> Vec<PendingEvent> {
     let pending: Vec<PendingEvent> = daemon
         .pending_sessions
         .lock()
@@ -10,7 +10,7 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc<Daemon>) {
         .flat_map(|state| state.events.iter().cloned())
         .collect();
     let Ok(mut entries) = tokio::fs::read_dir(journal::journal_dir(&daemon.data_dir)).await else {
-        return;
+        return Vec::new();
     };
     let mut candidates = Vec::new();
     while let Ok(Some(entry)) = entries.next_entry().await {
@@ -19,6 +19,12 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc<Daemon>) {
             continue;
         }
         let recorded_len = journal::JournalReader::recorded_len(&path).await;
+        let recorded_len = daemon
+            .startup_cutoffs
+            .lock()
+            .as_ref()
+            .and_then(|cutoffs| cutoffs.get(&path).copied())
+            .map_or(recorded_len, |cutoff| recorded_len.min(cutoff));
         let Ok(Some(mut reader)) = journal::JournalReader::open(&path, recorded_len).await else {
             continue;
         };
@@ -191,23 +197,25 @@ pub(super) async fn recover_unprocessed_journals(daemon: &Arc<Daemon>) {
     }
     candidates.sort_by_key(|event| event.env.ts_ms);
     let recovered = candidates.len();
-    for event in candidates {
+    if recovered > 0 {
+        tracing::info!(
+            recovered,
+            "queued unprocessed journal sessions for recovery"
+        );
+    }
+    candidates
+}
+
+#[cfg(test)]
+pub(super) async fn recover_unprocessed_journals(daemon: &Arc<Daemon>) {
+    for event in discover_unprocessed_journals(daemon).await {
         let _ = daemon
             .ingress_tx
             .send(IngressMsg::Event(Box::new(event)))
             .await;
     }
-    // Reconciliation stays in the daemon worker and follows every recovered
-    // event in queue order. Dropping the receiver is intentional: startup and
-    // hook capture do not wait for translation or reporting.
     let (reply, _ignored) = oneshot::channel();
     let _ = daemon.ingress_tx.send(IngressMsg::Barrier(reply)).await;
-    if recovered > 0 {
-        tracing::info!(
-            recovered,
-            "queued unprocessed journal sessions for recovery"
-        );
-    }
 }
 
 #[cfg(test)]
diff --git a/bt-daemon/src/server/sessions.rs b/bt-daemon/src/server/sessions.rs
index b1028f17..158c8a97 100644
--- a/bt-daemon/src/server/sessions.rs
+++ b/bt-daemon/src/server/sessions.rs
@@ -277,6 +277,10 @@ impl Daemon {
             .await?;
         let mut writer = writer.lock().await;
         let before = writer.position();
+        if let Some(cutoffs) = self.startup_cutoffs.lock().as_mut() {
+            let path = journal::source_journal_path(&self.data_dir, &env.source, &env.session_id);
+            cutoffs.entry(path).or_insert(before);
+        }
         let through = writer.append(env).await?;
         Ok((before, through))
     }
diff --git a/bt-daemon/src/transport.rs b/bt-daemon/src/transport.rs
index 6b7f829e..99436bb0 100644
--- a/bt-daemon/src/transport.rs
+++ b/bt-daemon/src/transport.rs
@@ -9,6 +9,40 @@ use std::path::Path;
 #[cfg(windows)]
 use std::time::Duration;
 
+/// Hold a stable lock file for the lifetime of a Unix listener. A timed-out
+/// RPC cannot establish that the process which bound the socket has died.
+#[cfg(unix)]
+pub(crate) fn lock_endpoint(endpoint: &Path) -> anyhow::Result<Option<std::fs::File>> {
+    let mut name = endpoint.as_os_str().to_os_string();
+    name.push(".lock");
+    let file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(std::path::PathBuf::from(name))?;
+    match fs2::FileExt::try_lock_exclusive(&file) {
+        Ok(()) => Ok(Some(file)),
+        Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => Ok(None),
+        Err(error) => Err(error.into()),
+    }
+}
+
+/// A departing process (or a short-lived inherited file descriptor) can hold
+/// the lock briefly after removing the socket. Allow that handover to finish.
+#[cfg(unix)]
+pub(crate) async fn wait_for_endpoint_lock(
+    endpoint: &Path,
+) -> anyhow::Result<Option<std::fs::File>> {
+    for _ in 0..50 {
+        if let Some(lock) = lock_endpoint(endpoint)? {
+            return Ok(Some(lock));
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+    }
+    Ok(None)
+}
+
 #[cfg(unix)]
 pub(crate) type ClientStream = tokio::net::UnixStream;
 #[cfg(windows)]
@@ -144,14 +178,14 @@ where
 #[cfg(unix)]
 async fn claim_unix<F, Fut>(
     endpoint: &Path,
-    probe_alive: &mut F,
+    _probe_alive: &mut F,
 ) -> anyhow::Result<Option<Listener>>
 where
     F: FnMut() -> Fut,
     Fut: Future<Output = bool>,
 {
     if endpoint.exists() {
-        if probe_alive().await {
+        if unix_endpoint_in_use(endpoint).await? {
             return Ok(None);
         }
         cleanup(endpoint);
@@ -159,7 +193,7 @@ where
     match Listener::bind_raw(endpoint) {
         Ok(listener) => Ok(Some(listener)),
         Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => {
-            if probe_alive().await {
+            if unix_endpoint_in_use(endpoint).await? {
                 Ok(None)
             } else {
                 cleanup(endpoint);
@@ -170,6 +204,25 @@ where
     }
 }
 
+/// A successful transport connection proves ownership even when an older
+/// daemon has not started accepting RPCs. A timeout is ambiguous and must not
+/// authorize unlinking a potentially live listener.
+#[cfg(unix)]
+async fn unix_endpoint_in_use(endpoint: &Path) -> anyhow::Result<bool> {
+    match tokio::time::timeout(std::time::Duration::from_secs(1), connect(endpoint)).await {
+        Ok(Ok(_)) | Err(_) => Ok(true),
+        Ok(Err(error))
+            if matches!(
+                error.kind(),
+                std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused
+            ) =>
+        {
+            Ok(false)
+        }
+        Ok(Err(error)) => Err(error.into()),
+    }
+}
+
 #[cfg(windows)]
 async fn claim_windows<F, Fut>(
     endpoint: &Path,
@@ -214,6 +267,56 @@ pub(crate) fn cleanup(endpoint: &Path) {
 #[cfg(windows)]
 pub(crate) fn cleanup(_endpoint: &Path) {}
 
+#[cfg(all(test, unix))]
+mod unix_tests {
+    #[test]
+    fn endpoint_lock_survives_an_unresponsive_owner_and_is_reusable() {
+        let tmp = tempfile::tempdir().unwrap();
+        let endpoint = tmp.path().join("daemon.sock");
+        let owner = super::lock_endpoint(&endpoint).unwrap().unwrap();
+        assert!(super::lock_endpoint(&endpoint).unwrap().is_none());
+        drop(owner);
+        assert!(super::lock_endpoint(&endpoint).unwrap().is_some());
+    }
+
+    #[tokio::test]
+    async fn startup_owner_excludes_a_contender_before_socket_bind() {
+        let tmp = tempfile::tempdir().unwrap();
+        let endpoint = tmp.path().join("daemon.sock");
+        let owner = super::lock_endpoint(&endpoint).unwrap().unwrap();
+        assert!(!endpoint.exists());
+        let contender_path = endpoint.clone();
+        let mut contender = tokio::spawn(async move {
+            super::wait_for_endpoint_lock(&contender_path)
+                .await
+                .unwrap()
+        });
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(100), &mut contender)
+                .await
+                .is_err()
+        );
+        assert!(!endpoint.exists());
+        drop(owner);
+        assert!(contender.await.unwrap().is_some());
+    }
+
+    #[tokio::test]
+    async fn unresponsive_legacy_listener_is_not_unlinked() {
+        let tmp = tempfile::tempdir().unwrap();
+        let endpoint = tmp.path().join("daemon.sock");
+        let listener = tokio::net::UnixListener::bind(&endpoint).unwrap();
+        // A legacy daemon doesn't take the new lock or answer initialize while
+        // recovering. Its successful transport connection still owns the path.
+        let claimed = super::claim(&endpoint, || async { false }).await.unwrap();
+        assert!(claimed.is_none());
+        assert!(endpoint.exists());
+        drop(listener);
+        let claimed = super::claim(&endpoint, || async { false }).await.unwrap();
+        assert!(claimed.is_some());
+    }
+}
+
 #[cfg(all(test, windows))]
 mod tests {
     use super::*;