pi_agent_rust 0.5.0

use std::collections::VecDeque;
use std::sync::Arc;

use asupersync::Cx;
use asupersync::sync::{Mutex, OwnedMutexGuard};
use serde_json::{Value, json};

use super::{
    AgentState, Cmd, ConversationMessage, EXTENSION_EVENT_TIMEOUT_MS, PiApp, PiMsg,
    conversation_from_session,
};
use crate::checkpoint::RetryPlan;
use crate::extension_events::{SessionBeforeCompactOutcome, apply_session_before_compact_response};
use crate::extensions::ExtensionEventName;
use crate::model::{Message, Usage};
use crate::session::{CompactionEntry, Session, SessionEntry};

async fn deliver_compaction_terminal_event(
    event_tx: &asupersync::channel::mpsc::Sender<PiMsg>,
    cx: &Cx,
    message: PiMsg,
) -> bool {
    let delivered = crate::interactive::enqueue_pi_event(event_tx, cx, message).await;
    if !delivered {
        tracing::error!("terminal compaction event was not delivered before runtime shutdown");
    }
    delivered
}

fn spawn_compaction_terminal_event(
    runtime_handle: &asupersync::runtime::RuntimeHandle,
    event_tx: asupersync::channel::mpsc::Sender<PiMsg>,
    message: PiMsg,
) {
    if let Err(err) = runtime_handle.try_spawn_with_cx(move |completion_cx| async move {
        deliver_compaction_terminal_event(&event_tx, &completion_cx, message).await;
    }) {
        tracing::error!(
            error = %err,
            "terminal compaction event could not be admitted by the runtime"
        );
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CompactionPersistenceOutcome {
    Confirmed,
    Disabled,
    ReconciledButUnconfirmed,
}

impl CompactionPersistenceOutcome {
    const fn may_emit_success_event(self) -> bool {
        matches!(self, Self::Confirmed | Self::Disabled)
    }
}

#[derive(Debug)]
struct CompactionSessionCommit {
    messages_for_agent: Vec<Message>,
    messages_for_ui: Vec<ConversationMessage>,
    usage: Usage,
    compaction_entry: CompactionEntry,
    persistence: CompactionPersistenceOutcome,
}

async fn confirm_exact_compaction_after_save_error(
    candidate: &Session,
    entry_id: &str,
    expected_entry: &[u8],
) -> Option<()> {
    let path = candidate.path.as_ref()?;
    let (reopened, diagnostics) = Session::open_with_diagnostics(path.to_string_lossy().as_ref())
        .await
        .ok()?;
    if !diagnostics.skipped_entries.is_empty() || !diagnostics.orphaned_parent_links.is_empty() {
        return None;
    }
    if reopened.header.id != candidate.header.id || reopened.leaf_id() != Some(entry_id) {
        return None;
    }
    if serde_json::to_vec(&reopened.header).ok()? != serde_json::to_vec(&candidate.header).ok()?
        || serde_json::to_vec(&reopened.entries).ok()?
            != serde_json::to_vec(&candidate.entries).ok()?
    {
        return None;
    }
    let reopened_entry = reopened.get_entry(entry_id)?;
    if serde_json::to_vec(reopened_entry).ok()?.as_slice() != expected_entry {
        return None;
    }
    Some(())
}

#[allow(clippy::too_many_arguments)]
async fn stage_and_commit_compaction_session(
    session: Arc<Mutex<Session>>,
    expected_session_id: &str,
    expected_leaf_id: Option<&str>,
    summary: String,
    first_kept_entry_id: String,
    tokens_before: u64,
    details: Option<Value>,
    from_hook: Option<bool>,
    save_enabled: bool,
    cx: &Cx,
) -> crate::error::Result<CompactionSessionCommit> {
    let mut live = OwnedMutexGuard::lock(session, cx)
        .await
        .map_err(|err| crate::error::Error::session(err.to_string()))?;
    if live.header.id != expected_session_id || live.leaf_id() != expected_leaf_id {
        return Err(crate::error::Error::session(
            "Session changed while compaction was running; compaction was not applied".to_string(),
        ));
    }

    let mut candidate = live.clone();
    let entry_id = candidate.append_compaction(
        summary,
        first_kept_entry_id,
        tokens_before,
        details,
        from_hook,
    );
    let compaction_entry = candidate
        .get_entry(&entry_id)
        .and_then(|entry| match entry {
            SessionEntry::Compaction(compaction) => Some(compaction.clone()),
            _ => None,
        })
        .ok_or_else(|| {
            crate::error::Error::session(
                "Staged compaction entry disappeared before commit".to_string(),
            )
        })?;
    let expected_entry = serde_json::to_vec(&SessionEntry::Compaction(compaction_entry.clone()))?;
    let persistence = if !save_enabled {
        CompactionPersistenceOutcome::Disabled
    } else if let Err(err) = candidate.save().await {
        tracing::error!(
            error = %err,
            entry_id,
            "compaction save failed; reconciling the exact operation against current disk state"
        );
        if confirm_exact_compaction_after_save_error(&candidate, &entry_id, &expected_entry)
            .await
            .is_none()
        {
            return Err(crate::error::Error::session(
                    "Compaction persistence was not confirmed, current disk state could not be reconciled, and the active in-memory session was left unchanged"
                        .to_string(),
                ));
        }
        CompactionPersistenceOutcome::ReconciledButUnconfirmed
    } else {
        CompactionPersistenceOutcome::Confirmed
    };

    let messages_for_agent = candidate.to_messages_for_current_path();
    let (messages_for_ui, usage) = conversation_from_session(&candidate);

    *live = candidate;
    Ok(CompactionSessionCommit {
        messages_for_agent,
        messages_for_ui,
        usage,
        compaction_entry,
        persistence,
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RetryPersistenceOutcome {
    Confirmed,
    Disabled,
    ReconciledButUnconfirmed,
}

#[derive(Debug)]
struct RetrySessionCommit {
    messages_for_agent: Vec<Message>,
    messages_for_ui: Vec<ConversationMessage>,
    usage: Usage,
    plan: RetryPlan,
    persistence: RetryPersistenceOutcome,
}

async fn confirm_exact_retry_after_save_error(
    candidate: &Session,
    expected_leaf_id: Option<&str>,
) -> Option<()> {
    let path = candidate.path.as_ref()?;
    let (reopened, diagnostics) = Session::open_with_diagnostics(path.to_string_lossy().as_ref())
        .await
        .ok()?;
    if !diagnostics.skipped_entries.is_empty() || !diagnostics.orphaned_parent_links.is_empty() {
        return None;
    }
    if reopened.header.id != candidate.header.id || reopened.leaf_id() != expected_leaf_id {
        return None;
    }
    if serde_json::to_vec(&reopened.header).ok()? != serde_json::to_vec(&candidate.header).ok()?
        || serde_json::to_vec(&reopened.entries).ok()?
            != serde_json::to_vec(&candidate.entries).ok()?
    {
        return None;
    }
    Some(())
}

async fn stage_and_commit_retry(
    session: Arc<Mutex<Session>>,
    expected_session_id: &str,
    expected_leaf_id: Option<&str>,
    save_enabled: bool,
    cx: &Cx,
) -> crate::error::Result<RetrySessionCommit> {
    let mut live = OwnedMutexGuard::lock(session, cx)
        .await
        .map_err(|err| crate::error::Error::session(err.to_string()))?;
    if live.header.id != expected_session_id || live.leaf_id() != expected_leaf_id {
        return Err(crate::error::Error::session(
            "Session changed while retry was preparing; retry was not applied".to_string(),
        ));
    }
    let plan = crate::checkpoint::plan_retry(&live)
        .ok_or_else(|| crate::error::Error::session("No user turn to retry".to_string()))?;
    let mut candidate = live.clone();
    crate::checkpoint::apply_retry_plan(&mut candidate, &plan).map_err(|err| {
        crate::error::Error::session(format!("Retry plan could not be applied: {err}"))
    })?;

    let new_leaf_id = candidate.leaf_id().map(str::to_string);
    let persistence = if !save_enabled {
        RetryPersistenceOutcome::Disabled
    } else if let Err(err) = candidate.save().await {
        tracing::error!(
            error = %err,
            ?new_leaf_id,
            "retry save failed; reconciling the exact operation against current disk state"
        );
        if confirm_exact_retry_after_save_error(&candidate, new_leaf_id.as_deref())
            .await
            .is_none()
        {
            return Err(crate::error::Error::session(format!(
                "Retry persistence was not confirmed ({err}), current disk state could not be reconciled, and the active in-memory session was left unchanged"
            )));
        }
        RetryPersistenceOutcome::ReconciledButUnconfirmed
    } else {
        RetryPersistenceOutcome::Confirmed
    };

    let messages_for_agent = candidate.to_messages_for_current_path();
    let (messages_for_ui, usage) = conversation_from_session(&candidate);
    *live = candidate;
    Ok(RetrySessionCommit {
        messages_for_agent,
        messages_for_ui,
        usage,
        plan,
        persistence,
    })
}

/// Safely convert `Duration::as_micros()` (u128) to u64 with saturation.
#[inline]
#[allow(clippy::cast_possible_truncation)]
pub(super) fn micros_as_u64(micros: u128) -> u64 {
    micros.min(u128::from(u64::MAX)) as u64
}

/// Microsecond-precision frame timing stats for TUI performance measurement.
///
/// Uses interior mutability (`RefCell`/`Cell`) because `view(&self)` cannot
/// take `&mut self` (the `bubbletea::Model` trait requires `&self` for `view`).
/// This is safe because the TUI event loop is single-threaded.
///
/// Gated behind `PI_PERF_TELEMETRY=1` environment variable.  When disabled,
/// no `Instant::now()` calls are made — zero runtime overhead.
pub struct FrameTimingStats {
    pub(super) frame_times_us: std::cell::RefCell<VecDeque<u64>>,
    pub(super) content_build_times_us: std::cell::RefCell<VecDeque<u64>>,
    pub(super) viewport_sync_times_us: std::cell::RefCell<VecDeque<u64>>,
    pub(super) update_times_us: VecDeque<u64>,
    pub(super) total_frames: std::cell::Cell<u64>,
    pub(super) budget_exceeded_count: std::cell::Cell<u64>,
    pub(super) enabled: bool,
}

pub(super) const FRAME_TIMING_WINDOW: usize = 60;
pub(super) const FRAME_BUDGET_US: u64 = 16_667;

impl FrameTimingStats {
    pub(super) fn new() -> Self {
        let enabled =
            std::env::var_os("PI_PERF_TELEMETRY").is_some_and(|v| v == "1" || v == "true");
        Self {
            frame_times_us: std::cell::RefCell::new(VecDeque::with_capacity(FRAME_TIMING_WINDOW)),
            content_build_times_us: std::cell::RefCell::new(VecDeque::with_capacity(
                FRAME_TIMING_WINDOW,
            )),
            viewport_sync_times_us: std::cell::RefCell::new(VecDeque::with_capacity(
                FRAME_TIMING_WINDOW,
            )),
            update_times_us: VecDeque::with_capacity(FRAME_TIMING_WINDOW),
            total_frames: std::cell::Cell::new(0),
            budget_exceeded_count: std::cell::Cell::new(0),
            enabled,
        }
    }

    pub(super) fn record_frame(&self, elapsed_us: u64) {
        if !self.enabled {
            return;
        }
        let metrics = crate::session_metrics::global();
        if metrics.enabled() {
            metrics.tui_render.record(elapsed_us);
        }
        let mut times = self.frame_times_us.borrow_mut();
        if times.len() >= FRAME_TIMING_WINDOW {
            times.pop_front();
        }
        times.push_back(elapsed_us);
        let total = self.total_frames.get() + 1;
        self.total_frames.set(total);
        if elapsed_us > FRAME_BUDGET_US {
            self.budget_exceeded_count
                .set(self.budget_exceeded_count.get() + 1);
        }
        if total.is_multiple_of(FRAME_TIMING_WINDOW as u64) {
            drop(times);
            self.emit_stats();
        }
    }

    pub(super) fn record_content_build(&self, elapsed_us: u64) {
        if !self.enabled {
            return;
        }
        let metrics = crate::session_metrics::global();
        if metrics.enabled() {
            metrics.tui_content_build.record(elapsed_us);
        }
        let mut times = self.content_build_times_us.borrow_mut();
        if times.len() >= FRAME_TIMING_WINDOW {
            times.pop_front();
        }
        times.push_back(elapsed_us);
    }

    pub(super) fn record_viewport_sync(&self, elapsed_us: u64) {
        if !self.enabled {
            return;
        }
        let metrics = crate::session_metrics::global();
        if metrics.enabled() {
            metrics.tui_viewport_sync.record(elapsed_us);
        }
        let mut times = self.viewport_sync_times_us.borrow_mut();
        if times.len() >= FRAME_TIMING_WINDOW {
            times.pop_front();
        }
        times.push_back(elapsed_us);
    }

    pub(super) fn record_update(&mut self, elapsed_us: u64) {
        if !self.enabled {
            return;
        }
        let metrics = crate::session_metrics::global();
        if metrics.enabled() {
            metrics.tui_update.record(elapsed_us);
        }
        if self.update_times_us.len() >= FRAME_TIMING_WINDOW {
            self.update_times_us.pop_front();
        }
        self.update_times_us.push_back(elapsed_us);
    }

    pub(super) fn frame_p99_us(&self) -> u64 {
        Self::percentiles(&self.frame_times_us.borrow()).2
    }

    pub(super) const fn enable_for_test(&mut self) {
        self.enabled = true;
    }

    pub(super) fn reset_for_test(&mut self) {
        self.frame_times_us.borrow_mut().clear();
        self.content_build_times_us.borrow_mut().clear();
        self.viewport_sync_times_us.borrow_mut().clear();
        self.update_times_us.clear();
        self.total_frames.set(0);
        self.budget_exceeded_count.set(0);
    }

    pub(super) fn percentiles(times: &VecDeque<u64>) -> (u64, u64, u64, u64) {
        if times.is_empty() {
            return (0, 0, 0, 0);
        }
        let mut sorted: Vec<u64> = times.iter().copied().collect();
        sorted.sort_unstable();
        let len = sorted.len();
        let p50 = sorted[len / 2];
        let p95 = sorted[(len * 95 / 100).min(len - 1)];
        let p99 = sorted[(len * 99 / 100).min(len - 1)];
        let p999 = sorted[(len * 999 / 1000).min(len - 1)];
        (p50, p95, p99, p999)
    }

    pub(super) fn snapshot_json(&self, surface: &str, fixture: &Value) -> Value {
        let frame_times = self.frame_times_us.borrow();
        let content_times = self.content_build_times_us.borrow();
        let viewport_times = self.viewport_sync_times_us.borrow();
        let update_times = &self.update_times_us;
        let recent_over_budget =
            Self::over_budget_count(frame_times.as_slices().0, FRAME_BUDGET_US)
                + Self::over_budget_count(frame_times.as_slices().1, FRAME_BUDGET_US);
        let verdict = if frame_times.is_empty() {
            "empty"
        } else if recent_over_budget == 0 {
            "pass"
        } else {
            "warn"
        };

        json!({
            "schema": "pi.tui.frame_budget.v1",
            "surface": surface,
            "enabled": self.enabled,
            "budget_us": FRAME_BUDGET_US,
            "window_capacity": FRAME_TIMING_WINDOW,
            "fixture": fixture,
            "samples": {
                "frame": Self::sample_stats_json(&frame_times, FRAME_BUDGET_US),
                "content_build": Self::sample_stats_json(&content_times, FRAME_BUDGET_US),
                "viewport_sync": Self::sample_stats_json(&viewport_times, FRAME_BUDGET_US),
                "update": Self::sample_stats_json(update_times, FRAME_BUDGET_US),
            },
            "totals": {
                "frames": self.total_frames.get(),
                "budget_exceeded": self.budget_exceeded_count.get(),
                "recent_budget_exceeded": recent_over_budget,
            },
            "verdict": verdict,
            "redaction": {
                "prompt_content": "omitted",
                "tool_payload_content": "omitted",
                "model_response_content": "omitted",
            },
        })
    }

    fn over_budget_count(times: &[u64], budget_us: u64) -> usize {
        times.iter().filter(|&&value| value > budget_us).count()
    }

    fn sample_stats_json(times: &VecDeque<u64>, budget_us: u64) -> Value {
        let (p50, p95, p99, p999) = Self::percentiles(times);
        let max = times.iter().copied().max().unwrap_or(0);
        let over_budget = Self::over_budget_count(times.as_slices().0, budget_us)
            + Self::over_budget_count(times.as_slices().1, budget_us);
        json!({
            "count": times.len(),
            "p50_us": p50,
            "p95_us": p95,
            "p99_us": p99,
            "p999_us": p999,
            "max_us": max,
            "over_budget_count": over_budget,
        })
    }

    #[allow(clippy::cast_precision_loss)]
    fn emit_stats(&self) {
        let frame = Self::percentiles(&self.frame_times_us.borrow());
        let content = Self::percentiles(&self.content_build_times_us.borrow());
        let viewport = Self::percentiles(&self.viewport_sync_times_us.borrow());
        let total = self.total_frames.get();
        let exceeded = self.budget_exceeded_count.get();
        let window = self.frame_times_us.borrow().len();
        let recent_exceeded = self
            .frame_times_us
            .borrow()
            .iter()
            .filter(|&&t| t > FRAME_BUDGET_US)
            .count();
        let fixture = json!({
            "name": "rolling_frame_window",
            "source": "PI_PERF_TELEMETRY",
            "sample_window": FRAME_TIMING_WINDOW,
        });
        let snapshot = self.snapshot_json("interactive_tui", &fixture);
        tracing::debug!(
            telemetry = %snapshot,
            "[perf] frame p50={:.1}ms p95={:.1}ms p99={:.1}ms p999={:.1}ms | \
             content p50={:.1}ms p95={:.1}ms p99={:.1}ms p999={:.1}ms | \
             viewport p50={:.1}ms p95={:.1}ms p99={:.1}ms p999={:.1}ms | \
             budget_exceeded={recent_exceeded}/{window} (total={exceeded}/{total})",
            frame.0 as f64 / 1000.0,
            frame.1 as f64 / 1000.0,
            frame.2 as f64 / 1000.0,
            frame.3 as f64 / 1000.0,
            content.0 as f64 / 1000.0,
            content.1 as f64 / 1000.0,
            content.2 as f64 / 1000.0,
            content.3 as f64 / 1000.0,
            viewport.0 as f64 / 1000.0,
            viewport.1 as f64 / 1000.0,
            viewport.2 as f64 / 1000.0,
            viewport.3 as f64 / 1000.0,
        );
    }

    #[allow(clippy::cast_precision_loss)]
    pub(super) fn summary(&self) -> String {
        if !self.enabled {
            return String::from("Frame telemetry disabled (set PI_PERF_TELEMETRY=1 to enable)");
        }
        let frame = Self::percentiles(&self.frame_times_us.borrow());
        let content = Self::percentiles(&self.content_build_times_us.borrow());
        let viewport = Self::percentiles(&self.viewport_sync_times_us.borrow());
        let update = Self::percentiles(&self.update_times_us);
        let total = self.total_frames.get();
        let exceeded = self.budget_exceeded_count.get();
        format!(
            "Frame timing (last {FRAME_TIMING_WINDOW} frames):\n  \
             view()   p50={:.1}ms  p95={:.1}ms  p99={:.1}ms  p999={:.1}ms\n  \
             content  p50={:.1}ms  p95={:.1}ms  p99={:.1}ms  p999={:.1}ms\n  \
             viewport p50={:.1}ms  p95={:.1}ms  p99={:.1}ms  p999={:.1}ms\n  \
             update() p50={:.1}ms  p95={:.1}ms  p99={:.1}ms  p999={:.1}ms\n  \
             Budget exceeded: {exceeded}/{total} frames (>{:.1}ms)",
            frame.0 as f64 / 1000.0,
            frame.1 as f64 / 1000.0,
            frame.2 as f64 / 1000.0,
            frame.3 as f64 / 1000.0,
            content.0 as f64 / 1000.0,
            content.1 as f64 / 1000.0,
            content.2 as f64 / 1000.0,
            content.3 as f64 / 1000.0,
            viewport.0 as f64 / 1000.0,
            viewport.1 as f64 / 1000.0,
            viewport.2 as f64 / 1000.0,
            viewport.3 as f64 / 1000.0,
            update.0 as f64 / 1000.0,
            update.1 as f64 / 1000.0,
            update.2 as f64 / 1000.0,
            update.3 as f64 / 1000.0,
            FRAME_BUDGET_US as f64 / 1000.0,
        )
    }
}

/// TUI responsiveness pressure derived from frame latency and pending output.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum TuiPressureLevel {
    Normal,
    Elevated,
    High,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct TuiPressureDecision {
    pub(super) level: TuiPressureLevel,
    pub(super) throttle_tool_updates: bool,
    pub(super) flush_interval: std::time::Duration,
    pub(super) max_pending_tool_events: usize,
    pub(super) max_pending_tool_output_bytes: usize,
}

impl TuiPressureDecision {
    pub(super) const fn normal() -> Self {
        Self {
            level: TuiPressureLevel::Normal,
            throttle_tool_updates: false,
            flush_interval: std::time::Duration::ZERO,
            max_pending_tool_events: 1,
            max_pending_tool_output_bytes: 32 * 1024,
        }
    }
}

pub(super) struct TuiPressureController;

impl TuiPressureController {
    pub(super) const ELEVATED_FRAME_P99_US: u64 = FRAME_BUDGET_US;
    pub(super) const HIGH_FRAME_P99_US: u64 = FRAME_BUDGET_US * 3;
    pub(super) const ELEVATED_TOOL_OUTPUT_BYTES: usize = 32 * 1024;
    pub(super) const HIGH_TOOL_OUTPUT_BYTES: usize = 256 * 1024;
    pub(super) const ELEVATED_PENDING_TOOL_EVENTS: usize = 2;
    pub(super) const HIGH_PENDING_TOOL_EVENTS: usize = 8;

    pub(super) const fn decide(
        frame_p99_us: u64,
        output_bytes: usize,
        pending_tool_events: usize,
    ) -> TuiPressureDecision {
        if frame_p99_us >= Self::HIGH_FRAME_P99_US
            || output_bytes >= Self::HIGH_TOOL_OUTPUT_BYTES
            || pending_tool_events >= Self::HIGH_PENDING_TOOL_EVENTS
        {
            TuiPressureDecision {
                level: TuiPressureLevel::High,
                throttle_tool_updates: true,
                flush_interval: std::time::Duration::from_millis(160),
                max_pending_tool_events: Self::HIGH_PENDING_TOOL_EVENTS,
                max_pending_tool_output_bytes: 1024 * 1024,
            }
        } else if frame_p99_us >= Self::ELEVATED_FRAME_P99_US
            || output_bytes >= Self::ELEVATED_TOOL_OUTPUT_BYTES
            || pending_tool_events >= Self::ELEVATED_PENDING_TOOL_EVENTS
        {
            TuiPressureDecision {
                level: TuiPressureLevel::Elevated,
                throttle_tool_updates: true,
                flush_interval: std::time::Duration::from_millis(80),
                max_pending_tool_events: 4,
                max_pending_tool_output_bytes: Self::HIGH_TOOL_OUTPUT_BYTES,
            }
        } else {
            TuiPressureDecision::normal()
        }
    }
}

/// Memory pressure level based on RSS.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum MemoryLevel {
    /// RSS < 50MB — no action needed.
    Normal,
    /// 50MB <= RSS < 100MB — log warning, show in /session.
    Warning,
    /// 100MB <= RSS < 200MB — progressive tool output collapse.
    Pressure,
    /// RSS >= 200MB — truncate old messages, force degraded rendering.
    Critical,
}

impl MemoryLevel {
    pub(super) const fn from_rss_bytes(rss: usize) -> Self {
        const MB: usize = 1_000_000;
        if rss >= 200 * MB {
            Self::Critical
        } else if rss >= 100 * MB {
            Self::Pressure
        } else if rss >= 50 * MB {
            Self::Warning
        } else {
            Self::Normal
        }
    }
}

/// Abstraction for reading RSS, injectable for testing.
pub(super) trait RssReader: Send {
    fn read_rss_bytes(&self) -> Option<usize>;
}

pub(super) struct FnRssReader {
    read_fn: Box<dyn Fn() -> Option<usize> + Send>,
}

impl FnRssReader {
    pub(super) fn new(read_fn: Box<dyn Fn() -> Option<usize> + Send>) -> Self {
        Self { read_fn }
    }
}

impl RssReader for FnRssReader {
    fn read_rss_bytes(&self) -> Option<usize> {
        (self.read_fn)()
    }
}

/// Reads RSS from /proc/self/statm on Linux.
pub(super) struct ProcSelfRssReader;

/// Page size in bytes. Hardcoded to 4096 (standard for x86_64/aarch64 Linux)
/// to avoid unsafe libc::sysconf — crate uses `#![forbid(unsafe_code)]`.
const PROC_PAGE_SIZE: usize = 4096;

impl RssReader for ProcSelfRssReader {
    fn read_rss_bytes(&self) -> Option<usize> {
        #[cfg(target_os = "linux")]
        {
            // /proc/self/statm: "total_pages resident_pages shared_pages ..."
            let content = std::fs::read_to_string("/proc/self/statm").ok()?;
            let resident_pages: usize = content.split_whitespace().nth(1)?.parse().ok()?;
            Some(resident_pages * PROC_PAGE_SIZE)
        }
        #[cfg(not(target_os = "linux"))]
        {
            // macOS (and other non-Linux) has no /proc; ask sysinfo for our
            // own RSS. Without this the memory tiers never sample and the
            // progressive-degradation path is inert off Linux (bd-1h9dp).
            // Sampling is interval-gated by MemoryMonitor, so the refresh
            // cost stays off the hot path.
            let pid = sysinfo::Pid::from_u32(std::process::id());
            let mut system = sysinfo::System::new();
            system.refresh_processes(sysinfo::ProcessesToUpdate::Some(&[pid]), true);
            let rss = system.process(pid)?.memory();
            usize::try_from(rss).ok()
        }
    }
}

/// Hysteresis threshold: stop collapsing when RSS drops below this.
pub(super) const MEMORY_RELIEF_BYTES: usize = 80_000_000;

/// Maximum messages retained when Critical truncation triggers.
pub(super) const CRITICAL_KEEP_MESSAGES: usize = 30;

/// Memory monitor that drives progressive conversation management.
pub(super) struct MemoryMonitor {
    pub(super) reader: Box<dyn RssReader>,
    pub(super) last_sample: std::time::Instant,
    pub(super) sample_interval: std::time::Duration,
    pub(super) current_rss_bytes: usize,
    pub(super) peak_rss_bytes: usize,
    pub(super) level: MemoryLevel,
    /// Index into messages vec: next tool output to collapse.
    pub(super) next_collapse_index: usize,
    /// Whether progressive collapse is in progress.
    pub(super) collapsing: bool,
    /// When the last collapse action was performed (rate-limit to 1/sec).
    pub(super) last_collapse: std::time::Instant,
    /// Whether Critical truncation has already been applied this session.
    pub(super) truncated: bool,
}

impl MemoryMonitor {
    pub(super) fn new(reader: Box<dyn RssReader>) -> Self {
        let now = std::time::Instant::now();
        Self {
            reader,
            last_sample: now,
            sample_interval: std::time::Duration::from_secs(5),
            current_rss_bytes: 0,
            peak_rss_bytes: 0,
            level: MemoryLevel::Normal,
            next_collapse_index: 0,
            collapsing: false,
            last_collapse: now,
            truncated: false,
        }
    }

    pub(super) fn new_with_reader_fn(read_fn: Box<dyn Fn() -> Option<usize> + Send>) -> Self {
        Self::new(Box::new(FnRssReader::new(read_fn)))
    }

    pub(super) fn new_default() -> Self {
        Self::new(Box::new(ProcSelfRssReader))
    }

    /// Sample RSS if the interval has elapsed. Returns true if level changed.
    pub(super) fn maybe_sample(&mut self) -> bool {
        if self.last_sample.elapsed() < self.sample_interval {
            return false;
        }
        self.last_sample = std::time::Instant::now();
        let Some(rss) = self.reader.read_rss_bytes() else {
            return false;
        };
        self.current_rss_bytes = rss;
        if rss > self.peak_rss_bytes {
            self.peak_rss_bytes = rss;
        }
        let new_level = MemoryLevel::from_rss_bytes(rss);
        let changed = new_level != self.level;
        if changed {
            match new_level {
                MemoryLevel::Warning => {
                    tracing::warn!(
                        rss_mb = rss / 1_000_000,
                        "Memory pressure: Warning level reached"
                    );
                }
                MemoryLevel::Pressure => {
                    tracing::warn!(
                        rss_mb = rss / 1_000_000,
                        "Memory pressure: Pressure level — starting progressive collapse"
                    );
                    self.collapsing = true;
                }
                MemoryLevel::Critical => {
                    tracing::error!(
                        rss_mb = rss / 1_000_000,
                        "Memory pressure: Critical level — truncating conversation"
                    );
                }
                MemoryLevel::Normal => {
                    tracing::info!(
                        rss_mb = rss / 1_000_000,
                        "Memory pressure relieved — back to Normal"
                    );
                    self.collapsing = false;
                }
            }
            self.level = new_level;
        }
        changed
    }

    /// Re-sample RSS immediately (used after collapse actions).
    pub(super) fn resample_now(&mut self) {
        if let Some(rss) = self.reader.read_rss_bytes() {
            self.current_rss_bytes = rss;
            if rss > self.peak_rss_bytes {
                self.peak_rss_bytes = rss;
            }
            self.level = MemoryLevel::from_rss_bytes(rss);
            if rss < MEMORY_RELIEF_BYTES {
                self.collapsing = false;
            }
        }
    }

    /// Format memory stats for /session display.
    #[allow(clippy::cast_precision_loss)]
    pub(super) fn summary(&self) -> String {
        let current_mb = self.current_rss_bytes as f64 / 1_000_000.0;
        let peak_mb = self.peak_rss_bytes as f64 / 1_000_000.0;
        let level_str = match self.level {
            MemoryLevel::Normal => "Normal",
            MemoryLevel::Warning => "Warning",
            MemoryLevel::Pressure => "Pressure (collapsing old outputs...)",
            MemoryLevel::Critical => "CRITICAL",
        };
        format!("Memory: {current_mb:.1}MB (peak {peak_mb:.1}MB) [{level_str}]")
    }

    /// Whether Critical-level rendering degradation should be forced.
    #[cfg(test)]
    pub(super) const fn should_force_degraded(&self) -> bool {
        matches!(self.level, MemoryLevel::Critical)
    }
}

impl PiApp {
    /// `/checkpoint [name] [note...]` (bd-cv653.3.7): cheap restore-point
    /// marker on the current leaf.
    #[allow(clippy::too_many_lines)]
    pub(super) fn handle_slash_checkpoint(&mut self, args: &str) -> Option<Cmd> {
        let Ok(agent_guard) = self.agent.try_lock() else {
            self.status_message = Some("Agent busy; try again".to_string());
            return None;
        };
        let messages: Vec<crate::model::Message> = agent_guard.messages().to_vec();
        drop(agent_guard);

        let (name, note) = args
            .split_once(char::is_whitespace)
            .map_or((args, ""), |(name, note)| (name, note));
        let cx = asupersync::Cx::for_request();
        let session = Arc::clone(&self.session);
        let note_owned = note.trim().to_string();
        let name_owned = name.trim().to_string();
        let event_tx = self.event_tx.clone();
        self.runtime_handle.spawn(async move {
            let Ok(mut guard) = OwnedMutexGuard::lock(session, &cx).await else {
                let _ = crate::interactive::enqueue_pi_event(
                    &event_tx,
                    &cx,
                    PiMsg::AgentError("Failed to lock session".to_string()),
                )
                .await;
                return;
            };
            let checkpoint = crate::checkpoint::mark_checkpoint(
                &mut guard,
                &name_owned,
                if note_owned.is_empty() {
                    None
                } else {
                    Some(note_owned.as_str())
                },
                &messages,
            );
            let _ = crate::interactive::enqueue_pi_event(
                &event_tx,
                &cx,
                PiMsg::System(format!(
                    "Checkpoint '{}' marked ({} messages, ~{} tokens). Rewind with /rewind{}.",
                    checkpoint.name,
                    checkpoint.message_count,
                    checkpoint.token_estimate,
                    if checkpoint.name == "checkpoint" {
                        String::new()
                    } else {
                        format!(" {}", checkpoint.name)
                    }
                )),
            )
            .await;
        });
        None
    }

    /// `/rewind [name]` (bd-cv653.3.7): collapse the span from a checkpoint
    /// to now into a concise report (tree keeps everything).
    #[allow(clippy::too_many_lines)]
    pub(super) fn handle_slash_rewind(&mut self, args: &str) -> Option<Cmd> {
        if self.agent_state != AgentState::Idle {
            self.status_message = Some("Cannot rewind while processing".to_string());
            return None;
        }
        let Ok(agent_guard) = self.agent.try_lock() else {
            self.status_message = Some("Agent busy; try again".to_string());
            return None;
        };
        let provider = agent_guard.provider();
        // Keyless providers (replay/test/local) summarize fine without a
        // key; credentialed ones carry theirs.
        let api_key = agent_guard
            .stream_options()
            .api_key
            .clone()
            .unwrap_or_default();
        drop(agent_guard);

        let name = args.trim().to_string();
        let session = Arc::clone(&self.session);
        let agent = Arc::clone(&self.agent);
        let event_tx = self.event_tx.clone();
        let runtime_handle = self.runtime_handle.clone();
        self.agent_state = AgentState::Processing;
        self.status_message = Some("Rewinding...".to_string());
        runtime_handle.spawn(async move {
            let cx = asupersync::Cx::for_request();
            let checkpoint = {
                let Ok(guard) = OwnedMutexGuard::lock(Arc::clone(&session), &cx).await else {
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::AgentError("Failed to lock session".to_string()),
                    )
                    .await;
                    return;
                };
                crate::checkpoint::find_checkpoint(
                    &guard,
                    if name.is_empty() {
                        None
                    } else {
                        Some(name.as_str())
                    },
                )
            };
            let Some(checkpoint) = checkpoint else {
                let _ = crate::interactive::enqueue_pi_event(
                    &event_tx,
                    &cx,
                    PiMsg::System(if name.is_empty() {
                        "No checkpoints yet — mark one with /checkpoint".to_string()
                    } else {
                        format!("No checkpoint named '{name}'")
                    }),
                )
                .await;
                return;
            };

            let span: Vec<crate::model::Message> = {
                let Ok(agent_guard) = OwnedMutexGuard::lock(Arc::clone(&agent), &cx).await else {
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::AgentError("Failed to lock agent".to_string()),
                    )
                    .await;
                    return;
                };
                agent_guard.messages()
                    [checkpoint.message_count.min(agent_guard.messages().len())..]
                    .to_vec()
            };
            if span.is_empty() {
                let _ = crate::interactive::enqueue_pi_event(
                    &event_tx,
                    &cx,
                    PiMsg::System(format!(
                        "Nothing to rewind — the active context is already at '{}'.",
                        checkpoint.name
                    )),
                )
                .await;
                return;
            }

            let settings = crate::compaction::ResolvedCompactionSettings {
                enabled: true,
                ..Default::default()
            };
            let summary = crate::checkpoint::summarize_span(&span, provider, &api_key, &settings)
                .await
                .unwrap_or_else(|err| {
                    format!(
                        "(summarization failed: {err}; the span was collapsed without a report)"
                    )
                });

            let outcome = {
                let Ok(mut agent_guard) = OwnedMutexGuard::lock(Arc::clone(&agent), &cx).await
                else {
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::AgentError("Failed to lock agent".to_string()),
                    )
                    .await;
                    return;
                };
                crate::checkpoint::apply_rewind_to_active(&mut agent_guard, &checkpoint, summary)
            };
            {
                let Ok(mut guard) = OwnedMutexGuard::lock(Arc::clone(&session), &cx).await else {
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::AgentError("Failed to lock session".to_string()),
                    )
                    .await;
                    return;
                };
                guard.append_custom_entry(
                    "rewind".to_string(),
                    Some(serde_json::to_value(&outcome).unwrap_or_default()),
                );
            }
            let _ = crate::interactive::enqueue_pi_event(
                &event_tx,
                &cx,
                PiMsg::System(format!(
                    "Rewound to '{}': {} messages collapsed into a report (~{} tokens). The tree kept everything.",
                    outcome.checkpoint, outcome.collapsed_messages, outcome.summary_tokens_estimate
                )),
            )
            .await;
        });
        None
    }

    /// `/undo [n] [force]` (bd-cv653.3.13): roll back the last n tool-path
    /// file mutations. `force` overrides the external-change guard.
    pub(super) fn handle_slash_undo(&mut self, args: &str) -> Option<Cmd> {
        self.apply_undo_redo(args, false)
    }

    /// `/redo [n] [force]` (bd-cv653.3.13): re-apply undone file mutations.
    pub(super) fn handle_slash_redo(&mut self, args: &str) -> Option<Cmd> {
        self.apply_undo_redo(args, true)
    }

    #[allow(clippy::too_many_lines)]
    fn apply_undo_redo(&mut self, args: &str, redo: bool) -> Option<Cmd> {
        let verb = if redo { "redo" } else { "undo" };
        if self.agent_state != AgentState::Idle {
            self.status_message = Some(format!("Cannot {verb} while processing"));
            return None;
        }
        let Ok(agent_guard) = self.agent.try_lock() else {
            self.status_message = Some("Agent busy; try again".to_string());
            return None;
        };
        let recorder = agent_guard.mutation_recorder();
        drop(agent_guard);
        let Some(recorder) = recorder else {
            self.status_message = Some(format!(
                "/{verb} unavailable: no mutation recorder in this session"
            ));
            return None;
        };

        let mut count = 1_usize;
        let mut force = false;
        for token in args.split_whitespace() {
            if token.eq_ignore_ascii_case("force") {
                force = true;
            } else if let Ok(n) = token.parse::<usize>() {
                count = n.max(1);
            } else {
                self.status_message = Some(format!("Usage: /{verb} [n] [force]"));
                return None;
            }
        }

        let outcome = if redo {
            recorder.redo(count, force)
        } else {
            recorder.undo(count, force)
        };

        let message = crate::undo::render_outcome_text(&outcome, redo, count);

        // Audit trail: record the applied operation as a session Custom entry
        // (mirrors checkpoint/rewind), then surface the report.
        let record = serde_json::json!({
            "schema": crate::undo::UNDO_SCHEMA,
            "action": verb,
            "outcome": outcome,
        });
        let applied_any = !outcome.applied.is_empty();
        let cx = asupersync::Cx::for_request();
        let session = Arc::clone(&self.session);
        let event_tx = self.event_tx.clone();
        self.runtime_handle.spawn(async move {
            if applied_any && let Ok(mut guard) = OwnedMutexGuard::lock(session, &cx).await {
                guard.append_custom_entry("undo".to_string(), Some(record));
            }
            let _ =
                crate::interactive::enqueue_pi_event(&event_tx, &cx, PiMsg::System(message)).await;
        });
        None
    }

    /// `/usage [refresh]` (bd-cv653.7.4): provider quota/credit table.
    pub(super) fn handle_slash_usage(&mut self, args: &str) -> Option<Cmd> {
        let refresh = args.trim().eq_ignore_ascii_case("refresh");
        let cx = asupersync::Cx::for_request();
        let event_tx = self.event_tx.clone();
        self.status_message = Some("Fetching provider usage...".to_string());
        self.runtime_handle.spawn(async move {
            let message = match crate::auth::AuthStorage::load(crate::config::Config::auth_path()) {
                Ok(auth) => {
                    let rows = crate::usage::gather_usage(&auth, refresh).await;
                    crate::usage::render_usage_text(&rows)
                }
                Err(err) => format!("Failed to load credentials: {err}"),
            };
            let _ =
                crate::interactive::enqueue_pi_event(&event_tx, &cx, PiMsg::System(message)).await;
        });
        None
    }

    /// `/fresh` (bd-cv653.3.7): reset provider stream state; transcript
    /// untouched.
    pub(super) fn handle_slash_fresh(&mut self) -> Option<Cmd> {
        let Ok(mut agent_guard) = self.agent.try_lock() else {
            self.status_message = Some("Agent busy; try again".to_string());
            return None;
        };
        // Agent-side reset is synchronous; only the session log spawns.
        // uuid suffix: a bare millisecond stamp can collide across rapid
        // calls, defeating the cache-reset purpose.
        let new_id = format!(
            "fresh-{}-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(0, |d| d.as_millis()),
            uuid::Uuid::new_v4().simple()
        );
        crate::app::rebind_stream_options_session(agent_guard.stream_options_mut(), &new_id);
        let messages_len = agent_guard.messages().len();
        drop(agent_guard);

        let cx = asupersync::Cx::for_request();
        let session = Arc::clone(&self.session);
        let event_tx = self.event_tx.clone();
        self.runtime_handle.spawn(async move {
            {
                let Ok(mut guard) = OwnedMutexGuard::lock(session, &cx).await else {
                    return;
                };
                guard.append_custom_entry(
                    "fresh".to_string(),
                    Some(serde_json::json!({
                        "schema": "pi.fresh.v1",
                        "newSessionId": new_id,
                        "reason": "operator /fresh: provider cache + stream bookkeeping reset",
                    })),
                );
            }
            let _ = crate::interactive::enqueue_pi_event(
                &event_tx,
                &cx,
                PiMsg::System(format!(
                    "Fresh stream state (session id {new_id}); transcript untouched ({messages_len} messages)."
                )),
            )
            .await;
        });
        None
    }

    /// `/retry` (bd-cv653.3.7, durable branching bd-r7icz): re-issue the last
    /// user turn as a SIBLING branch. Staging clones the Session, persists the
    /// rewound leaf, then swaps Session/Agent only after that save. The prompt
    /// is emitted as one `RetryCommitted` event so slash-command reparse cannot
    /// steal the sibling parent.
    #[allow(clippy::too_many_lines)]
    pub(super) fn handle_slash_retry(&mut self) -> Option<Cmd> {
        if self.agent_state != AgentState::Idle {
            self.status_message = Some("Cannot retry while processing".to_string());
            return None;
        }
        if !self.pending_inputs.is_empty() {
            self.status_message = Some(
                "Queued input is still pending; finish or restore it before retrying".to_string(),
            );
            return None;
        }
        let (session_id, expected_leaf_id) = {
            let Ok(session_guard) = self.session.try_lock() else {
                self.status_message = Some("Session busy; try again".to_string());
                return None;
            };
            if crate::checkpoint::plan_retry(&session_guard).is_none() {
                self.status_message = Some("No user turn to retry".to_string());
                return None;
            }
            (
                session_guard.header.id.clone(),
                session_guard.leaf_id().map(str::to_string),
            )
        };

        self.agent_state = AgentState::Processing;
        self.status_message = Some("Retrying last turn...".to_string());

        let session = Arc::clone(&self.session);
        let agent = Arc::clone(&self.agent);
        let event_tx = self.event_tx.clone();
        let extensions = self.extensions.clone();
        let save_enabled = self.save_enabled;
        let runtime_handle = self.runtime_handle.clone();

        runtime_handle.spawn_with_cx(move |cx| async move {
            if let Some(manager) = extensions {
                let cancelled = manager
                    .dispatch_cancellable_event(
                        ExtensionEventName::SessionBeforeSwitch,
                        Some(json!({
                            "reason": "retry",
                            "sessionId": session_id,
                        })),
                        EXTENSION_EVENT_TIMEOUT_MS,
                    )
                    .await
                    .unwrap_or(false);
                if cancelled {
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::System("Retry cancelled by extension".to_string()),
                    )
                    .await;
                    return;
                }
            }

            let mut agent_guard = match OwnedMutexGuard::lock(Arc::clone(&agent), &cx).await {
                Ok(guard) => guard,
                Err(err) => {
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::AgentError(format!("Failed to lock agent: {err}")),
                    )
                    .await;
                    return;
                }
            };
            let commit = match stage_and_commit_retry(
                Arc::clone(&session),
                &session_id,
                expected_leaf_id.as_deref(),
                save_enabled,
                &cx,
            )
            .await
            {
                Ok(commit) => commit,
                Err(err) => {
                    drop(agent_guard);
                    let _ = crate::interactive::enqueue_pi_event(
                        &event_tx,
                        &cx,
                        PiMsg::AgentError(format!("Retry could not be confirmed: {err}")),
                    )
                    .await;
                    return;
                }
            };
            agent_guard.replace_messages(commit.messages_for_agent);
            drop(agent_guard);

            let status = match commit.persistence {
                RetryPersistenceOutcome::Confirmed | RetryPersistenceOutcome::Disabled => {
                    Some("Retrying last turn".to_string())
                }
                RetryPersistenceOutcome::ReconciledButUnconfirmed => Some(
                    "Persistence warning: retry branch is present in the current disk and active session state, but final durability was not confirmed".to_string(),
                ),
            };
            let _ = crate::interactive::enqueue_pi_event(
                &event_tx,
                &cx,
                PiMsg::RetryCommitted {
                    session_id,
                    messages: commit.messages_for_ui,
                    usage: commit.usage,
                    text: commit.plan.text,
                    status,
                },
            )
            .await;
        });
        None
    }

    #[allow(clippy::too_many_lines)]
    pub(super) fn handle_slash_compact(&mut self, args: &str) -> Option<Cmd> {
        if self.agent_state != AgentState::Idle {
            self.status_message = Some("Cannot compact while processing".to_string());
            return None;
        }

        let Ok(agent_guard) = self.agent.try_lock() else {
            self.status_message = Some("Agent busy; try again".to_string());
            return None;
        };
        let provider = agent_guard.provider();
        let api_key_opt = agent_guard.stream_options().api_key.clone();
        drop(agent_guard);

        // Mode selection (bd-cv653.3.18): a leading `shake` drops bulky tool
        // results deterministically with zero LLM calls; `aggressive` runs the
        // LLM summary with a halved keep-recent window. Remaining words stay
        // custom instructions for the summarizer.
        let trimmed_args = args.trim();
        let first_token = trimmed_args.split_whitespace().next();
        let shake_mode = matches!(first_token, Some("shake" | "--shake"));
        let aggressive_mode = matches!(first_token, Some("aggressive" | "--aggressive"));
        let rest = if shake_mode || aggressive_mode {
            trimmed_args
                .split_once(char::is_whitespace)
                .map_or("", |(_, rest)| rest)
                .trim()
        } else {
            trimmed_args
        };

        if !shake_mode && api_key_opt.is_none() {
            self.status_message = Some("No API key configured; cannot run compaction".to_string());
            return None;
        }

        let event_tx = self.event_tx.clone();
        let session = Arc::clone(&self.session);
        let agent = Arc::clone(&self.agent);
        let extensions = self.extensions.clone();
        let runtime_handle = self.runtime_handle.clone();
        let completion_runtime_handle = runtime_handle.clone();
        let reserve_tokens = self.config.compaction_reserve_tokens();
        let save_enabled = self.save_enabled;
        let keep_recent_tokens = if aggressive_mode {
            self.config.compaction_keep_recent_tokens() / 2
        } else {
            self.config.compaction_keep_recent_tokens()
        };
        let custom_instructions = if rest.is_empty() {
            None
        } else {
            Some(rest.to_string())
        };
        let is_compacting = Arc::clone(&self.extension_compacting);

        self.agent_state = AgentState::Processing;
        self.status_message = Some("Compacting session...".to_string());
        self.extension_compacting
            .store(true, std::sync::atomic::Ordering::SeqCst);

        runtime_handle.spawn_with_cx(move |cx| async move {
            let (path_entries, expected_session_id, expected_leaf_id) = {
                let mut guard = match OwnedMutexGuard::lock(Arc::clone(&session), &cx).await {
                    Ok(guard) => guard,
                    Err(err) => {
                        is_compacting.store(false, std::sync::atomic::Ordering::SeqCst);
                        spawn_compaction_terminal_event(
                            &completion_runtime_handle,
                            event_tx.clone(),
                            PiMsg::AgentError(format!("Failed to lock session: {err}")),
                        );
                        return;
                    }
                };
                guard.ensure_entry_ids();
                (
                    guard
                        .entries_for_current_path()
                        .into_iter()
                        .cloned()
                        .collect::<Vec<_>>(),
                    guard.header.id.clone(),
                    guard.leaf_id().map(str::to_string),
                )
            };

            let settings = crate::compaction::ResolvedCompactionSettings {
                enabled: true,
                reserve_tokens,
                keep_recent_tokens,
                ..Default::default()
            };
            let Some(prep) = crate::compaction::prepare_compaction(&path_entries, settings) else {
                is_compacting.store(false, std::sync::atomic::Ordering::SeqCst);
                spawn_compaction_terminal_event(
                    &completion_runtime_handle,
                    event_tx.clone(),
                    PiMsg::System(
                        "Nothing to compact (already compacted or too little history)".to_string(),
                    ),
                );
                return;
            };

            let before_outcome = if let Some(manager) = extensions.clone() {
                let prep_value = crate::compaction::compaction_preparation_to_value(&prep);
                let branch_entries_value =
                    serde_json::to_value(&path_entries).unwrap_or(Value::Array(Vec::new()));
                let mut payload = serde_json::Map::new();
                payload.insert("preparation".to_string(), prep_value);
                payload.insert("branchEntries".to_string(), branch_entries_value);
                if let Some(custom_instructions) = custom_instructions.as_deref() {
                    payload.insert(
                        "customInstructions".to_string(),
                        Value::String(custom_instructions.to_string()),
                    );
                }

                let response = manager
                    .dispatch_event_with_response(
                        crate::extensions::ExtensionEventName::SessionBeforeCompact,
                        Some(Value::Object(payload)),
                        EXTENSION_EVENT_TIMEOUT_MS,
                    )
                    .await
                    .unwrap_or(None);
                apply_session_before_compact_response(response, prep.tokens_before)
            } else {
                SessionBeforeCompactOutcome::default()
            };

            if before_outcome.cancel {
                is_compacting.store(false, std::sync::atomic::Ordering::SeqCst);
                spawn_compaction_terminal_event(
                    &completion_runtime_handle,
                    event_tx.clone(),
                    PiMsg::System("Compaction cancelled by extension".to_string()),
                );
                return;
            }

            let (summary, first_kept_entry_id, tokens_before, details, from_extension) =
                if let Some(compaction) = before_outcome.compaction {
                    (
                        compaction.summary,
                        compaction.first_kept_entry_id,
                        compaction.tokens_before,
                        compaction.details,
                        true,
                    )
                } else {
                    let compact_outcome = if shake_mode {
                        Ok(crate::compaction::compact_shake(prep))
                    } else {
                        crate::compaction::compact(
                            prep,
                            Arc::clone(&provider),
                            api_key_opt.as_deref().unwrap_or_default(),
                            custom_instructions.as_deref(),
                        )
                        .await
                    };
                    let result = match compact_outcome {
                        Ok(result) => result,
                        Err(err) => {
                            is_compacting.store(false, std::sync::atomic::Ordering::SeqCst);
                            spawn_compaction_terminal_event(
                                &completion_runtime_handle,
                                event_tx.clone(),
                                PiMsg::AgentError(format!("Compaction failed: {err}")),
                            );
                            return;
                        }
                    };

                    let details =
                        crate::compaction::compaction_details_to_value(&result.details).ok();
                    (
                        result.summary,
                        result.first_kept_entry_id,
                        result.tokens_before,
                        details,
                        false,
                    )
                };

            let summary_tokens_after = crate::compaction::estimate_text_tokens(&summary);
            // Provider/extension work can consume or cancel the original task
            // context. Commit and terminal delivery therefore run in a fresh
            // runtime child so already-incurred work still resolves to exactly
            // one coherent Session/Agent/UI outcome.
            let completion_is_compacting = Arc::clone(&is_compacting);
            let fallback_event_tx = event_tx.clone();
            let completion_spawn = completion_runtime_handle.try_spawn_with_cx(
                move |completion_cx| async move {
                    // Acquire every fallible live-state lock before persistence.
                    // The staged helper either confirms/reconciles the exact
                    // operation or leaves the active Session untouched;
                    // installing the matching Agent transcript is then infallible.
                    let mut agent_guard = match OwnedMutexGuard::lock(
                        Arc::clone(&agent),
                        &completion_cx,
                    )
                    .await
                    {
                        Ok(guard) => guard,
                        Err(err) => {
                            completion_is_compacting
                                .store(false, std::sync::atomic::Ordering::SeqCst);
                            deliver_compaction_terminal_event(
                                &event_tx,
                                &completion_cx,
                                PiMsg::AgentError(format!("Failed to lock agent: {err}")),
                            )
                            .await;
                            return;
                        }
                    };
                    let from_hook = if from_extension { Some(true) } else { None };
                    let commit = match stage_and_commit_compaction_session(
                        Arc::clone(&session),
                        &expected_session_id,
                        expected_leaf_id.as_deref(),
                        summary,
                        first_kept_entry_id,
                        tokens_before,
                        details,
                        from_hook,
                        save_enabled,
                        &completion_cx,
                    )
                    .await
                    {
                        Ok(commit) => commit,
                        Err(err) => {
                            drop(agent_guard);
                            completion_is_compacting
                                .store(false, std::sync::atomic::Ordering::SeqCst);
                            deliver_compaction_terminal_event(
                                &event_tx,
                                &completion_cx,
                                PiMsg::AgentError(format!(
                                    "Compaction could not be confirmed: {err}"
                                )),
                            )
                            .await;
                            return;
                        }
                    };
                    let CompactionSessionCommit {
                        messages_for_agent,
                        messages_for_ui: messages,
                        usage,
                        compaction_entry,
                        persistence,
                    } = commit;
                    agent_guard.replace_messages(messages_for_agent);
                    drop(agent_guard);

                    completion_is_compacting
                        .store(false, std::sync::atomic::Ordering::SeqCst);
                    let label = if shake_mode {
                        "shake"
                    } else if aggressive_mode {
                        "aggressive"
                    } else {
                        "summary"
                    };
                    let status = match persistence {
                        CompactionPersistenceOutcome::Confirmed
                        | CompactionPersistenceOutcome::Disabled => format!(
                            "Compaction complete ({label}: {tokens_before} → ~{summary_tokens_after} tokens in compacted span)"
                        ),
                        CompactionPersistenceOutcome::ReconciledButUnconfirmed => format!(
                            "Persistence warning: compaction is present in the current disk and active session state, but final durability was not confirmed ({label}: {tokens_before} → ~{summary_tokens_after} tokens)"
                        ),
                    };
                    let delivered = deliver_compaction_terminal_event(
                        &event_tx,
                        &completion_cx,
                        PiMsg::ConversationReset {
                            session_id: expected_session_id.clone(),
                            messages,
                            usage,
                            status: Some(status),
                        },
                    )
                    .await;

                    if delivered
                        && persistence.may_emit_success_event()
                        && let Some(manager) = extensions
                    {
                        let _ = manager
                            .dispatch_event(
                                crate::extensions::ExtensionEventName::SessionCompact,
                                Some(json!({
                                    "compactionEntry": compaction_entry,
                                    "fromExtension": from_extension,
                                })),
                            )
                            .await;
                    }
                },
            );
            if let Err(err) = completion_spawn {
                is_compacting.store(false, std::sync::atomic::Ordering::SeqCst);
                tracing::error!(
                    error = %err,
                    "compaction completion could not be admitted by the runtime"
                );
                deliver_compaction_terminal_event(
                    &fallback_event_tx,
                    &cx,
                    PiMsg::AgentError(
                        "Compaction completion could not be admitted by the runtime".to_string(),
                    ),
                )
                .await;
            }
        });
        None
    }
}

// ---------------------------------------------------------------------------
// MessageRenderCache — per-message rendered content memoization (PERF-1)
// ---------------------------------------------------------------------------

use crate::interactive::state::MessageRole;
use std::cell::RefCell;

/// Lightweight cache key for a rendered conversation message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct MessageCacheKey {
    content_hash: u64,
    collapsed: bool,
    role: MessageRole,
}

/// Per-message render cache that stores the rendered output for each
/// `ConversationMessage`. Avoids re-rendering unchanged messages every frame.
///
/// Uses interior mutability (`RefCell`) because `view(&self)` cannot take
/// `&mut self` — same pattern as `FrameTimingStats`.
pub struct MessageRenderCache {
    /// Cached entries indexed by message position. `None` = cache miss.
    entries: RefCell<Vec<Option<(MessageCacheKey, String)>>>,
    /// Bumped on global invalidation: terminal resize, theme change,
    /// toggle-thinking, tool-expand toggle. All entries from a previous
    /// generation are considered stale.
    generation: std::cell::Cell<u64>,
    /// The generation at which each entry was cached. Stored separately
    /// to avoid duplicating generation in every entry.
    entry_generations: RefCell<Vec<u64>>,

    // -- PERF-2: Conversation prefix cache --
    // During streaming, only the tail (current_response/current_thinking)
    // changes. The prefix (all finalized messages) is cached here so
    // `build_conversation_content()` can skip re-iterating messages.
    /// Cached rendered content of all finalized messages.
    prefix: RefCell<String>,
    /// Number of messages included when the prefix was built.
    prefix_message_count: std::cell::Cell<usize>,
    /// The render-cache generation at which the prefix was built.
    /// If the generation has advanced, the prefix is stale.
    prefix_generation: std::cell::Cell<u64>,
}

impl MessageRenderCache {
    #[allow(clippy::missing_const_for_fn)]
    pub(super) fn new() -> Self {
        Self {
            entries: RefCell::new(Vec::new()),
            generation: std::cell::Cell::new(0),
            entry_generations: RefCell::new(Vec::new()),
            prefix: RefCell::new(String::new()),
            prefix_message_count: std::cell::Cell::new(0),
            prefix_generation: std::cell::Cell::new(0),
        }
    }

    /// Bump the generation counter, causing all cached entries and the
    /// conversation prefix to be considered stale on next lookup.
    /// O(1) — does not touch entries or the prefix buffer.
    pub(super) fn invalidate_all(&self) {
        self.generation.set(self.generation.get() + 1);
        // Prefix staleness is detected by comparing prefix_generation
        // with the current generation — no explicit flag needed.
    }

    /// Clear all cached entries and the prefix. Used on `/clear` or
    /// conversation reset.
    pub(super) fn clear(&self) {
        self.entries.borrow_mut().clear();
        self.entry_generations.borrow_mut().clear();
        self.prefix.borrow_mut().clear();
        self.prefix_message_count.set(0);
        self.prefix_generation.set(0);
    }

    /// Look up the cached rendered string for message at `index`.
    /// Returns `Some(&str)` on cache hit, `None` on miss.
    #[cfg(test)]
    pub(super) fn get(&self, index: usize, key: &MessageCacheKey) -> Option<String> {
        let entries = self.entries.borrow();
        let gens = self.entry_generations.borrow();
        if index >= entries.len() {
            return None;
        }
        let generation = self.generation.get();
        if gens[index] != generation {
            return None;
        }
        entries[index].as_ref().and_then(|(cached_key, rendered)| {
            if cached_key == key {
                Some(rendered.clone())
            } else {
                None
            }
        })
    }

    /// Append cached content for message at `index` directly into `output`.
    ///
    /// Returns `true` on cache hit, `false` on miss. This avoids allocating
    /// a cloned `String` on every hit in the view hot path.
    pub(super) fn append_cached(
        &self,
        output: &mut String,
        index: usize,
        key: &MessageCacheKey,
    ) -> bool {
        let entries = self.entries.borrow();
        let gens = self.entry_generations.borrow();
        if index >= entries.len() || index >= gens.len() {
            return false;
        }
        if gens[index] != self.generation.get() {
            return false;
        }
        if let Some((cached_key, rendered)) = &entries[index]
            && cached_key == key
        {
            output.push_str(rendered);
            return true;
        }
        false
    }

    /// Store a rendered string for message at `index`.
    pub(super) fn put(&self, index: usize, key: MessageCacheKey, rendered: String) {
        let mut entries = self.entries.borrow_mut();
        let mut gens = self.entry_generations.borrow_mut();
        // Grow vectors if needed.
        if index >= entries.len() {
            entries.resize_with(index + 1, || None);
            gens.resize(index + 1, 0);
        }
        let generation = self.generation.get();
        entries[index] = Some((key, rendered));
        gens[index] = generation;
    }

    /// Compute the cache key for a conversation message.
    pub(super) fn compute_key(
        msg: &super::ConversationMessage,
        thinking_visible: bool,
        tools_expanded: bool,
    ) -> MessageCacheKey {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::hash::DefaultHasher::new();
        msg.content.hash(&mut hasher);
        if thinking_visible && let Some(thinking) = &msg.thinking {
            thinking.hash(&mut hasher);
        }
        // Include tools_expanded in hash for tool messages since it affects rendering
        if msg.role == MessageRole::Tool {
            tools_expanded.hash(&mut hasher);
        }
        MessageCacheKey {
            content_hash: hasher.finish(),
            collapsed: msg.collapsed,
            role: msg.role,
        }
    }

    // -- PERF-2: Prefix cache accessors --

    /// Returns `true` if the cached prefix is still valid for the given
    /// message count. The prefix is stale when:
    /// - The message count changed (messages added/removed)
    /// - The render-cache generation advanced (theme/resize/toggle)
    /// - The prefix is empty and there are messages to render
    pub(super) const fn prefix_valid(&self, message_count: usize) -> bool {
        message_count > 0
            && self.prefix_message_count.get() == message_count
            && self.prefix_generation.get() == self.generation.get()
    }

    /// Return a clone of the cached prefix string.
    pub(super) fn prefix_get(&self) -> String {
        self.prefix.borrow().clone()
    }

    /// Append the cached prefix directly into `output`, avoiding a clone.
    ///
    /// PERF-7: This is the zero-copy alternative to `prefix_get()` for the
    /// streaming hot path where we build into a reusable buffer.
    pub(super) fn prefix_append_to(&self, output: &mut String) {
        output.push_str(&self.prefix.borrow());
    }

    /// Store a new prefix and snapshot the current message count / generation.
    pub(super) fn prefix_set(&self, content: &str, message_count: usize) {
        let mut p = self.prefix.borrow_mut();
        p.clear();
        p.push_str(content);
        self.prefix_message_count.set(message_count);
        self.prefix_generation.set(self.generation.get());
    }
}

// ---------------------------------------------------------------------------
// RenderBuffers — pre-allocated reusable buffers for view() hot path (PERF-7)
// ---------------------------------------------------------------------------

/// Pre-allocated buffers that are cleared and reused each frame, avoiding
/// repeated heap allocations in the 60fps render loop.
///
/// Uses `RefCell` for interior mutability because `view(&self)` cannot take
/// `&mut self` (same pattern as `FrameTimingStats` and `MessageRenderCache`).
pub struct RenderBuffers {
    /// Reusable buffer for `build_conversation_content()`.
    /// Taken via `std::mem::take`, built into, then returned.
    /// The buffer is put back (capacity preserved) after use.
    conversation: RefCell<String>,
    /// Reusable header buffer retained for render-buffer tests.
    #[cfg(test)]
    header: RefCell<String>,
    /// Reusable footer buffer retained for render-buffer tests.
    #[cfg(test)]
    footer: RefCell<String>,
    /// Capacity of the previous frame's final view output.
    /// Used to pre-allocate the next frame's output String via
    /// `String::with_capacity()`, avoiding incremental grows.
    view_capacity_hint: std::cell::Cell<usize>,
}

/// Default initial capacity for the view assembly buffer.
/// 80 columns x 24 rows x 4 bytes (UTF-8 + ANSI escapes).
const INITIAL_VIEW_CAPACITY: usize = 80 * 24 * 4;

/// Initial capacity for header/footer buffers (small: ~512 bytes typical).
#[cfg(test)]
const INITIAL_CHROME_CAPACITY: usize = 512;

impl RenderBuffers {
    pub(super) fn new() -> Self {
        Self {
            conversation: RefCell::new(String::with_capacity(INITIAL_VIEW_CAPACITY)),
            #[cfg(test)]
            header: RefCell::new(String::with_capacity(INITIAL_CHROME_CAPACITY)),
            #[cfg(test)]
            footer: RefCell::new(String::with_capacity(INITIAL_CHROME_CAPACITY)),
            view_capacity_hint: std::cell::Cell::new(INITIAL_VIEW_CAPACITY),
        }
    }

    /// Take the conversation buffer for reuse. The caller must put it back
    /// via [`return_conversation_buffer`] after building content.
    pub(super) fn take_conversation_buffer(&self) -> String {
        let mut buf = self.conversation.borrow_mut();
        let mut taken = std::mem::take(&mut *buf);
        taken.clear();
        taken
    }

    /// Return the conversation buffer after use, preserving its heap capacity.
    pub(super) fn return_conversation_buffer(&self, buf: String) {
        *self.conversation.borrow_mut() = buf;
    }

    /// Borrow the header buffer mutably, clearing it for reuse.
    /// The caller writes into the returned `RefMut` via `push_str` / `write!`.
    #[cfg(test)]
    pub(super) fn header_buf(&self) -> std::cell::RefMut<'_, String> {
        let mut buf = self.header.borrow_mut();
        buf.clear();
        buf
    }

    /// Borrow the footer buffer mutably, clearing it for reuse.
    #[cfg(test)]
    pub(super) fn footer_buf(&self) -> std::cell::RefMut<'_, String> {
        let mut buf = self.footer.borrow_mut();
        buf.clear();
        buf
    }

    /// Get the capacity hint for the next frame's view assembly.
    pub(super) const fn view_capacity_hint(&self) -> usize {
        self.view_capacity_hint.get()
    }

    /// Update the capacity hint after a frame completes.
    pub(super) fn set_view_capacity_hint(&self, capacity: usize) {
        self.view_capacity_hint.set(capacity);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::interactive::state::MessageRole;
    use asupersync::runtime::RuntimeBuilder;
    use std::collections::VecDeque;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, OnceLock};
    use tempfile::TempDir;

    fn runtime() -> &'static asupersync::runtime::Runtime {
        static RT: OnceLock<asupersync::runtime::Runtime> = OnceLock::new();
        RT.get_or_init(|| {
            RuntimeBuilder::multi_thread()
                .blocking_threads(1, 4)
                .build()
                .expect("build runtime")
        })
    }

    #[test]
    fn compaction_terminal_event_waits_for_capacity_instead_of_being_dropped() {
        let (event_tx, mut event_rx) = asupersync::channel::mpsc::channel(1);
        event_tx
            .try_send(PiMsg::System("occupy channel".to_string()))
            .expect("fill event channel");
        let runtime_handle = runtime().handle();
        spawn_compaction_terminal_event(
            &runtime_handle,
            event_tx,
            PiMsg::System("terminal compaction result".to_string()),
        );

        std::thread::sleep(std::time::Duration::from_millis(50));
        assert!(matches!(
            event_rx.try_recv(),
            Ok(PiMsg::System(message)) if message == "occupy channel"
        ));

        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
        loop {
            match event_rx.try_recv() {
                Ok(PiMsg::System(message)) if message == "terminal compaction result" => break,
                Ok(other) => {
                    panic!("unexpected event while awaiting compaction result: {other:?}")
                }
                Err(_) if std::time::Instant::now() < deadline => {
                    std::thread::sleep(std::time::Duration::from_millis(10));
                }
                Err(err) => {
                    panic!("compaction result was not delivered after capacity freed: {err}")
                }
            }
        }
    }

    #[test]
    fn staged_compaction_save_failure_leaves_live_session_exact() {
        let temp = TempDir::new().expect("tempdir");
        let blocked_path = temp.path().join("blocked.jsonl");
        std::fs::create_dir(&blocked_path).expect("create directory at session path");

        let mut raw_session = Session::create_with_dir(Some(temp.path().join("sessions")));
        raw_session.path = Some(blocked_path);
        let expected_session_id = raw_session.header.id.clone();
        let expected_leaf_id = raw_session.leaf_id().map(str::to_string);
        let expected_entries =
            serde_json::to_value(&raw_session.entries).expect("serialize entries");
        let session = Arc::new(Mutex::new(raw_session));
        let cx = Cx::for_testing();

        let error = runtime()
            .block_on(stage_and_commit_compaction_session(
                Arc::clone(&session),
                &expected_session_id,
                expected_leaf_id.as_deref(),
                "staged summary".to_string(),
                "first-kept".to_string(),
                42,
                None,
                None,
                true,
                &cx,
            ))
            .expect_err("directory session path must reject compaction save");
        assert!(!error.to_string().is_empty());

        runtime().block_on(async {
            let guard = OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock unchanged session");
            assert_eq!(guard.header.id, expected_session_id);
            assert_eq!(guard.leaf_id(), expected_leaf_id.as_deref());
            assert_eq!(
                serde_json::to_value(&guard.entries).expect("serialize live entries"),
                expected_entries
            );
            assert_eq!(guard.autosave_metrics().pending_mutations, 0);
        });
    }

    #[test]
    fn staged_compaction_rejects_stale_session_identity_before_mutation() {
        let session = Arc::new(Mutex::new(Session::in_memory()));
        let cx = Cx::for_testing();
        let error = runtime()
            .block_on(stage_and_commit_compaction_session(
                Arc::clone(&session),
                "replaced-session-id",
                None,
                "stale summary".to_string(),
                "first-kept".to_string(),
                1,
                None,
                None,
                false,
                &cx,
            ))
            .expect_err("stale compaction must fail closed");
        assert!(error.to_string().contains("Session changed"));

        runtime().block_on(async {
            let guard = OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock rejected session");
            assert!(guard.entries.is_empty());
            assert!(guard.leaf_id().is_none());
            assert_eq!(guard.autosave_metrics().pending_mutations, 0);
        });
    }

    #[test]
    fn staged_compaction_rejects_stale_leaf_before_mutation() {
        let mut raw_session = Session::in_memory();
        let session_id = raw_session.header.id.clone();
        let live_leaf = raw_session.append_message(crate::session::SessionMessage::User {
            content: crate::model::UserContent::Text("new live turn".to_string()),
            timestamp: Some(0),
        });
        let expected_entries =
            serde_json::to_value(&raw_session.entries).expect("serialize entries");
        let session = Arc::new(Mutex::new(raw_session));
        let cx = Cx::for_testing();

        let error = runtime()
            .block_on(stage_and_commit_compaction_session(
                Arc::clone(&session),
                &session_id,
                None,
                "stale summary".to_string(),
                "first-kept".to_string(),
                1,
                None,
                None,
                false,
                &cx,
            ))
            .expect_err("stale leaf must fail closed");
        assert!(error.to_string().contains("Session changed"));

        runtime().block_on(async {
            let guard = OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock rejected session");
            assert_eq!(guard.leaf_id(), Some(live_leaf.as_str()));
            assert_eq!(
                serde_json::to_value(&guard.entries).expect("serialize live entries"),
                expected_entries
            );
            assert_eq!(guard.autosave_metrics().pending_mutations, 1);
        });
    }

    #[test]
    fn staged_compaction_with_saving_disabled_commits_only_in_memory() {
        let temp = TempDir::new().expect("tempdir");
        let session_dir = temp.path().join("sessions");
        let raw_session = Session::create_with_dir(Some(session_dir.clone()));
        let expected_session_id = raw_session.header.id.clone();
        let session = Arc::new(Mutex::new(raw_session));
        let cx = Cx::for_testing();

        let commit = runtime()
            .block_on(stage_and_commit_compaction_session(
                Arc::clone(&session),
                &expected_session_id,
                None,
                "memory-only summary".to_string(),
                "first-kept".to_string(),
                7,
                None,
                Some(true),
                false,
                &cx,
            ))
            .expect("memory-only compaction");
        assert_eq!(commit.compaction_entry.summary, "memory-only summary");
        assert_eq!(commit.persistence, CompactionPersistenceOutcome::Disabled);

        runtime().block_on(async {
            let guard = OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock memory-only session");
            assert!(guard.path.is_none());
            assert!(matches!(
                guard.entries.as_slice(),
                [SessionEntry::Compaction(entry)] if entry.summary == "memory-only summary"
            ));
            assert_eq!(guard.autosave_metrics().pending_mutations, 1);
        });
        assert!(
            !session_dir.exists(),
            "--no-session compaction must not create durable state"
        );
    }

    #[test]
    fn staged_compaction_success_reopens_committed_entry() {
        let temp = TempDir::new().expect("tempdir");
        let session = Arc::new(Mutex::new(Session::create_with_dir(Some(
            temp.path().join("sessions"),
        ))));
        let cx = Cx::for_testing();
        let expected_session_id = runtime().block_on(async {
            OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock new session")
                .header
                .id
                .clone()
        });

        let persisted_path = runtime().block_on(async {
            let commit = stage_and_commit_compaction_session(
                Arc::clone(&session),
                &expected_session_id,
                None,
                "durable summary".to_string(),
                "first-kept".to_string(),
                99,
                None,
                None,
                true,
                &cx,
            )
            .await
            .expect("durable compaction");
            assert_eq!(commit.persistence, CompactionPersistenceOutcome::Confirmed);
            OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock saved session")
                .path
                .clone()
                .expect("saved path")
        });

        let reopened = runtime()
            .block_on(Session::open(persisted_path.to_string_lossy().as_ref()))
            .expect("reopen compacted session");
        assert!(matches!(
            reopened.entries.as_slice(),
            [SessionEntry::Compaction(entry)] if entry.summary == "durable summary"
        ));
    }

    #[test]
    fn post_error_reconciliation_requires_exact_compaction_identity() {
        let temp = TempDir::new().expect("tempdir");
        let mut candidate = Session::create_with_dir(Some(temp.path().join("sessions")));
        runtime()
            .block_on(candidate.save())
            .expect("persist baseline session");

        let entry_id = candidate.append_compaction(
            "durable despite terminal error".to_string(),
            "first-kept".to_string(),
            11,
            None,
            None,
        );
        let expected_entry = serde_json::to_vec(
            candidate
                .get_entry(&entry_id)
                .expect("staged compaction entry"),
        )
        .expect("serialize compaction entry");
        runtime()
            .block_on(candidate.save())
            .expect("persist simulated post-write state");

        assert!(
            runtime()
                .block_on(confirm_exact_compaction_after_save_error(
                    &candidate,
                    &entry_id,
                    &expected_entry,
                ))
                .is_some(),
            "exact durable operation should reconcile"
        );

        let mut wrong_surrounding_state = candidate.clone();
        wrong_surrounding_state.header.cwd.push_str("-different");
        assert!(
            runtime()
                .block_on(confirm_exact_compaction_after_save_error(
                    &wrong_surrounding_state,
                    &entry_id,
                    &expected_entry,
                ))
                .is_none(),
            "reconciliation must reject a matching operation on mismatched session state"
        );

        let mut wrong_entry = expected_entry;
        wrong_entry.push(b' ');
        assert!(
            runtime()
                .block_on(confirm_exact_compaction_after_save_error(
                    &candidate,
                    &entry_id,
                    &wrong_entry,
                ))
                .is_none(),
            "reconciliation must reject a mismatched operation payload"
        );
    }

    fn session_user(text: &str) -> crate::session::SessionMessage {
        crate::session::SessionMessage::User {
            content: crate::model::UserContent::Text(text.to_string()),
            timestamp: Some(0),
        }
    }

    fn session_assistant(text: &str) -> crate::session::SessionMessage {
        crate::session::SessionMessage::from(crate::model::Message::Assistant(std::sync::Arc::new(
            crate::model::AssistantMessage {
                content: vec![crate::model::ContentBlock::Text(
                    crate::model::TextContent::new(text),
                )],
                ..Default::default()
            },
        )))
    }

    fn linear_retry_session(session_dir: Option<std::path::PathBuf>) -> (Session, String, String) {
        let mut session = session_dir.map_or_else(Session::in_memory, |dir| {
            Session::create_with_dir(Some(dir))
        });
        session.append_message(session_user("first question"));
        let first_answer = session.append_message(session_assistant("first answer"));
        let abandoned = session.append_message(session_user("second question"));
        session.append_message(session_assistant("second answer"));
        (session, first_answer, abandoned)
    }

    #[test]
    fn staged_retry_save_failure_leaves_live_session_exact() {
        let temp = TempDir::new().expect("tempdir");
        let blocked_path = temp.path().join("blocked.jsonl");
        std::fs::create_dir(&blocked_path).expect("create directory at session path");
        let (mut raw_session, _parent, _abandoned) =
            linear_retry_session(Some(temp.path().join("sessions")));
        raw_session.path = Some(blocked_path);
        let expected_session_id = raw_session.header.id.clone();
        let expected_leaf_id = raw_session.leaf_id().map(str::to_string);
        let expected_entries =
            serde_json::to_value(&raw_session.entries).expect("serialize entries");
        let session = Arc::new(Mutex::new(raw_session));
        let cx = Cx::for_testing();

        let error = runtime()
            .block_on(stage_and_commit_retry(
                Arc::clone(&session),
                &expected_session_id,
                expected_leaf_id.as_deref(),
                true,
                &cx,
            ))
            .expect_err("directory session path must reject retry save");
        assert!(
            !error.to_string().is_empty(),
            "retry save failure must be observable"
        );

        runtime().block_on(async {
            let guard = OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock unchanged session");
            assert_eq!(guard.header.id, expected_session_id);
            assert_eq!(guard.leaf_id(), expected_leaf_id.as_deref());
            assert_eq!(
                serde_json::to_value(&guard.entries).expect("serialize live entries"),
                expected_entries
            );
        });
    }

    #[test]
    fn staged_retry_with_saving_disabled_commits_sibling_parent_in_memory() {
        let (raw_session, first_answer, abandoned) = linear_retry_session(None);
        let expected_session_id = raw_session.header.id.clone();
        let expected_leaf_id = raw_session.leaf_id().map(str::to_string);
        let session = Arc::new(Mutex::new(raw_session));
        let cx = Cx::for_testing();

        let commit = runtime()
            .block_on(stage_and_commit_retry(
                Arc::clone(&session),
                &expected_session_id,
                expected_leaf_id.as_deref(),
                false,
                &cx,
            ))
            .expect("memory-only retry");
        assert_eq!(commit.plan.abandoned_entry_id, abandoned);
        assert_eq!(commit.plan.text, "second question");
        assert_eq!(commit.persistence, RetryPersistenceOutcome::Disabled);
        assert_eq!(
            commit.plan.expected_parent_id.as_deref(),
            Some(first_answer.as_str())
        );

        runtime().block_on(async {
            let guard = OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock retried session");
            assert_eq!(guard.leaf_id(), Some(first_answer.as_str()));
            let abandoned_on_path = guard
                .entries_for_current_path()
                .iter()
                .any(|entry| entry.base().id.as_ref() == Some(&abandoned));
            assert!(!abandoned_on_path);
            assert!(guard.get_entry(&abandoned).is_some());
        });
    }

    #[test]
    fn staged_retry_save_reopen_keeps_sibling_parent_topology() {
        let temp = tempfile::Builder::new()
            .prefix("pi-r7icz-")
            .tempdir_in("/tmp")
            .unwrap_or_else(|_| TempDir::new().expect("tempdir"));
        let session_dir = temp.path().join("sessions");
        let (mut raw_session, first_answer, abandoned) = linear_retry_session(Some(session_dir));
        raw_session.path = Some(temp.path().join("session.jsonl"));
        let expected_session_id = raw_session.header.id.clone();
        let expected_leaf_id = raw_session.leaf_id().map(str::to_string);
        let cx = Cx::for_testing();
        runtime()
            .block_on(raw_session.save())
            .expect("pin baseline retry session");
        let session = Arc::new(Mutex::new(raw_session));

        let commit = runtime()
            .block_on(stage_and_commit_retry(
                Arc::clone(&session),
                &expected_session_id,
                expected_leaf_id.as_deref(),
                true,
                &cx,
            ))
            .expect("durable retry");
        assert_eq!(commit.persistence, RetryPersistenceOutcome::Confirmed);

        let persisted_path = runtime().block_on(async {
            OwnedMutexGuard::lock(Arc::clone(&session), &cx)
                .await
                .expect("lock saved session")
                .path
                .clone()
                .expect("saved path")
        });
        let reopened = runtime()
            .block_on(Session::open(persisted_path.to_string_lossy().as_ref()))
            .expect("reopen retried session");
        assert_eq!(reopened.leaf_id(), Some(first_answer.as_str()));
        let abandoned_on_path = reopened
            .entries_for_current_path()
            .iter()
            .any(|entry| entry.base().id.as_ref() == Some(&abandoned));
        assert!(!abandoned_on_path);
        assert!(reopened.get_entry(&abandoned).is_some());
        let mut live = reopened;
        let retried = live.append_message(session_user("second question"));
        let retried_parent = live
            .get_entry(&retried)
            .and_then(|entry| entry.base().parent_id.clone());
        assert_eq!(retried_parent.as_deref(), Some(first_answer.as_str()));
    }

    // ========================================================================
    // FrameTimingStats unit tests (PERF-3)
    // ========================================================================

    fn make_stats(enabled: bool) -> FrameTimingStats {
        FrameTimingStats {
            frame_times_us: std::cell::RefCell::new(VecDeque::new()),
            content_build_times_us: std::cell::RefCell::new(VecDeque::new()),
            viewport_sync_times_us: std::cell::RefCell::new(VecDeque::new()),
            update_times_us: VecDeque::new(),
            total_frames: std::cell::Cell::new(0),
            budget_exceeded_count: std::cell::Cell::new(0),
            enabled,
        }
    }

    #[test]
    fn frame_timing_disabled_by_default() {
        let stats = make_stats(false);
        stats.record_frame(5000);
        assert_eq!(stats.total_frames.get(), 0);
        assert!(stats.frame_times_us.borrow().is_empty());
    }

    #[test]
    fn frame_timing_records_when_enabled() {
        let stats = make_stats(true);
        stats.record_frame(5000);
        stats.record_frame(10_000);
        stats.record_frame(20_000);
        assert_eq!(stats.total_frames.get(), 3);
        assert_eq!(stats.budget_exceeded_count.get(), 1);
        assert_eq!(stats.frame_times_us.borrow().len(), 3);
    }

    #[test]
    fn frame_timing_content_build_records() {
        let stats = make_stats(true);
        stats.record_content_build(1500);
        stats.record_content_build(2500);
        assert_eq!(stats.content_build_times_us.borrow().len(), 2);
    }

    #[test]
    fn frame_timing_viewport_sync_records() {
        let stats = make_stats(true);
        stats.record_viewport_sync(800);
        stats.record_viewport_sync(1200);
        assert_eq!(stats.viewport_sync_times_us.borrow().len(), 2);
    }

    #[test]
    fn frame_timing_update_records() {
        let mut stats = make_stats(true);
        stats.record_update(500);
        stats.record_update(1000);
        assert_eq!(stats.update_times_us.len(), 2);
    }

    #[test]
    fn frame_timing_rolling_window_evicts_oldest() {
        let stats = make_stats(true);
        for i in 0..=FRAME_TIMING_WINDOW as u64 {
            stats.record_frame(i * 100);
        }
        assert_eq!(stats.frame_times_us.borrow().len(), FRAME_TIMING_WINDOW);
        assert_eq!(*stats.frame_times_us.borrow().front().unwrap(), 100);
    }

    #[test]
    fn frame_timing_percentiles_empty() {
        let empty = VecDeque::new();
        assert_eq!(FrameTimingStats::percentiles(&empty), (0, 0, 0, 0));
    }

    #[test]
    fn frame_timing_percentiles_single_value() {
        let mut times = VecDeque::new();
        times.push_back(5000);
        assert_eq!(
            FrameTimingStats::percentiles(&times),
            (5000, 5000, 5000, 5000)
        );
    }

    #[test]
    fn frame_timing_percentiles_known_distribution() {
        let mut times = VecDeque::new();
        for i in 1..=100 {
            times.push_back(i * 1000);
        }
        let (p50, p95, p99, p999) = FrameTimingStats::percentiles(&times);
        assert_eq!(p50, 51_000);
        assert_eq!(p95, 96_000);
        assert_eq!(p99, 100_000);
        assert_eq!(p999, 100_000);
    }

    #[test]
    fn frame_timing_summary_disabled() {
        let stats = make_stats(false);
        assert!(stats.summary().contains("disabled"));
    }

    #[test]
    fn frame_timing_summary_enabled_contains_stats() {
        let stats = make_stats(true);
        stats.record_frame(5000);
        stats.record_content_build(2000);
        let summary = stats.summary();
        assert!(summary.contains("Frame timing"));
        assert!(summary.contains("view()"));
        assert!(summary.contains("content"));
        assert!(summary.contains("viewport"));
        assert!(summary.contains("update()"));
        assert!(summary.contains("p999"));
        assert!(summary.contains("Budget exceeded"));
    }

    #[test]
    fn frame_timing_snapshot_is_structured_and_redaction_safe() {
        let mut stats = make_stats(false);
        stats.enable_for_test();
        stats.record_frame(8_000);
        stats.record_frame(FRAME_BUDGET_US + 25);
        stats.record_content_build(3_000);
        stats.record_viewport_sync(1_000);
        stats.record_update(750);

        let fixture = json!({
            "message_count": 600,
            "tool_preview_count": 12,
            "model_count": 80,
            "branch_count": 64,
        });
        let snapshot = stats.snapshot_json("large_conversation", &fixture);

        assert_eq!(snapshot["schema"], "pi.tui.frame_budget.v1");
        assert_eq!(snapshot["surface"], "large_conversation");
        assert_eq!(snapshot["enabled"], true);
        assert_eq!(snapshot["samples"]["frame"]["count"], 2);
        assert_eq!(snapshot["samples"]["frame"]["over_budget_count"], 1);
        assert_eq!(snapshot["verdict"], "warn");
        assert_eq!(snapshot["redaction"]["prompt_content"], "omitted");
        assert_eq!(snapshot["redaction"]["tool_payload_content"], "omitted");
        let serialized = snapshot.to_string();
        assert!(
            !serialized.contains("payload content"),
            "snapshot must not include prompt or tool payload text"
        );
    }

    #[test]
    fn frame_timing_budget_exceeded_counts_correctly() {
        let stats = make_stats(true);
        stats.record_frame(10_000);
        stats.record_frame(16_000);
        stats.record_frame(FRAME_BUDGET_US);
        assert_eq!(stats.budget_exceeded_count.get(), 0);
        stats.record_frame(FRAME_BUDGET_US + 1);
        stats.record_frame(20_000);
        assert_eq!(stats.budget_exceeded_count.get(), 2);
    }

    #[test]
    fn frame_timing_exposes_current_p99() {
        let stats = make_stats(true);
        stats.record_frame(8_000);
        stats.record_frame(12_000);
        stats.record_frame(40_000);
        assert_eq!(stats.frame_p99_us(), 40_000);
    }

    #[test]
    fn tui_pressure_controller_uses_latency_output_and_pending_events() {
        assert_eq!(
            TuiPressureController::decide(0, 1024, 0).level,
            TuiPressureLevel::Normal
        );

        let by_frame =
            TuiPressureController::decide(TuiPressureController::ELEVATED_FRAME_P99_US, 1024, 0);
        assert_eq!(by_frame.level, TuiPressureLevel::Elevated);
        assert!(by_frame.throttle_tool_updates);

        let by_output =
            TuiPressureController::decide(0, TuiPressureController::ELEVATED_TOOL_OUTPUT_BYTES, 0);
        assert_eq!(by_output.level, TuiPressureLevel::Elevated);

        let by_pending =
            TuiPressureController::decide(0, 1024, TuiPressureController::HIGH_PENDING_TOOL_EVENTS);
        assert_eq!(by_pending.level, TuiPressureLevel::High);
        assert!(by_pending.flush_interval > by_frame.flush_interval);
    }

    // ========================================================================
    // MemoryMonitor unit tests (PERF-6)
    // ========================================================================

    struct MockRssReader {
        value: Arc<AtomicUsize>,
    }

    impl MockRssReader {
        fn new(initial: usize) -> (Self, Arc<AtomicUsize>) {
            let shared = Arc::new(AtomicUsize::new(initial));
            (
                Self {
                    value: Arc::clone(&shared),
                },
                shared,
            )
        }
    }

    impl RssReader for MockRssReader {
        fn read_rss_bytes(&self) -> Option<usize> {
            Some(self.value.load(Ordering::Relaxed))
        }
    }

    fn make_memory_monitor(initial_rss: usize) -> (MemoryMonitor, Arc<AtomicUsize>) {
        let (reader, shared) = MockRssReader::new(initial_rss);
        let mut monitor = MemoryMonitor::new(Box::new(reader));
        monitor.sample_interval = std::time::Duration::ZERO;
        (monitor, shared)
    }

    #[test]
    fn memory_level_classification() {
        assert_eq!(MemoryLevel::from_rss_bytes(0), MemoryLevel::Normal);
        assert_eq!(MemoryLevel::from_rss_bytes(30_000_000), MemoryLevel::Normal);
        assert_eq!(MemoryLevel::from_rss_bytes(49_999_999), MemoryLevel::Normal);
        assert_eq!(
            MemoryLevel::from_rss_bytes(50_000_000),
            MemoryLevel::Warning
        );
        assert_eq!(
            MemoryLevel::from_rss_bytes(99_999_999),
            MemoryLevel::Warning
        );
        assert_eq!(
            MemoryLevel::from_rss_bytes(100_000_000),
            MemoryLevel::Pressure
        );
        assert_eq!(
            MemoryLevel::from_rss_bytes(199_999_999),
            MemoryLevel::Pressure
        );
        assert_eq!(
            MemoryLevel::from_rss_bytes(200_000_000),
            MemoryLevel::Critical
        );
        assert_eq!(
            MemoryLevel::from_rss_bytes(500_000_000),
            MemoryLevel::Critical
        );
    }

    #[test]
    fn memory_monitor_sampling_tracks_rss_and_peak() {
        let (mut monitor, shared) = make_memory_monitor(30_000_000);
        monitor.maybe_sample();
        assert_eq!(monitor.current_rss_bytes, 30_000_000);
        assert_eq!(monitor.peak_rss_bytes, 30_000_000);
        assert_eq!(monitor.level, MemoryLevel::Normal);

        shared.store(60_000_000, Ordering::Relaxed);
        monitor.maybe_sample();
        assert_eq!(monitor.current_rss_bytes, 60_000_000);
        assert_eq!(monitor.peak_rss_bytes, 60_000_000);
        assert_eq!(monitor.level, MemoryLevel::Warning);

        shared.store(20_000_000, Ordering::Relaxed);
        monitor.maybe_sample();
        assert_eq!(monitor.current_rss_bytes, 20_000_000);
        assert_eq!(monitor.peak_rss_bytes, 60_000_000);
        assert_eq!(monitor.level, MemoryLevel::Normal);
    }

    #[test]
    fn memory_monitor_pressure_starts_collapsing() {
        let (mut monitor, shared) = make_memory_monitor(10_000_000);
        monitor.maybe_sample();
        assert!(!monitor.collapsing);

        shared.store(120_000_000, Ordering::Relaxed);
        monitor.maybe_sample();
        assert_eq!(monitor.level, MemoryLevel::Pressure);
        assert!(monitor.collapsing);
    }

    #[test]
    fn memory_monitor_hysteresis_stops_collapsing() {
        let (mut monitor, shared) = make_memory_monitor(120_000_000);
        monitor.maybe_sample();
        assert!(monitor.collapsing);

        // 70MB < 80MB relief threshold => collapsing stops.
        // Level is Warning (50-100MB), not Normal.
        shared.store(70_000_000, Ordering::Relaxed);
        monitor.resample_now();
        assert!(!monitor.collapsing);
        assert_eq!(monitor.level, MemoryLevel::Warning);

        // Drop fully below 50MB => Normal.
        shared.store(30_000_000, Ordering::Relaxed);
        monitor.resample_now();
        assert!(!monitor.collapsing);
        assert_eq!(monitor.level, MemoryLevel::Normal);
    }

    #[test]
    fn memory_monitor_summary_format() {
        let (mut monitor, _) = make_memory_monitor(55_000_000);
        monitor.maybe_sample();
        let summary = monitor.summary();
        assert!(summary.contains("55.0MB"));
        assert!(summary.contains("Warning"));
    }

    /// bd-1h9dp: the RSS reading must be plausible for the test process
    /// (Linux via /proc, macOS/others via sysinfo).
    #[test]
    fn proc_self_rss_reader_reports_plausible_rss() {
        let rss = ProcSelfRssReader.read_rss_bytes();
        assert!(
            rss.is_some_and(|bytes| bytes > 1_000_000),
            "expected a plausible RSS reading for the test process, got {rss:?}"
        );
    }

    #[test]
    fn memory_monitor_should_force_degraded_only_at_critical() {
        let (mut monitor, shared) = make_memory_monitor(10_000_000);
        monitor.maybe_sample();
        assert!(!monitor.should_force_degraded());

        shared.store(60_000_000, Ordering::Relaxed);
        monitor.maybe_sample();
        assert!(!monitor.should_force_degraded());

        shared.store(150_000_000, Ordering::Relaxed);
        monitor.maybe_sample();
        assert!(!monitor.should_force_degraded());

        shared.store(250_000_000, Ordering::Relaxed);
        monitor.maybe_sample();
        assert!(monitor.should_force_degraded());
    }

    #[test]
    fn memory_progressive_collapse_ordering() {
        let messages = [
            ConversationMessage::new(MessageRole::User, "hello".into(), None),
            ConversationMessage::new(MessageRole::Tool, "output 1".into(), None),
            ConversationMessage::new(MessageRole::Assistant, "response".into(), None),
            ConversationMessage::new(MessageRole::Tool, "output 2".into(), None),
            ConversationMessage::new(MessageRole::Tool, "output 3".into(), None),
        ];
        let mut next_idx = 0usize;
        let mut found = Vec::new();
        loop {
            let result = messages[next_idx..]
                .iter()
                .enumerate()
                .find(|(_, m)| matches!(m.role, MessageRole::Tool) && !m.collapsed)
                .map(|(i, _)| next_idx + i);
            match result {
                Some(idx) => {
                    found.push(idx);
                    next_idx = idx + 1;
                }
                None => break,
            }
        }
        assert_eq!(found, vec![1, 3, 4]);
    }

    #[test]
    fn memory_critical_truncation_keeps_last_messages() {
        let mut messages: Vec<ConversationMessage> = (0..50)
            .map(|i| {
                ConversationMessage::new(
                    if i % 2 == 0 {
                        MessageRole::User
                    } else {
                        MessageRole::Assistant
                    },
                    format!("msg {i}"),
                    None,
                )
            })
            .collect();

        let msg_count = messages.len();
        assert!(msg_count > CRITICAL_KEEP_MESSAGES);
        let remove_count = msg_count - CRITICAL_KEEP_MESSAGES;
        messages.drain(..remove_count);
        messages.insert(
            0,
            ConversationMessage::new(MessageRole::System, "[truncated]".into(), None),
        );

        assert_eq!(messages[0].role, MessageRole::System);
        assert!(messages[0].content.contains("truncated"));
        assert_eq!(messages.len(), CRITICAL_KEEP_MESSAGES + 1);
        assert_eq!(messages.last().unwrap().content, "msg 49");
    }

    // ========================================================================
    // Cross-platform fallback tests (PERF-CROSS-PLATFORM / bd-32sj0)
    // Verify MemoryMonitor degrades gracefully when RssReader returns None
    // (i.e. on non-Linux platforms where /proc/self/statm is unavailable).
    // ========================================================================

    /// An RssReader that always returns None — simulates non-Linux platforms.
    struct NullRssReader;

    impl RssReader for NullRssReader {
        fn read_rss_bytes(&self) -> Option<usize> {
            None
        }
    }

    fn make_null_memory_monitor() -> MemoryMonitor {
        let mut monitor = MemoryMonitor::new(Box::new(NullRssReader));
        monitor.sample_interval = std::time::Duration::ZERO;
        monitor
    }

    #[test]
    fn memory_monitor_null_reader_stays_normal() {
        let mut monitor = make_null_memory_monitor();
        // maybe_sample should return false (no level change) when reader returns None.
        assert!(!monitor.maybe_sample());
        assert_eq!(monitor.level, MemoryLevel::Normal);
        assert_eq!(monitor.current_rss_bytes, 0);
        assert_eq!(monitor.peak_rss_bytes, 0);
        assert!(!monitor.collapsing);
        assert!(!monitor.should_force_degraded());
    }

    #[test]
    fn memory_monitor_null_reader_repeated_sampling_stable() {
        let mut monitor = make_null_memory_monitor();
        // Many sampling cycles should not cause drift, panic, or state corruption.
        for _ in 0..100 {
            assert!(!monitor.maybe_sample());
        }
        assert_eq!(monitor.level, MemoryLevel::Normal);
        assert_eq!(monitor.current_rss_bytes, 0);
        assert_eq!(monitor.peak_rss_bytes, 0);
    }

    #[test]
    fn memory_monitor_null_reader_resample_now_no_panic() {
        let mut monitor = make_null_memory_monitor();
        // resample_now should silently do nothing when reader returns None.
        monitor.resample_now();
        assert_eq!(monitor.level, MemoryLevel::Normal);
        assert_eq!(monitor.current_rss_bytes, 0);
    }

    #[test]
    fn memory_monitor_null_reader_summary_shows_zero() {
        let mut monitor = make_null_memory_monitor();
        monitor.maybe_sample();
        let summary = monitor.summary();
        assert!(
            summary.contains("0.0MB"),
            "Summary should show 0.0MB when no RSS available, got: {summary}"
        );
        assert!(
            summary.contains("Normal"),
            "Summary should show Normal level when no RSS available, got: {summary}"
        );
    }

    #[test]
    fn frame_timing_operates_independently_of_memory_pressure() {
        // FrameTimingStats does not depend on MemoryMonitor or CPU pressure.
        // It should work correctly even when memory monitoring is unavailable.
        let stats = make_stats(true);
        // Simulate a realistic frame sequence.
        stats.record_frame(8_000);
        stats.record_frame(12_000);
        stats.record_frame(FRAME_BUDGET_US + 500);
        stats.record_content_build(3_000);
        stats.record_viewport_sync(1_500);
        // Verify all counters updated correctly.
        assert_eq!(stats.total_frames.get(), 3);
        assert_eq!(stats.budget_exceeded_count.get(), 1);
        assert_eq!(stats.content_build_times_us.borrow().len(), 1);
        assert_eq!(stats.viewport_sync_times_us.borrow().len(), 1);
        // Summary should produce valid output without any memory/CPU context.
        let summary = stats.summary();
        assert!(
            summary.contains("Frame timing"),
            "Summary should work without memory pressure context"
        );
        assert!(
            summary.contains("Budget exceeded: 1"),
            "Budget exceeded count should be accurate"
        );
    }

    #[test]
    fn proc_self_rss_reader_returns_some_on_all_platforms() {
        // bd-1h9dp: non-Linux used to return None, leaving the entire
        // memory-pressure degradation path silently inert off Linux.
        let reader = ProcSelfRssReader;
        let result = reader.read_rss_bytes();
        assert!(result.is_some());
    }

    // --- MessageRenderCache tests (PERF-1) ---

    #[test]
    fn cache_hit_returns_same_content() {
        let cache = MessageRenderCache::new();
        let msg = ConversationMessage::new(MessageRole::User, "Hello".to_string(), None);
        let key = MessageRenderCache::compute_key(&msg, false, true);
        cache.put(0, key.clone(), "rendered-hello".to_string());
        assert_eq!(cache.get(0, &key), Some("rendered-hello".to_string()));
    }

    #[test]
    fn append_cached_writes_output_on_hit() {
        let cache = MessageRenderCache::new();
        let msg = ConversationMessage::new(MessageRole::User, "Hello".to_string(), None);
        let key = MessageRenderCache::compute_key(&msg, false, true);
        cache.put(0, key.clone(), "rendered-hello".to_string());

        let mut output = String::new();
        assert!(cache.append_cached(&mut output, 0, &key));
        assert_eq!(output, "rendered-hello");
    }

    #[test]
    fn append_cached_noop_on_miss() {
        let cache = MessageRenderCache::new();
        let msg = ConversationMessage::new(MessageRole::User, "Hello".to_string(), None);
        let key = MessageRenderCache::compute_key(&msg, false, true);

        let mut output = String::new();
        assert!(!cache.append_cached(&mut output, 0, &key));
        assert!(output.is_empty());
    }

    #[test]
    fn cache_miss_after_content_change() {
        let cache = MessageRenderCache::new();
        let msg1 = ConversationMessage::new(MessageRole::User, "Hello".to_string(), None);
        let key1 = MessageRenderCache::compute_key(&msg1, false, true);
        cache.put(0, key1, "rendered-hello".to_string());

        let msg2 = ConversationMessage::new(MessageRole::User, "Goodbye".to_string(), None);
        let key2 = MessageRenderCache::compute_key(&msg2, false, true);
        assert_eq!(cache.get(0, &key2), None);
    }

    #[test]
    fn tool_message_cache_miss_when_collapse_toggles() {
        let cache = MessageRenderCache::new();
        let mut msg = ConversationMessage::tool("Tool bash:\nline1\nline2".to_string());
        let key_expanded = MessageRenderCache::compute_key(&msg, false, true);
        cache.put(0, key_expanded.clone(), "expanded-output".to_string());

        // Toggle collapse
        msg.collapsed = !msg.collapsed;
        let key_collapsed = MessageRenderCache::compute_key(&msg, false, true);
        assert_ne!(key_expanded, key_collapsed);
        assert_eq!(cache.get(0, &key_collapsed), None);
    }

    #[test]
    fn generation_bump_forces_full_miss() {
        let cache = MessageRenderCache::new();
        let msg = ConversationMessage::new(MessageRole::Assistant, "Response".to_string(), None);
        let key = MessageRenderCache::compute_key(&msg, false, true);
        cache.put(0, key.clone(), "old-render".to_string());

        // Simulate terminal resize → generation bump
        cache.invalidate_all();
        assert_eq!(cache.get(0, &key), None);
    }

    #[test]
    fn clear_removes_all_entries() {
        let cache = MessageRenderCache::new();
        let msg = ConversationMessage::new(MessageRole::User, "Hello".to_string(), None);
        let key = MessageRenderCache::compute_key(&msg, false, true);
        cache.put(0, key.clone(), "rendered".to_string());
        cache.put(1, key.clone(), "rendered2".to_string());
        cache.clear();
        assert_eq!(cache.get(0, &key), None);
        assert_eq!(cache.get(1, &key), None);
    }

    #[test]
    fn thinking_visibility_changes_key() {
        let msg = ConversationMessage::new(
            MessageRole::Assistant,
            "Response".to_string(),
            Some("Thinking...".to_string()),
        );
        let key_visible = MessageRenderCache::compute_key(&msg, true, true);
        let key_hidden = MessageRenderCache::compute_key(&msg, false, true);
        assert_ne!(
            key_visible, key_hidden,
            "Thinking visibility should change the key"
        );
    }

    #[test]
    fn tools_expanded_changes_key_for_tool_messages() {
        let msg = ConversationMessage::tool("Tool output\nline1\nline2".to_string());
        let key_expanded = MessageRenderCache::compute_key(&msg, false, true);
        let key_collapsed = MessageRenderCache::compute_key(&msg, false, false);
        assert_ne!(
            key_expanded, key_collapsed,
            "tools_expanded should change key for tool messages"
        );
    }

    #[test]
    fn out_of_bounds_index_returns_none() {
        let cache = MessageRenderCache::new();
        let msg = ConversationMessage::new(MessageRole::User, "Hello".to_string(), None);
        let key = MessageRenderCache::compute_key(&msg, false, true);
        assert_eq!(cache.get(42, &key), None);
    }

    // --- Prefix cache tests (PERF-2) ---

    #[test]
    fn prefix_initially_invalid() {
        let cache = MessageRenderCache::new();
        assert!(!cache.prefix_valid(0));
        assert!(!cache.prefix_valid(1));
    }

    #[test]
    fn prefix_valid_after_set() {
        let cache = MessageRenderCache::new();
        cache.prefix_set("rendered-prefix", 5);
        assert!(cache.prefix_valid(5));
        assert_eq!(cache.prefix_get(), "rendered-prefix");
    }

    #[test]
    fn prefix_invalid_after_message_count_change() {
        let cache = MessageRenderCache::new();
        cache.prefix_set("prefix-for-5", 5);
        assert!(cache.prefix_valid(5));
        // New message added → count changed
        assert!(!cache.prefix_valid(6));
    }

    #[test]
    fn prefix_invalid_after_invalidate_all() {
        let cache = MessageRenderCache::new();
        cache.prefix_set("prefix", 3);
        assert!(cache.prefix_valid(3));
        // Simulate theme change / resize / toggle
        cache.invalidate_all();
        assert!(!cache.prefix_valid(3));
    }

    #[test]
    fn prefix_cleared_on_clear() {
        let cache = MessageRenderCache::new();
        cache.prefix_set("prefix", 3);
        cache.clear();
        assert!(!cache.prefix_valid(3));
        assert!(cache.prefix_get().is_empty());
    }

    #[test]
    fn prefix_revalidates_after_rebuild() {
        let cache = MessageRenderCache::new();
        cache.prefix_set("old-prefix", 3);
        cache.invalidate_all();
        assert!(!cache.prefix_valid(3));
        // Full rebuild sets new prefix
        cache.prefix_set("new-prefix", 3);
        assert!(cache.prefix_valid(3));
        assert_eq!(cache.prefix_get(), "new-prefix");
    }

    // ========================================================================
    // RenderBuffers unit tests (PERF-7)
    // ========================================================================

    #[test]
    fn render_buffers_initial_capacity_hint() {
        let rb = RenderBuffers::new();
        assert_eq!(rb.view_capacity_hint(), INITIAL_VIEW_CAPACITY);
    }

    #[test]
    fn render_buffers_capacity_hint_updates() {
        let rb = RenderBuffers::new();
        rb.set_view_capacity_hint(12_345);
        assert_eq!(rb.view_capacity_hint(), 12_345);
    }

    #[test]
    fn render_buffers_take_returns_cleared_buffer() {
        let rb = RenderBuffers::new();
        let buf = rb.take_conversation_buffer();
        assert!(buf.is_empty());
        assert!(buf.capacity() >= INITIAL_VIEW_CAPACITY);
    }

    #[test]
    fn render_buffers_return_preserves_capacity() {
        let rb = RenderBuffers::new();
        let mut buf = rb.take_conversation_buffer();
        // Write enough data to grow the buffer well beyond initial capacity.
        let big = "x".repeat(INITIAL_VIEW_CAPACITY * 3);
        buf.push_str(&big);
        let grown_cap = buf.capacity();
        rb.return_conversation_buffer(buf);

        // Taking again should reuse the grown allocation (cleared but same cap).
        let buf2 = rb.take_conversation_buffer();
        assert!(buf2.is_empty());
        assert_eq!(buf2.capacity(), grown_cap);
    }

    #[test]
    fn render_buffers_take_without_return_gives_fresh() {
        let rb = RenderBuffers::new();
        let buf1 = rb.take_conversation_buffer();
        // Don't return buf1 — simulates the buffer being consumed.
        drop(buf1);
        // Next take gets a fresh (empty, zero-cap) String.
        let buf2 = rb.take_conversation_buffer();
        assert!(buf2.is_empty());
    }

    #[test]
    fn render_buffers_header_buf_cleared_on_each_call() {
        let rb = RenderBuffers::new();
        {
            let mut hdr = rb.header_buf();
            hdr.push_str("old header");
        }
        // Next call should return a cleared buffer with preserved capacity.
        let hdr = rb.header_buf();
        assert!(hdr.is_empty());
        assert!(hdr.capacity() >= INITIAL_CHROME_CAPACITY);
    }

    #[test]
    fn render_buffers_footer_buf_cleared_on_each_call() {
        let rb = RenderBuffers::new();
        {
            let mut ftr = rb.footer_buf();
            ftr.push_str("old footer");
        }
        let ftr = rb.footer_buf();
        assert!(ftr.is_empty());
        assert!(ftr.capacity() >= INITIAL_CHROME_CAPACITY);
    }
}