smith

//! Interactive observer: worker-owned bounded snapshots, no agent or provider (`SMH-SPEC-SPEC0001`, Observer mode).
use smith::config::Config;
use smith::error::{Result, SmithError};
use smith::id::{EntryId, SessionId};
use smith::tool::CancelHandle;
use smith_alloc::{Scope, enter};
use smith_core::content_codec::{BlockView, EntryContentView};
use smith_core::session::EntryFrame;
use smith_core::store::SessionReader;
use smith_ui::{Observer, ROW_BYTES, ROWS, Row};
use std::path::{Path, PathBuf};
use std::sync::mpsc::{self, SyncSender};
use std::time::Duration;

#[expect(
    clippy::needless_pass_by_value,
    reason = "convert owned errors before discarding them"
)]
fn fault(code: smith::error::SessionCode, message: impl ToString) -> SmithError {
    SmithError::Session {
        code,
        message: message.to_string(),
    }
}

enum Request {
    Fetch {
        generation: u64,
        start: usize,
        follow: bool,
        search: Option<String>,
    },
    Fork(EntryId),
    Export(PathBuf),
    Status,
}
impl Request {
    const fn snapshot(ui: &Observer, generation: u64) -> Self {
        Self::Fetch {
            generation,
            start: ui.start,
            follow: ui.follow,
            search: None,
        }
    }
}

enum Response {
    Snapshot { generation: u64, snapshot: Snapshot },
    Forked(SessionId),
    Exported(PathBuf),
    Status(bool),
}
struct Snapshot {
    total: usize,
    writer_active: bool,
    rows: Vec<Row>,
    found: Option<(usize, EntryId)>,
    visible_tip: Option<EntryId>,
}

/// Local identity of the snapshot requested by the frontend.
/// A response may update visible data only when its generation still matches.
#[derive(Default)]
struct SnapshotState {
    generation: u64,
    published_generation: Option<u64>,
    visible_tip: Option<EntryId>,
}

impl SnapshotState {
    const fn invalidate(&mut self) {
        self.generation = self.generation.wrapping_add(1);
        self.published_generation = None;
        self.visible_tip = None;
    }

    fn publish(&mut self, ui: &mut Observer, generation: u64, snapshot: Snapshot) -> bool {
        if generation != self.generation {
            return false;
        }
        self.published_generation = Some(generation);
        self.visible_tip = snapshot.visible_tip;
        ui.publish(snapshot.total, snapshot.writer_active, snapshot.rows);
        true
    }

    fn refresh_needed(&self) -> bool {
        self.published_generation != Some(self.generation)
    }

    const fn fork_target(&self) -> Option<EntryId> {
        self.visible_tip
    }
}

fn next_snapshot_request(
    ui: &Observer,
    snapshots: &SnapshotState,
    initialized: bool,
    queued: &mut Option<Request>,
) -> Option<Request> {
    queued.take().or_else(|| {
        if ui.follow || !initialized || snapshots.refresh_needed() {
            Some(Request::snapshot(ui, snapshots.generation))
        } else {
            None
        }
    })
}

fn clipped(text: &str, budget: usize) -> String {
    let mut line = String::with_capacity(budget.min(text.len()));
    for character in text.chars() {
        let safe = if character.is_control() {
            ' '
        } else {
            character
        };
        if line.len() + safe.len_utf8() > budget {
            if budget >= 3 {
                while line.len() + 3 > budget {
                    line.pop();
                }
                line.push('…');
            }
            break;
        }
        line.push(safe);
    }
    line
}

fn append_preview(line: &mut String, text: &str) {
    let remaining = ROW_BYTES.saturating_sub(line.len());
    if remaining != 0 {
        line.push_str(&clipped(text, remaining));
    }
}

fn row(entry: &EntryFrame) -> Result<Row> {
    let content = EntryContentView::decode(&entry.content)?;
    let mut text = match &content {
        EntryContentView::Message(message) => format!("{:?}: ", message.role),
        EntryContentView::ToolCall { .. } => "ToolCall: ".into(),
        EntryContentView::ToolResult { .. } => "ToolResult: ".into(),
        EntryContentView::Meta { .. } => "Meta: ".into(),
    };
    match &content {
        EntryContentView::Message(message) => {
            for (index, block) in message.blocks().enumerate() {
                if index != 0 {
                    append_preview(&mut text, " | ");
                }
                append_preview(
                    &mut text,
                    match block {
                        BlockView::Text(text) | BlockView::Thinking(text) => text,
                        BlockView::ToolResult { output, .. } => output,
                        BlockView::ToolUse { name, .. } => name,
                    },
                );
            }
        }
        EntryContentView::ToolCall { name, .. } => append_preview(&mut text, name),
        EntryContentView::ToolResult { output, .. } => append_preview(&mut text, output),
        EntryContentView::Meta { kind, .. } => append_preview(&mut text, kind),
    }
    Ok(Row {
        id: entry.id.to_string(),
        text,
    })
}

fn fetch(
    path: &Path,
    start: usize,
    follow: bool,
    search: Option<&str>,
    cancel: &CancelHandle,
) -> Result<Snapshot> {
    if cancel.is_cancelled() {
        return Err(SmithError::Cancelled);
    }
    let reader = SessionReader::open_cancelled(path, cancel)?;
    let max = Config::default_valid().frame_bytes_max;
    let tip = reader.selected_tip_cancelled(max, cancel)?;
    let mut rows = Vec::with_capacity(ROWS);
    let mut total = 0;
    let mut found = None;
    let mut visible_tip = None;
    if let Some(tip) = tip {
        reader.replay_at_cancelled(tip, max, cancel, |entry| {
            if let Some(term) = search
                && EntryContentView::decode(&entry.content)?.contains(term)
            {
                found = Some((total, entry.id));
            }
            if follow {
                if rows.len() == ROWS {
                    rows.remove(0);
                }
                rows.push(row(&entry)?);
                visible_tip = Some(entry.id);
            } else if total >= start && rows.len() < ROWS {
                rows.push(row(&entry)?);
                visible_tip = Some(entry.id);
            }
            total += 1;
            Ok(())
        })?;
    }
    Ok(Snapshot {
        total,
        writer_active: reader.writer_active(),
        rows,
        found,
        visible_tip,
    })
}

fn export(path: &Path, target: &Path, id: SessionId, cancel: &CancelHandle) -> Result<()> {
    if cancel.is_cancelled() {
        return Err(SmithError::Cancelled);
    }
    let reader = SessionReader::open_cancelled(path, cancel)?;
    let tip = reader.selected_tip_cancelled(Config::default_valid().frame_bytes_max, cancel)?;
    let mut out = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(target)
        .map_err(|error| fault(smith::error::SessionCode::ObserverExport, error))?;
    let result = tip
        .map(|tip| {
            reader.replay_at_cancelled(
                tip,
                Config::default_valid().frame_bytes_max,
                cancel,
                |entry| {
                    let view = EntryContentView::decode(&entry.content)?;
                    crate::inspection::output(&mut out, id, &entry, &view, true, true)
                        .map_err(|error| fault(smith::error::SessionCode::ObserverExport, error))
                },
            )
        })
        .transpose();
    if let Err(error) = result.and_then(|_| {
        out.sync_all()
            .map_err(|error| fault(smith::error::SessionCode::ObserverExport, error))
    }) {
        drop(out);
        // Preserve the export failure; unpublished output removal is best-effort cleanup.
        let _cleanup_result: std::io::Result<()> = std::fs::remove_file(target);
        return Err(error);
    }
    Ok(())
}

fn work(path: &Path, id: SessionId, request: Request, cancel: &CancelHandle) -> Result<Response> {
    match request {
        Request::Fetch {
            generation,
            start,
            follow,
            search,
        } => fetch(path, start, follow, search.as_deref(), cancel).map(|snapshot| {
            Response::Snapshot {
                generation,
                snapshot,
            }
        }),
        Request::Fork(at) => {
            if cancel.is_cancelled() {
                return Err(SmithError::Cancelled);
            }
            let target_id = SessionId::new();
            let target = path.with_file_name(format!("{target_id}.smh"));
            let opened = SessionReader::open_cancelled(path, cancel)?.fork_at_cancelled(
                at,
                &target,
                Config::default_valid(),
                cancel,
            )?;
            drop(opened);
            Ok(Response::Forked(target_id))
        }
        Request::Export(target) => {
            export(path, &target, id, cancel)?;
            Ok(Response::Exported(target))
        }
        Request::Status => smith_core::store::writer_active(path).map(Response::Status),
    }
}

fn serve(
    path: &Path,
    id: SessionId,
    receiver: &mpsc::Receiver<Request>,
    responses: &SyncSender<Result<Response>>,
    cancel: &CancelHandle,
) {
    while let Ok(request) = receiver.recv() {
        let _request = enter(Scope::IoRequest);
        if cancel.is_cancelled() || responses.send(work(path, id, request, cancel)).is_err() {
            break;
        }
    }
}

/// Attach by session identity, reading immutable snapshots on a dedicated worker.
///
/// `export_path` must be explicitly supplied for export; never overwrites a file.
/// Exiting or losing a writer never promotes this observer.
///
/// # Errors
/// Returns terminal or storage faults. The alternate screen is restored on errors.
pub fn run(id: SessionId, path: PathBuf, export_path: Option<&PathBuf>) -> Result<()> {
    let (requests, receiver) = mpsc::sync_channel::<Request>(1);
    let (responses, results) = mpsc::sync_channel::<Result<Response>>(1);
    let cancel = CancelHandle::new();
    let worker_cancel = cancel.clone();
    let thread = std::thread::spawn(move || {
        let _session = enter(Scope::Session);
        serve(&path, id, &receiver, &responses, &worker_cancel);
    });
    let result = interact(id, requests, &results, export_path, &cancel);
    cancel.cancel();
    let joined = thread.join().map_err(|_| {
        fault(
            smith::error::SessionCode::ObserverWorker,
            "storage worker panicked",
        )
    });
    match (result, joined) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
        (Err(error), Err(worker)) => Err(fault(
            smith::error::SessionCode::ObserverShutdown,
            format!("{error}; {worker}"),
        )),
    }
}

fn interact(
    id: SessionId,
    requests: SyncSender<Request>,
    results: &mpsc::Receiver<Result<Response>>,
    export_path: Option<&PathBuf>,
    cancel: &CancelHandle,
) -> Result<()> {
    let mut screen = smith_tui::Screen::open()
        .map_err(|e| fault(smith::error::SessionCode::ObserverTerminal, e))?;
    let result = event_loop(id, requests, results, export_path, cancel, &mut screen);
    let cleanup = screen
        .close()
        .map_err(|e| fault(smith::error::SessionCode::ObserverTerminal, e));
    match (result, cleanup) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
        (Err(error), Err(cleanup)) => Err(fault(
            smith::error::SessionCode::ObserverShutdown,
            format!("{error}; {cleanup}"),
        )),
    }
}

#[expect(
    clippy::too_many_lines,
    reason = "one terminal event loop owns all frontend transitions"
)]
fn event_loop(
    id: SessionId,
    requests: SyncSender<Request>,
    results: &mpsc::Receiver<Result<Response>>,
    export_path: Option<&PathBuf>,
    cancel: &CancelHandle,
    screen: &mut smith_tui::Screen,
) -> Result<()> {
    let session_label = id.to_string();
    let mut ui = Observer::default();
    let mut input = false;
    let mut pending = false;
    let mut initialized = false;
    let mut snapshots = SnapshotState::default();
    let mut search_request = None;
    let mut last_status = std::time::Instant::now();
    loop {
        if !pending {
            let request = next_snapshot_request(&ui, &snapshots, initialized, &mut search_request)
                .or_else(|| {
                    if last_status.elapsed() >= Duration::from_secs(1) {
                        last_status = std::time::Instant::now();
                        Some(Request::Status)
                    } else {
                        None
                    }
                });
            if let Some(request) = request {
                requests
                    .send(request)
                    .map_err(|e| fault(smith::error::SessionCode::ObserverWorker, e))?;
                pending = true;
            }
        }
        screen
            .draw(&session_label, &ui, input)
            .map_err(|e| fault(smith::error::SessionCode::ObserverTerminal, e))?;
        match results.try_recv() {
            Ok(Ok(Response::Snapshot {
                generation,
                snapshot,
            })) => {
                pending = false;
                if generation != snapshots.generation {
                    continue;
                }
                initialized = true;
                if let Some((index, found_id)) = snapshot.found {
                    ui.found = Some(found_id.to_string());
                    ui.start = index;
                    snapshots.invalidate();
                    search_request = Some(Request::snapshot(&ui, snapshots.generation));
                } else if !ui.search.is_empty() && ui.found.is_none() {
                    ui.notice = "No matching entry".into();
                } else {
                    snapshots.publish(&mut ui, generation, snapshot);
                }
            }
            Ok(Ok(Response::Forked(new_id))) => {
                pending = false;
                ui.notice = format!(
                    "Fork created: {new_id}. Resume with smith eval --session {new_id} <prompt>"
                );
            }
            Ok(Ok(Response::Exported(target))) => {
                pending = false;
                ui.notice = clipped(&format!("Exported to {}", target.display()), ROW_BYTES);
            }
            Ok(Ok(Response::Status(active))) => {
                pending = false;
                ui.status_known = true;
                ui.writer_active = active;
            }
            Ok(Err(SmithError::Cancelled)) => {
                pending = false;
            }
            Ok(Err(e)) => {
                pending = false;
                ui.notice = clipped(&format!("Error: {e}"), ROW_BYTES);
            }
            Err(mpsc::TryRecvError::Disconnected) => {
                return Err(fault(
                    smith::error::SessionCode::ObserverWorker,
                    "storage worker stopped",
                ));
            }
            Err(mpsc::TryRecvError::Empty) => {}
        }
        match smith_tui::next_input(Duration::from_millis(150))
            .map_err(|e| fault(smith::error::SessionCode::ObserverTerminal, e))?
        {
            Some(smith_tui::Input::Key(key)) => {
                use smith_tui::Key;
                if input {
                    match key {
                        Key::Escape | Key::Interrupt => {
                            input = false;
                            ui.search.clear();
                        }
                        Key::Enter => {
                            input = false;
                            ui.follow = false;
                            ui.found = None;
                            snapshots.invalidate();
                            search_request = Some(Request::Fetch {
                                generation: snapshots.generation,
                                start: ui.start,
                                follow: false,
                                search: Some(ui.search.clone()),
                            });
                        }
                        Key::Backspace => {
                            ui.search.pop();
                        }
                        Key::Char(c) if ui.search.len() < 128 => ui.search.push(c),
                        _ => {}
                    }
                } else {
                    match key {
                        Key::Char('q') | Key::Escape | Key::Interrupt => {
                            cancel.cancel();
                            break;
                        }
                        Key::PageUp | Key::Char('b') => {
                            ui.back();
                            snapshots.invalidate();
                            search_request = Some(Request::snapshot(&ui, snapshots.generation));
                        }
                        Key::PageDown | Key::Char('n') => {
                            ui.forward();
                            snapshots.invalidate();
                            search_request = Some(Request::snapshot(&ui, snapshots.generation));
                        }
                        Key::Char('f') => {
                            ui.toggle_follow();
                            snapshots.invalidate();
                            search_request = Some(Request::snapshot(&ui, snapshots.generation));
                        }
                        Key::Char('r') => {
                            snapshots.invalidate();
                            search_request = Some(Request::snapshot(&ui, snapshots.generation));
                        }
                        Key::Char('/') => {
                            ui.search.clear();
                            input = true;
                        }
                        Key::Char('k') => {
                            if let Some(at) = snapshots.fork_target() {
                                search_request = Some(Request::Fork(at));
                            } else {
                                ui.notice = "No visible entry to fork; wait for snapshot".into();
                            }
                        }
                        Key::Char('e') => {
                            if let Some(target) = &export_path {
                                search_request = Some(Request::Export((*target).clone()));
                            } else {
                                ui.notice = "Export requires --export PATH".into();
                            }
                        }
                        Key::Enter | Key::Char(_) => {
                            ui.notice = "READ ONLY: prompt submission disabled".into();
                        }
                        Key::Backspace => {}
                    }
                }
            }
            Some(smith_tui::Input::Resize) | None => {}
        }
    }
    drop(requests);
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use smith_alloc::Scope;
    use std::io::Write;

    const SMALL_ENTRIES: usize = 4;
    const LARGE_ENTRIES: usize = 128;
    const PAYLOAD_BYTES: usize = 32 * 1024;

    fn observed_history(
        path: &Path,
        entries: usize,
        tag: &str,
    ) -> (SessionId, smith_core::store::SessionWriter) {
        use smith::message::{Message, Role};
        use smith_core::frame::Frame;
        use smith_core::session::{EntryContent, Session};
        use smith_core::store::SessionWriter;

        let session = Session::new();
        let mut writer = SessionWriter::open(path, Config::default_valid()).unwrap();
        writer
            .append_frame(&Frame::Header {
                header: session.header(),
            })
            .unwrap();
        let mut tip = None;
        for index in 0..entries {
            let content = if index % 2 == 0 {
                EntryContent::Message(Message::with_text(
                    Role::Assistant,
                    format!("visible needle {tag} {index} {}", "x".repeat(PAYLOAD_BYTES)),
                ))
            } else {
                EntryContent::Meta {
                    kind: "event".into(),
                    detail: smith::StructuredData::from_serializable(&serde_json::json!({
                        "needle": tag,
                        "payload": "x".repeat(PAYLOAD_BYTES),
                        "nested": [{"values": [true, 0, null]}],
                    }))
                    .unwrap(),
                }
            };
            let entry = EntryFrame::new(tip, &content).unwrap();
            tip = Some(entry.id);
            writer.append_frame(&Frame::Known { entry }).unwrap();
        }
        writer.select_entry(tip.unwrap()).unwrap();
        (session.id(), writer)
    }

    fn worker_request(
        path: PathBuf,
        id: SessionId,
        request: Request,
    ) -> (Response, smith_alloc::Stats) {
        let (requests, receiver) = mpsc::sync_channel(1);
        let (responses, results) = mpsc::sync_channel(1);
        let before = smith_alloc::snapshot(Scope::IoRequest);
        let thread = std::thread::spawn(move || {
            let cancel = CancelHandle::new();
            serve(&path, id, &receiver, &responses, &cancel);
        });
        requests.send(request).unwrap();
        drop(requests);
        let response = results.recv_timeout(Duration::from_secs(5)).unwrap();
        thread.join().unwrap();
        let after = smith_alloc::snapshot(Scope::IoRequest);
        (
            response.unwrap(),
            smith_alloc::Stats {
                allocations: after.allocations - before.allocations,
                bytes: after.bytes - before.bytes,
                live: after.live,
                peak: after.peak,
            },
        )
    }

    fn observer_cost(
        path: &Path,
        id: SessionId,
        entries: usize,
        target: &Path,
    ) -> (smith_alloc::Stats, EntryId) {
        let (response, fetch_cost) = worker_request(
            path.to_path_buf(),
            id,
            Request::Fetch {
                generation: 0,
                start: entries.saturating_sub(ROWS),
                follow: false,
                search: Some("needle".into()),
            },
        );
        let Response::Snapshot { snapshot, .. } = response else {
            panic!("expected observer snapshot")
        };
        assert_eq!(snapshot.total, entries);
        assert_eq!(snapshot.rows.len(), entries.min(ROWS));
        assert!(snapshot.found.is_some());
        let fork_at = snapshot.visible_tip.unwrap();
        let retained = fetch_cost.live;
        drop(snapshot);
        let released = smith_alloc::snapshot(Scope::IoRequest).live;
        assert!(
            fetch_cost.allocations > 0 && retained != released,
            "fetch response retention must be visible in IoRequest accounting"
        );
        let (response, export_cost) = worker_request(
            path.to_path_buf(),
            id,
            Request::Export(target.to_path_buf()),
        );
        assert!(matches!(response, Response::Exported(_)));
        assert!(
            fetch_cost.allocations > 0 && export_cost.allocations > 0,
            "production observer fetch and export requests must charge allocations to IoRequest"
        );
        (
            smith_alloc::Stats {
                allocations: fetch_cost.allocations + export_cost.allocations,
                bytes: fetch_cost.bytes + export_cost.bytes,
                live: fetch_cost.live.max(export_cost.live),
                peak: fetch_cost.peak.max(export_cost.peak),
            },
            fork_at,
        )
    }

    fn snapshot(id: EntryId, row: &str) -> Snapshot {
        Snapshot {
            total: 24,
            writer_active: true,
            rows: vec![Row {
                id: id.to_string(),
                text: row.into(),
            }],
            found: None,
            visible_tip: Some(id),
        }
    }

    #[cfg(target_arch = "x86_64")]
    #[test]
    fn snapshot_state_stays_within_one_cache_line() {
        assert_eq!(std::mem::size_of::<SnapshotState>(), 48);
    }

    #[test]
    fn stale_snapshot_after_navigation_cannot_replace_rows_or_fork_target() {
        let initial = EntryId::new();
        let stale = EntryId::new();
        let current = EntryId::new();
        let mut ui = Observer::default();
        let mut state = SnapshotState::default();
        assert!(state.publish(&mut ui, 0, snapshot(initial, "initial")));
        ui.forward();
        state.invalidate();
        let old_generation = state.generation;
        ui.back();
        state.invalidate();
        assert!(!state.publish(&mut ui, old_generation, snapshot(stale, "stale")));
        assert_eq!(ui.start, 0);
        assert_eq!(ui.rows[0].text, "initial");
        assert_eq!(state.fork_target(), None);
        assert!(state.publish(&mut ui, state.generation, snapshot(current, "current")));
        assert_eq!(ui.rows[0].text, "current");
        assert_eq!(state.fork_target(), Some(current));
    }

    #[test]
    fn export_after_navigation_requeues_current_snapshot_before_fork() {
        let old = EntryId::new();
        let current = EntryId::new();
        let mut ui = Observer::default();
        let mut state = SnapshotState::default();
        let mut queued = Some(Request::snapshot(&ui, state.generation));
        let stale_generation = match queued.take().unwrap() {
            Request::Fetch { generation, .. } => generation,
            _ => unreachable!(),
        };
        ui.back();
        state.invalidate();
        queued = Some(Request::snapshot(&ui, state.generation));
        assert!(matches!(queued, Some(Request::Fetch { .. })));
        queued = Some(Request::Export(PathBuf::from("transcript.json")));
        assert!(matches!(queued, Some(Request::Export(_))));
        queued = Some(Request::Export(PathBuf::from("latest-transcript.json")));
        assert!(matches!(queued.take(), Some(Request::Export(_))));
        assert!(!state.publish(&mut ui, stale_generation, snapshot(old, "stale")));
        assert!(state.refresh_needed());
        let Request::Fetch { generation, .. } =
            next_snapshot_request(&ui, &state, true, &mut queued)
                .expect("export completion must requeue the invalidated snapshot")
        else {
            unreachable!()
        };
        assert!(state.publish(&mut ui, generation, snapshot(current, "current")));
        assert_eq!(state.fork_target(), Some(current));
    }

    #[test]
    fn published_empty_snapshot_does_not_request_again_while_browsing() {
        let mut ui = Observer::default();
        let mut state = SnapshotState::default();
        assert!(state.publish(
            &mut ui,
            state.generation,
            Snapshot {
                total: 0,
                writer_active: false,
                rows: Vec::new(),
                found: None,
                visible_tip: None,
            },
        ));
        assert!(!state.refresh_needed());
        let mut queued = None;
        assert!(next_snapshot_request(&ui, &state, true, &mut queued).is_none());
    }

    #[test]
    fn multi_block_message_preview_retains_order_within_row_budget() {
        use smith::message::{ContentBlock, Message, Role};
        use smith_core::session::EntryContent;
        let mut message = Message::with_text(Role::Assistant, "first");
        message.add_block(ContentBlock::thinking("second"));
        message.add_block(ContentBlock::text("third"));
        let entry = EntryFrame::new(None, &EntryContent::Message(message)).unwrap();
        let row = row(&entry).unwrap();
        assert!(row.text.contains("first | second | third"));
        assert!(row.text.len() <= ROW_BYTES);
    }

    #[test]
    fn observer_snapshots_follow_append_and_fork_without_mutating_source() {
        use smith::message::{Message, Role};
        use smith_core::session::{EntryContent, Session};
        use smith_core::store::SessionWriter;
        let dir =
            std::env::temp_dir().join(format!("smith-observer-worker-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let session = Session::new();
        let path = dir.join(format!("{}.smh", session.id()));
        let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
        writer
            .append_frame(&smith_core::frame::Frame::Header {
                header: session.header(),
            })
            .unwrap();
        let first = EntryFrame::new(
            None,
            &EntryContent::Message(Message::with_text(Role::User, "first")),
        )
        .unwrap();
        writer
            .append_entry(
                &smith_core::session::Entry {
                    id: first.id,
                    parent: first.parent,
                    timestamp_ms: first.timestamp_ms,
                    kind: first.kind,
                    content: smith_core::session::Span { offset: 0, len: 0 },
                },
                &first.content,
            )
            .unwrap();
        let cancel = CancelHandle::new();
        let initial = fetch(&path, 0, false, None, &cancel).unwrap();
        assert!(initial.writer_active);
        assert_eq!(initial.rows.len(), 1);
        let next = EntryFrame::new(
            Some(first.id),
            &EntryContent::Message(Message::with_text(Role::Assistant, "second")),
        )
        .unwrap();
        writer
            .append_entry(
                &smith_core::session::Entry {
                    id: next.id,
                    parent: next.parent,
                    timestamp_ms: next.timestamp_ms,
                    kind: next.kind,
                    content: smith_core::session::Span { offset: 0, len: 0 },
                },
                &next.content,
            )
            .unwrap();
        let live = fetch(&path, 0, true, Some("second"), &cancel).unwrap();
        assert_eq!(
            fetch(&path, 0, false, None, &cancel).unwrap().visible_tip,
            Some(next.id)
        );
        assert_eq!(live.total, 2);
        assert_eq!(live.found, Some((1, next.id)));
        assert!(live.rows.last().unwrap().text.contains("second"));
        let before = std::fs::read(&path).unwrap();
        let Response::Forked(new_id) =
            work(&path, session.id(), Request::Fork(first.id), &cancel).unwrap()
        else {
            panic!("expected fork")
        };
        let fork_path = dir.join(format!("{new_id}.smh"));
        assert_eq!(fetch(&fork_path, 0, true, None, &cancel).unwrap().total, 1);
        assert_eq!(std::fs::read(&path).unwrap(), before);
        drop(writer);
        assert!(!fetch(&path, 0, false, None, &cancel).unwrap().writer_active);
        let Response::Status(active) = work(&path, session.id(), Request::Status, &cancel).unwrap()
        else {
            panic!("expected status")
        };
        assert!(!active);
        assert_eq!(fetch(&fork_path, 0, true, None, &cancel).unwrap().total, 1);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn observer_visible_window_and_export_peak_ignore_history_payload() {
        // SMH-SPEC-SPEC0001: worker consumer paths retain ROWS previews and
        // stream JSONL export; incomplete live tails are excluded.
        let root =
            std::env::temp_dir().join(format!("smith-observer-bounds-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).unwrap();
        let small_path = root.join("small.smh");
        let large_path = root.join("large.smh");
        let (small_id, small_writer) = observed_history(&small_path, SMALL_ENTRIES, "small");
        let (large_id, large_writer) = observed_history(&large_path, LARGE_ENTRIES, "large");
        let mut tail = std::fs::OpenOptions::new()
            .append(true)
            .open(&large_path)
            .unwrap();
        tail.write_all(&[4, 0]).unwrap();
        drop(tail);
        let source_before = std::fs::read(&large_path).unwrap();

        let (small, _) = observer_cost(
            &small_path,
            small_id,
            SMALL_ENTRIES,
            &root.join("small.jsonl"),
        );
        let (large, fork_at) = observer_cost(
            &large_path,
            large_id,
            LARGE_ENTRIES,
            &root.join("large.jsonl"),
        );
        assert!(
            large.peak <= small.peak + 64 * 1024,
            "worker fetch/export peak grew from {} to {} bytes",
            small.peak,
            large.peak
        );
        assert!(
            large.peak < 256 * 1024,
            "worker request peak {} bytes for {} bytes of history",
            large.peak,
            std::fs::metadata(&large_path).unwrap().len()
        );
        assert_eq!(std::fs::read(&large_path).unwrap(), source_before);
        assert_eq!(
            std::fs::read_to_string(root.join("large.jsonl"))
                .unwrap()
                .lines()
                .count(),
            LARGE_ENTRIES,
            "incomplete live tail is not exported"
        );
        let before_fork = smith_alloc::snapshot(Scope::IoRequest);
        let (response, fork_cost) = worker_request(large_path, large_id, Request::Fork(fork_at));
        assert!(matches!(response, Response::Forked(_)));
        assert!(
            fork_cost.allocations > 0
                && smith_alloc::snapshot(Scope::IoRequest).allocations > before_fork.allocations,
            "production observer fork requests must charge allocations to IoRequest"
        );
        drop((small_writer, large_writer));
        let _ = std::fs::remove_dir_all(root);
    }

    #[test]
    fn clipping_is_utf8_safe_and_bounded() {
        assert_eq!(clipped("aébc", 4), "a…");
        assert_eq!(clipped("a\nb", 3), "a b");
        assert_eq!(clipped("\u{1b}[31m", 5), " [31m");
        assert!(clipped(&"é".repeat(600), ROW_BYTES).len() <= ROW_BYTES);
    }
}