Add reconnectable async chat-turn flow with in-memory TurnRegistry

Replace the one-shot SSE chat stream with an async dispatch + reconnectable
replay flow so the mobile client survives backgrounding, network blips, and
OS-killed sockets without losing an in-flight agentic turn.

- TurnRegistry/TurnEntry: in-memory per-turn event buffer (cap 500, front
  eviction) shared by the agentic loop (writer) and SSE replay readers.
  ReplayOutcome + replay_from/next_batch distinguish Events/CaughtUp/Gone;
  next_batch registers the Notify before reading state (no lost wakeup) and
  drains every buffered event before signaling terminal, so the final
  Done/Error is never dropped and the stream closes cleanly.
- Endpoints: POST /insights/chat/turn (202 + turn_id), GET
  /insights/chat/turn/{id} (SSE replay, ?skip_before= resume, per-event seq,
  410 on eviction), DELETE /insights/chat/turn/{id} (real task abort +
  cooperative is_running() check at each loop boundary).
- Cancellation actually stops the task (AbortHandle stored on the entry) and
  emits a Done{cancelled:true}; callers skip persistence on cancel.
- Background sweeper drops stale turns; interval clamped to <=300s.
- OpenTelemetry spans: ai.chat.turn.execute/replay/cancel.
- Legacy POST /insights/chat/stream path preserved unchanged.

Tests: registry coverage for terminal delivery (race guard), waiting, Gone,
abort, eviction; handler integration tests for 404/410, skip_before, seq
stamping, completed replay, and cancel.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/ai/turn_registry.rs

@@ -0,0 +1,748 @@
+use crate::ai::insight_chat::ChatStreamEvent;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::time::Instant;
+use tokio::sync::{Mutex, Notify};
+use tokio::task::AbortHandle;
+
+/// Maximum number of events buffered per turn. Agentic turns typically
+/// produce ~120 events; 500 provides 4× headroom. When exceeded, oldest
+/// events are evicted from the front.
+const MAX_BUFFERED_EVENTS: usize = 500;
+
+/// Turn status codes used by `TurnEntry::status`.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum TurnStatus {
+    Running = 0,
+    Done = 1,
+    Error = 2,
+    Cancelled = 3,
+}
+
+impl From<u32> for TurnStatus {
+    fn from(v: u32) -> Self {
+        match v {
+            0 => TurnStatus::Running,
+            1 => TurnStatus::Done,
+            2 => TurnStatus::Error,
+            3 => TurnStatus::Cancelled,
+            _ => TurnStatus::Running,
+        }
+    }
+}
+
+impl TurnStatus {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            TurnStatus::Running => "running",
+            TurnStatus::Done => "done",
+            TurnStatus::Error => "error",
+            TurnStatus::Cancelled => "cancelled",
+        }
+    }
+}
+
+/// Shared metadata about a turn, read by the SSE replay handler to emit
+/// the initial `turn_info` event and to decide whether to wait for new
+/// events or close immediately.
+#[derive(Debug, Clone)]
+pub struct TurnInfo {
+    pub turn_id: String,
+    pub file_path: String,
+    pub library_id: i32,
+    pub status: TurnStatus,
+    pub total_events_pushed: u32,
+    pub buffered_count: u32,
+}
+
+/// Result of reading events at or after an absolute `skip_before` index.
+#[derive(Debug)]
+pub enum ReplayOutcome {
+    /// New events are available. `next_skip` is the absolute index to pass
+    /// on the next read (i.e. one past the last event returned).
+    Events {
+        events: Vec<ChatStreamEvent>,
+        next_skip: u32,
+    },
+    /// The reader is caught up to the live edge — no events past `skip_before`
+    /// yet. `next_skip` is the current high-water mark.
+    CaughtUp { next_skip: u32 },
+    /// `skip_before` points below the buffer's base index: the requested
+    /// events were evicted. Maps to HTTP 410 Gone.
+    Gone,
+}
+
+/// Per-turn state shared between the agentic loop (writer) and all SSE
+/// replay connections (readers).
+pub struct TurnEntry {
+    pub turn_id: String,
+    pub file_path: String,
+    pub library_id: i32,
+    /// Shared event buffer — multiple SSE connections can read independently.
+    /// Each connection tracks its own `skip_before` offset.
+    events: Mutex<Vec<ChatStreamEvent>>,
+    /// Monotonic counter: total events pushed (may exceed events.len()
+    /// due to eviction). Used for skip_before indexing.
+    total_events_pushed: AtomicU32,
+    /// The event index that this entry started with. Adjusts on eviction
+    /// so that `skip_before` stays absolute across connections.
+    base_index: AtomicU32,
+    pub status: AtomicU32,
+    /// Abort handle for the spawned agentic task, set once after spawn.
+    /// Behind a std `Mutex` because the entry is shared via `Arc` and the
+    /// handle is installed after the entry is already in the registry.
+    abort_handle: StdMutex<Option<AbortHandle>>,
+    pub created_at: Instant,
+    notify: Arc<Notify>,
+}
+
+impl TurnEntry {
+    pub fn new(turn_id: String, file_path: String, library_id: i32) -> Self {
+        Self {
+            turn_id,
+            file_path,
+            library_id,
+            events: Mutex::new(Vec::new()),
+            total_events_pushed: AtomicU32::new(0),
+            base_index: AtomicU32::new(0),
+            status: AtomicU32::new(TurnStatus::Running as u32),
+            abort_handle: StdMutex::new(None),
+            created_at: Instant::now(),
+            notify: Arc::new(Notify::new()),
+        }
+    }
+
+    /// Install the abort handle for the spawned agentic task. Called once,
+    /// right after the task is spawned.
+    pub fn set_abort_handle(&self, handle: AbortHandle) {
+        *self.abort_handle.lock().expect("abort_handle poisoned") = Some(handle);
+    }
+
+    /// Abort the spawned agentic task, if a handle was installed. Returns
+    /// `true` if a task was aborted.
+    pub fn abort(&self) -> bool {
+        if let Some(handle) = self
+            .abort_handle
+            .lock()
+            .expect("abort_handle poisoned")
+            .take()
+        {
+            handle.abort();
+            true
+        } else {
+            false
+        }
+    }
+
+    /// Push an event into the buffer. Evicts oldest events if the buffer
+    /// exceeds `MAX_BUFFERED_EVENTS`. Notifies all waiting SSE connections.
+    pub async fn push_event(&self, event: ChatStreamEvent) {
+        {
+            let mut events = self.events.lock().await;
+
+            // Evict oldest events if we've hit the cap.
+            if events.len() >= MAX_BUFFERED_EVENTS {
+                // Drop the oldest event to make room and advance the base
+                // index so skip_before stays absolute across connections.
+                events.remove(0);
+                self.base_index.fetch_add(1, Ordering::Relaxed);
+            }
+
+            events.push(event);
+            // Increment while holding the buffer lock so the counter stays in
+            // lock-step with the buffer even if multiple writers ever exist.
+            self.total_events_pushed.fetch_add(1, Ordering::Relaxed);
+        }
+
+        self.notify.notify_waiters();
+    }
+
+    /// Get a snapshot of turn metadata for the `turn_info` SSE event.
+    pub async fn info(&self) -> TurnInfo {
+        let events = self.events.lock().await;
+        let buffered = events.len() as u32;
+        let total = self.total_events_pushed.load(Ordering::Relaxed);
+        drop(events);
+
+        TurnInfo {
+            turn_id: self.turn_id.clone(),
+            file_path: self.file_path.clone(),
+            library_id: self.library_id,
+            status: self.status.load(Ordering::Relaxed).into(),
+            total_events_pushed: total,
+            buffered_count: buffered,
+        }
+    }
+
+    /// Set the terminal status and notify all waiters.
+    pub fn set_terminal_status(&self, status: TurnStatus) {
+        self.status.store(status as u32, Ordering::Relaxed);
+        self.notify.notify_waiters();
+    }
+
+    /// Read buffered events at or after absolute index `skip_before` without
+    /// waiting. Distinguishes "evicted" (Gone) from "caught up" (no new
+    /// events yet) — the previous boolean/`Option` API conflated the two.
+    pub async fn replay_from(&self, skip_before: u32) -> ReplayOutcome {
+        let events = self.events.lock().await;
+        let base = self.base_index.load(Ordering::Relaxed);
+
+        // The buffer holds absolute indices [base, base + len). A request
+        // below `base` asked for events that have been evicted.
+        if skip_before < base {
+            return ReplayOutcome::Gone;
+        }
+
+        let offset = (skip_before - base) as usize;
+        let next_skip = base + events.len() as u32;
+        if offset >= events.len() {
+            // Caught up to (or past) the live edge — nothing new yet.
+            return ReplayOutcome::CaughtUp { next_skip };
+        }
+
+        ReplayOutcome::Events {
+            events: events[offset..].to_vec(),
+            next_skip,
+        }
+    }
+
+    /// Wait for the next batch of events past `skip_before`, the turn to
+    /// finish, or eviction. Returns:
+    /// - `Events` when new events are available (drained before any terminal
+    ///   signal so the final `Done`/`Error` is never dropped),
+    /// - `CaughtUp` only when the turn has reached a terminal status and the
+    ///   reader is fully drained (the caller should close the stream),
+    /// - `Gone` when `skip_before` points into evicted territory.
+    pub async fn next_batch(&self, skip_before: u32) -> ReplayOutcome {
+        loop {
+            // Register interest BEFORE inspecting state so a push/terminal that
+            // races between our read and our await can't be lost (Notify's
+            // `notify_waiters` does not store a permit).
+            let notified = self.notify.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
+
+            match self.replay_from(skip_before).await {
+                ReplayOutcome::CaughtUp { next_skip } => {
+                    // No new events. If the turn is finished, every event
+                    // (including the terminal one) has already been drained
+                    // above on a prior call, so signal the caller to close.
+                    if !self.is_running() {
+                        return ReplayOutcome::CaughtUp { next_skip };
+                    }
+                    // Still running — wait for the next push or terminal.
+                }
+                other => return other, // Events or Gone
+            }
+
+            notified.await;
+        }
+    }
+
+    /// Check if this turn is still running.
+    pub fn is_running(&self) -> bool {
+        self.status.load(Ordering::Relaxed) == TurnStatus::Running as u32
+    }
+}
+
+/// In-memory registry of all active chat turns. Injected into `AppState`
+/// and shared across all handlers.
+pub struct TurnRegistry {
+    entries: Mutex<HashMap<String, Arc<TurnEntry>>>,
+    timeout_secs: u64,
+}
+
+impl TurnRegistry {
+    pub fn new(timeout_secs: u64) -> Self {
+        Self {
+            entries: Mutex::new(HashMap::new()),
+            timeout_secs,
+        }
+    }
+
+    /// Returns the cleanup timeout in seconds.
+    pub fn timeout_secs(&self) -> u64 {
+        self.timeout_secs
+    }
+
+    /// Insert a new turn entry. Returns the turn_id.
+    pub async fn insert(&self, entry: Arc<TurnEntry>) -> String {
+        let turn_id = entry.turn_id.clone();
+        let mut entries = self.entries.lock().await;
+        entries.insert(turn_id.clone(), entry);
+        turn_id
+    }
+
+    /// Look up a turn by id. Returns None if not found or expired.
+    pub async fn get(&self, turn_id: &str) -> Option<Arc<TurnEntry>> {
+        let entries = self.entries.lock().await;
+        entries.get(turn_id).cloned()
+    }
+
+    /// Clean up stale entries older than the timeout. Returns the count of
+    /// entries removed.
+    pub async fn cleanup_stale(&self) -> usize {
+        let mut entries = self.entries.lock().await;
+        let _now = Instant::now();
+        let stale: Vec<String> = entries
+            .iter()
+            .filter(|(_, entry)| entry.created_at.elapsed().as_secs() > self.timeout_secs)
+            .map(|(id, _)| id.clone())
+            .collect();
+
+        for id in &stale {
+            entries.remove(id);
+        }
+
+        if !stale.is_empty() {
+            log::info!(
+                "TurnRegistry: cleaned up {} stale entries (timeout={}s)",
+                stale.len(),
+                self.timeout_secs
+            );
+        }
+
+        stale.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ai::insight_chat::ChatStreamEvent;
+    use std::time::Duration;
+
+    /// Unwrap the events from a `ReplayOutcome::Events`, panicking otherwise.
+    fn events_of(outcome: ReplayOutcome) -> Vec<ChatStreamEvent> {
+        match outcome {
+            ReplayOutcome::Events { events, .. } => events,
+            other => panic!("expected Events, got {other:?}"),
+        }
+    }
+
+    // ── TurnStatus ──────────────────────────────────────────────────
+
+    #[test]
+    fn turn_status_from_u32_valid_values() {
+        assert_eq!(TurnStatus::from(0), TurnStatus::Running);
+        assert_eq!(TurnStatus::from(1), TurnStatus::Done);
+        assert_eq!(TurnStatus::from(2), TurnStatus::Error);
+        assert_eq!(TurnStatus::from(3), TurnStatus::Cancelled);
+    }
+
+    #[test]
+    fn turn_status_from_u32_unknown_defaults_to_running() {
+        assert_eq!(TurnStatus::from(4), TurnStatus::Running);
+        assert_eq!(TurnStatus::from(u32::MAX), TurnStatus::Running);
+    }
+
+    #[test]
+    fn turn_status_as_str() {
+        assert_eq!(TurnStatus::Running.as_str(), "running");
+        assert_eq!(TurnStatus::Done.as_str(), "done");
+        assert_eq!(TurnStatus::Error.as_str(), "error");
+        assert_eq!(TurnStatus::Cancelled.as_str(), "cancelled");
+    }
+
+    // ── TurnEntry ───────────────────────────────────────────────────
+
+    #[tokio::test]
+    async fn turn_entry_push_and_replay() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+
+        entry
+            .push_event(ChatStreamEvent::TextDelta("hello".to_string()))
+            .await;
+        entry
+            .push_event(ChatStreamEvent::TextDelta(" world".to_string()))
+            .await;
+
+        let events = events_of(entry.replay_from(0).await);
+        assert_eq!(events.len(), 2);
+    }
+
+    #[tokio::test]
+    async fn turn_entry_replay_with_skip() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+
+        for i in 0..5 {
+            entry
+                .push_event(ChatStreamEvent::TextDelta(format!("e{i}")))
+                .await;
+        }
+
+        // skip_before=0 → all 5 events
+        let all = events_of(entry.replay_from(0).await);
+        assert_eq!(all.len(), 5);
+
+        // skip_before=2 → events 2,3,4 (3 events)
+        let skipped = events_of(entry.replay_from(2).await);
+        assert_eq!(skipped.len(), 3);
+
+        // skip_before=5 → caught up to the live edge (not Gone).
+        assert!(matches!(
+            entry.replay_from(5).await,
+            ReplayOutcome::CaughtUp { next_skip: 5 }
+        ));
+    }
+
+    #[tokio::test]
+    async fn turn_entry_replay_empty_by_default() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+        // Empty buffer with skip_before=0 → caught up (nothing to replay yet).
+        assert!(matches!(
+            entry.replay_from(0).await,
+            ReplayOutcome::CaughtUp { next_skip: 0 }
+        ));
+    }
+
+    #[tokio::test]
+    async fn turn_entry_is_running_initially() {
+        let entry = TurnEntry::new("t1".to_string(), "/photo.jpg".to_string(), 1);
+        assert!(entry.is_running());
+    }
+
+    #[tokio::test]
+    async fn turn_entry_set_terminal_status() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+        assert!(entry.is_running());
+        entry.set_terminal_status(TurnStatus::Done);
+        assert!(!entry.is_running());
+    }
+
+    #[tokio::test]
+    async fn turn_entry_info() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            42,
+        ));
+
+        entry
+            .push_event(ChatStreamEvent::TextDelta("x".to_string()))
+            .await;
+        entry.set_terminal_status(TurnStatus::Done);
+
+        let info = entry.info().await;
+        assert_eq!(info.turn_id, "t1");
+        assert_eq!(info.file_path, "/photo.jpg");
+        assert_eq!(info.library_id, 42);
+        assert_eq!(info.status, TurnStatus::Done);
+        assert_eq!(info.total_events_pushed, 1);
+        assert_eq!(info.buffered_count, 1);
+    }
+
+    #[tokio::test]
+    async fn turn_entry_eviction_caps_buffer() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+
+        // Push MAX_BUFFERED_EVENTS + 10 events.
+        for i in 0..(MAX_BUFFERED_EVENTS + 10) {
+            entry
+                .push_event(ChatStreamEvent::TextDelta(format!("e{i}")))
+                .await;
+        }
+
+        // Asking from absolute 0 after eviction is Gone (0-9 were dropped).
+        assert!(matches!(entry.replay_from(0).await, ReplayOutcome::Gone));
+
+        // Reading from the new base (10) returns the full capped buffer.
+        let events = events_of(entry.replay_from(10).await);
+        assert_eq!(events.len(), MAX_BUFFERED_EVENTS);
+
+        // First event should be at index 10 (0-9 were evicted).
+        if let ChatStreamEvent::TextDelta(s) = &events[0] {
+            assert_eq!(s, "e10");
+        } else {
+            panic!("expected TextDelta");
+        }
+
+        // Last event should be at index MAX_BUFFERED_EVENTS + 9.
+        if let ChatStreamEvent::TextDelta(s) = &events[events.len() - 1] {
+            assert_eq!(s, &format!("e{}", MAX_BUFFERED_EVENTS + 9));
+        } else {
+            panic!("expected TextDelta");
+        }
+    }
+
+    #[tokio::test]
+    async fn turn_entry_replay_evicted_index_is_gone() {
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+
+        // Push one past the cap so exactly one event (index 0) is evicted.
+        for i in 0..=MAX_BUFFERED_EVENTS {
+            entry
+                .push_event(ChatStreamEvent::TextDelta(format!("e{i}")))
+                .await;
+        }
+
+        // Base is now 1; asking from absolute 0 is evicted territory → Gone.
+        assert!(matches!(entry.replay_from(0).await, ReplayOutcome::Gone));
+
+        // skip_before = MAX_BUFFERED_EVENTS → last event only (index valid).
+        let last = events_of(entry.replay_from(MAX_BUFFERED_EVENTS as u32).await);
+        assert_eq!(last.len(), 1);
+
+        // skip_before = MAX_BUFFERED_EVENTS + 1 → caught up to the live edge.
+        assert!(matches!(
+            entry.replay_from((MAX_BUFFERED_EVENTS + 1) as u32).await,
+            ReplayOutcome::CaughtUp { .. }
+        ));
+    }
+
+    // ── TurnRegistry ────────────────────────────────────────────────
+
+    #[tokio::test]
+    async fn turn_registry_insert_and_get() {
+        let registry = TurnRegistry::new(300);
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+        let id = registry.insert(entry).await;
+        assert_eq!(id, "t1");
+
+        let retrieved = registry.get("t1").await;
+        assert!(retrieved.is_some());
+        assert_eq!(retrieved.unwrap().turn_id, "t1");
+    }
+
+    #[tokio::test]
+    async fn turn_registry_get_nonexistent_returns_none() {
+        let registry = TurnRegistry::new(300);
+        assert!(registry.get("nonexistent").await.is_none());
+    }
+
+    #[tokio::test]
+    async fn turn_registry_cleanup_stale_removes_old_entries() {
+        let registry = TurnRegistry::new(0);
+        let mut entry = TurnEntry::new("t1".to_string(), "/photo.jpg".to_string(), 1);
+        entry.created_at = Instant::now() - Duration::from_secs(1);
+        registry.insert(Arc::new(entry)).await;
+
+        let cleaned = registry.cleanup_stale().await;
+        assert_eq!(cleaned, 1);
+        assert!(registry.get("t1").await.is_none());
+    }
+
+    #[tokio::test]
+    async fn turn_registry_cleanup_stale_preserves_recent() {
+        let registry = TurnRegistry::new(3600); // 1 hour
+        let entry = Arc::new(TurnEntry::new(
+            "t1".to_string(),
+            "/photo.jpg".to_string(),
+            1,
+        ));
+        registry.insert(entry).await;
+
+        let cleaned = registry.cleanup_stale().await;
+        assert_eq!(cleaned, 0);
+        assert!(registry.get("t1").await.is_some());
+    }
+
+    #[tokio::test]
+    async fn turn_registry_cleanup_stale_multiple() {
+        let registry = TurnRegistry::new(0);
+
+        for i in 0..5 {
+            let mut entry = TurnEntry::new(format!("t{i}"), "/photo.jpg".to_string(), 1);
+            entry.created_at = Instant::now() - Duration::from_secs(1);
+            registry.insert(Arc::new(entry)).await;
+        }
+
+        let cleaned = registry.cleanup_stale().await;
+        assert_eq!(cleaned, 5);
+    }
+
+    #[tokio::test]
+    async fn turn_registry_timeout_secs() {
+        let registry = TurnRegistry::new(600);
+        assert_eq!(registry.timeout_secs(), 600);
+    }
+
+    // ── next_batch / live replay ────────────────────────────────────
+
+    /// Drain a turn the way the SSE replay handler does: pull batches via
+    /// `next_batch` until the turn is finished and fully drained.
+    async fn drain_to_end(entry: Arc<TurnEntry>) -> Vec<ChatStreamEvent> {
+        let mut out = Vec::new();
+        let mut skip = 0u32;
+        while let ReplayOutcome::Events { events, next_skip } = entry.next_batch(skip).await {
+            out.extend(events);
+            skip = next_skip;
+        }
+        out
+    }
+
+    fn is_terminal(ev: &ChatStreamEvent) -> bool {
+        matches!(ev, ChatStreamEvent::Done { .. } | ChatStreamEvent::Error(_))
+    }
+
+    /// The core guarantee behind the replay rewrite: a reader waiting on
+    /// `next_batch` always receives the terminal event, even though the
+    /// writer flips status to terminal immediately after pushing it.
+    #[tokio::test]
+    async fn next_batch_always_delivers_terminal_event() {
+        for _ in 0..50 {
+            let entry = Arc::new(TurnEntry::new("t".into(), "/p.jpg".into(), 1));
+
+            let writer = entry.clone();
+            let w = tokio::spawn(async move {
+                writer
+                    .push_event(ChatStreamEvent::IterationStart { n: 1, max: 6 })
+                    .await;
+                writer
+                    .push_event(ChatStreamEvent::TextDelta("hi".into()))
+                    .await;
+                // Push terminal then flip status with no await between — the
+                // race that previously dropped the Done on the reader side.
+                writer
+                    .push_event(ChatStreamEvent::Done {
+                        tool_calls_made: 0,
+                        iterations_used: 1,
+                        truncated: false,
+                        prompt_tokens: None,
+                        eval_tokens: None,
+                        num_ctx: None,
+                        amended_insight_id: None,
+                        backend_used: "local".into(),
+                        model_used: "m".into(),
+                        cancelled: false,
+                    })
+                    .await;
+                writer.set_terminal_status(TurnStatus::Done);
+            });
+
+            let events = drain_to_end(entry).await;
+            w.await.unwrap();
+
+            assert!(
+                events.last().is_some_and(is_terminal),
+                "terminal event missing; got {} events",
+                events.len()
+            );
+            assert_eq!(events.len(), 3, "expected IterationStart, TextDelta, Done");
+        }
+    }
+
+    /// A reader that connects before any event is pushed blocks in
+    /// `next_batch` and then receives events as the writer produces them.
+    #[tokio::test]
+    async fn next_batch_waits_for_late_events() {
+        let entry = Arc::new(TurnEntry::new("t".into(), "/p.jpg".into(), 1));
+
+        let writer = entry.clone();
+        tokio::spawn(async move {
+            tokio::task::yield_now().await;
+            writer
+                .push_event(ChatStreamEvent::TextDelta("late".into()))
+                .await;
+            writer.set_terminal_status(TurnStatus::Done);
+        });
+
+        // First call blocks until the writer pushes, rather than returning
+        // CaughtUp on the empty buffer of a running turn.
+        match entry.next_batch(0).await {
+            ReplayOutcome::Events { events, next_skip } => {
+                assert_eq!(events.len(), 1);
+                assert_eq!(next_skip, 1);
+            }
+            other => panic!("expected Events, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn next_batch_closes_on_terminal_when_caught_up() {
+        let entry = Arc::new(TurnEntry::new("t".into(), "/p.jpg".into(), 1));
+        entry
+            .push_event(ChatStreamEvent::TextDelta("x".into()))
+            .await;
+        entry.set_terminal_status(TurnStatus::Done);
+
+        // Caught up (skip past the one buffered event) on a finished turn →
+        // CaughtUp so the handler closes the stream rather than hanging.
+        assert!(matches!(
+            entry.next_batch(1).await,
+            ReplayOutcome::CaughtUp { .. }
+        ));
+    }
+
+    #[tokio::test]
+    async fn next_batch_reports_gone_for_evicted_index() {
+        let entry = Arc::new(TurnEntry::new("t".into(), "/p.jpg".into(), 1));
+        for i in 0..=MAX_BUFFERED_EVENTS {
+            entry
+                .push_event(ChatStreamEvent::TextDelta(format!("e{i}")))
+                .await;
+        }
+        // Index 0 was evicted (base advanced to 1).
+        assert!(matches!(entry.next_batch(0).await, ReplayOutcome::Gone));
+    }
+
+    // ── abort handle (#1 cancellation) ──────────────────────────────
+
+    #[tokio::test]
+    async fn abort_handle_aborts_task_once() {
+        let entry = Arc::new(TurnEntry::new("t".into(), "/p.jpg".into(), 1));
+
+        // No handle installed yet → abort is a no-op.
+        assert!(!entry.abort());
+
+        let handle = tokio::spawn(async {
+            // Long-lived task that only ends via abort.
+            futures::future::pending::<()>().await;
+        });
+        entry.set_abort_handle(handle.abort_handle());
+
+        assert!(entry.abort(), "first abort should fire");
+        assert!(!entry.abort(), "handle is taken; second abort is a no-op");
+
+        // The aborted task resolves to a cancellation JoinError.
+        let join = handle.await;
+        assert!(join.unwrap_err().is_cancelled());
+    }
+
+    #[tokio::test]
+    async fn base_index_tracks_eviction() {
+        let entry = Arc::new(TurnEntry::new("t".into(), "/p.jpg".into(), 1));
+        for i in 0..(MAX_BUFFERED_EVENTS + 5) {
+            entry
+                .push_event(ChatStreamEvent::TextDelta(format!("e{i}")))
+                .await;
+        }
+        let info = entry.info().await;
+        // 5 events evicted; total keeps climbing, buffer stays capped.
+        assert_eq!(info.total_events_pushed, (MAX_BUFFERED_EVENTS + 5) as u32);
+        assert_eq!(info.buffered_count, MAX_BUFFERED_EVENTS as u32);
+        // First live index is 5: reading from there yields the full buffer.
+        let from_base = events_of(entry.replay_from(5).await);
+        assert_eq!(from_base.len(), MAX_BUFFERED_EVENTS);
+    }
+}