ImageApi/src/ai/turn_registry.rs

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,
    /// Persona id for persona-scoped turns. None for insight-scoped turns.
    pub persona_id: Option<String>,
    /// `"insight"` (file-anchored chat) | `"persona"` (open chat).
    pub scope: String,
    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,
    /// Stable identity used to scope the turn. For insight chat this is
    /// `(file_path, library_id)`. For persona chat it is `(persona_id)`.
    /// The generic fields stay populated regardless of scope so the SSE
    /// `turn_info` frame can always surface a useful label — a future
    /// "list active turns" debugging surface won't need to discriminate.
    pub file_path: String,
    pub library_id: i32,
    /// Persona id for persona-scoped turns (`scope == "persona"`). None for
    /// the existing insight-scoped turns. The field is the cheapest way to
    /// let the SSE `turn_info` payload round-trip without a parallel map.
    pub persona_id: Option<String>,
    /// `"insight"` (default — file-anchored chat) | `"persona"` (open chat).
    /// Stable string so the SSE `turn_info` payload can carry it as-is.
    pub scope: String,
    /// 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::with_scope(turn_id, file_path, library_id, None, "insight")
    }

    /// Persona-scoped constructor. `file_path` is unused by the persona
    /// chat loop but kept on the struct so the SSE `turn_info` frame has
    /// the same shape across scopes.
    pub fn new_persona(turn_id: String, persona_id: String, library_id: i32) -> Self {
        Self::with_scope(
            turn_id,
            String::new(),
            library_id,
            Some(persona_id),
            "persona",
        )
    }

    fn with_scope(
        turn_id: String,
        file_path: String,
        library_id: i32,
        persona_id: Option<String>,
        scope: &str,
    ) -> Self {
        Self {
            turn_id,
            file_path,
            library_id,
            persona_id,
            scope: scope.to_string(),
            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,
            persona_id: self.persona_id.clone(),
            scope: self.scope.clone(),
            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);
    }

    // ── Persona-scope regression guards ───────────────────────────
    //
    // The open-chat surface relies on `scope == "persona"` and the
    // optional `persona_id` surviving a round-trip through `info()`.
    // A future refactor that hard-codes "insight" or drops the field
    // would silently break the SSE `turn_info` payload — guard it.

    #[tokio::test]
    async fn turn_entry_supports_persona_scope_label() {
        let entry = Arc::new(TurnEntry::new_persona(
            "tp1".to_string(),
            "journal".to_string(),
            1,
        ));
        assert_eq!(entry.scope, "persona");
        assert_eq!(entry.persona_id.as_deref(), Some("journal"));
        let info = entry.info().await;
        assert_eq!(info.scope, "persona");
        assert_eq!(info.persona_id.as_deref(), Some("journal"));
        // File-anchored fields stay blank — the persona chat has no file.
        assert!(info.file_path.is_empty());
    }

    #[tokio::test]
    async fn turn_info_payload_for_persona_does_not_leak_file_path() {
        // Defensive: even if some future code path accidentally populates
        // file_path on a persona entry, the SSE `turn_info` payload
        // should not surface it to clients. Today file_path is always
        // empty for persona entries, so we assert that contract.
        let entry = Arc::new(TurnEntry::new_persona(
            "tp2".to_string(),
            "default".to_string(),
            1,
        ));
        let info = entry.info().await;
        let json = serde_json::to_string(&serde_json::json!({
            "turn_id": info.turn_id,
            "persona_id": info.persona_id,
            "scope": info.scope,
            "status": info.status.as_str(),
            "total_events_pushed": info.total_events_pushed,
            "buffered_count": info.buffered_count,
        }))
        .unwrap();
        // Privacy: a persona chat reply that quotes an SMS must not echo
        // the file path that message originally surfaced from.
        assert!(
            !json.contains("file_path"),
            "persona turn_info payload should not leak file_path: {json}"
        );
    }

    #[tokio::test]
    async fn insight_scope_defaults_when_using_legacy_constructor() {
        // The legacy `TurnEntry::new` constructor (used by the existing
        // file-anchored insight chat) must still report scope="insight"
        // so callers can discriminate without a parallel map.
        let entry = Arc::new(TurnEntry::new("t-insight".into(), "/p.jpg".into(), 1));
        let info = entry.info().await;
        assert_eq!(info.scope, "insight");
        assert!(info.persona_id.is_none());
    }
}