//! Durable session storage: append-only writer, incremental bounded loader,
//! and atomic repair publication (`SMH-SPEC-SPEC0001`, Sessions).

use crate::frame::{
    FILE_PROLOGUE, Frame, Recovery, encode_entry, index_bytes, visit_frames,
    visit_frames_reporting, visit_metadata_reporting, write_frame,
};
use crate::session::{Entry, EntryFrame, Session};
use smith::config::Config;
use smith::error::{Result, SmithError};
use smith::id::{EntryId, SessionId};
use smith::tool::CancelHandle;
use std::fs::{File, OpenOptions};
use std::io::{BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

/// Exclusive access to one session, held on a sidecar lock file.
///
/// The lock lives beside the session file rather than on it, because repair
/// publishes a replacement inode: a lock taken on the data file would leave
/// a waiter holding a descriptor on the replaced inode, and its appends
/// would vanish. The sidecar is never replaced, so every participant
/// serializes on the same object and opens the data file only after winning.
#[derive(Debug)]
struct SessionLock {
    file: File,
    path: PathBuf,
}

impl SessionLock {
    /// Acquire the lock for `path`, creating parent directories and the
    /// sidecar as needed. Blocks while another holder owns the session.
    fn acquire(path: &Path) -> Result<Self> {
        Self::lock(path, false)
    }

    fn try_acquire(path: &Path) -> Result<Self> {
        Self::lock(path, true)
    }

    fn lock(path: &Path, nonblocking: bool) -> Result<Self> {
        if let Some(parent) = path.parent()
            && !parent.as_os_str().is_empty()
        {
            std::fs::create_dir_all(parent)
                .map_err(io_error(smith::error::StorageCode::StoreMkdir, parent))?;
        }
        let path = canonical_session_path(path)?;
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(lock_path(&path))
            .map_err(io_error(
                smith::error::StorageCode::StoreLockOpen,
                &lock_path(&path),
            ))?;
        if nonblocking {
            fs2::FileExt::try_lock_exclusive(&file).map_err(io_error(
                smith::error::StorageCode::StoreBusy,
                &lock_path(&path),
            ))?;
        } else {
            fs2::FileExt::lock_exclusive(&file).map_err(io_error(
                smith::error::StorageCode::StoreLock,
                &lock_path(&path),
            ))?;
        }
        Ok(Self { file, path })
    }

    fn path(&self) -> &Path {
        &self.path
    }
}

impl Drop for SessionLock {
    fn drop(&mut self) {
        // Best-effort unlock; the OS also releases on close.
        let _unlock_result: std::io::Result<()> = fs2::FileExt::unlock(&self.file);
    }
}

fn scratch_file(operation: &str) -> Result<(PathBuf, File)> {
    let (path, file) = new_temp_file(
        &std::env::temp_dir().join("smith-scratch"),
        operation,
        SessionId::new(),
    )?;
    std::fs::remove_file(&path).map_err(io_error(
        smith::error::StorageCode::StoreScratchRemove,
        &path,
    ))?;
    Ok((path, file))
}

/// Sidecar lock file path for a session file.
fn readers_path(path: &Path) -> PathBuf {
    let mut name = path.as_os_str().to_os_string();
    name.push(".readers.lock");
    PathBuf::from(name)
}

fn lock_path(path: &Path) -> PathBuf {
    let mut name = path.as_os_str().to_os_string();
    name.push(".lock");
    PathBuf::from(name)
}

/// Resolve existing data symlinks, or a new file's parent symlinks, before
/// deriving sidecar paths.
///
/// Unix rejects data files with multiple hard links.
/// Bind mounts and non-Unix hard-link aliases have no portable safe identity
/// query, so callers must not expose one session through them.
fn canonical_session_path(path: &Path) -> Result<PathBuf> {
    match std::fs::canonicalize(path) {
        Ok(path) => {
            reject_ambiguous_data_alias(&path)?;
            Ok(path)
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            if std::fs::symlink_metadata(path).is_ok() {
                return Err(io_error(smith::error::StorageCode::StoreCanonicalize, path)(error));
            }
            let parent = path
                .parent()
                .filter(|parent| !parent.as_os_str().is_empty())
                .unwrap_or_else(|| Path::new("."));
            let parent = std::fs::canonicalize(parent)
                .map_err(io_error(smith::error::StorageCode::StoreCanonicalize, path))?;
            let name = path.file_name().ok_or_else(|| {
                storage_fault(
                    smith::error::StorageCode::StoreCanonicalize,
                    path,
                    format!("{} has no file name", path.display()),
                )
            })?;
            Ok(parent.join(name))
        }
        Err(error) => Err(io_error(smith::error::StorageCode::StoreCanonicalize, path)(error)),
    }
}

#[cfg(unix)]
fn reject_ambiguous_data_alias(path: &Path) -> Result<()> {
    use std::os::unix::fs::MetadataExt;

    let metadata =
        std::fs::metadata(path).map_err(io_error(smith::error::StorageCode::StoreStat, path))?;
    if metadata.is_file() && metadata.nlink() > 1 {
        return Err(SmithError::Session {
            code: smith::error::SessionCode::SessionAmbiguousAlias,
            message: format!("{} has multiple hard links", path.display()),
        });
    }
    Ok(())
}

#[cfg(not(unix))]
fn reject_ambiguous_data_alias(_: &Path) -> Result<()> {
    Ok(())
}

fn validate_named_identity(path: &Path, session_id: SessionId) -> Result<()> {
    if let Some(named_id) = path
        .file_stem()
        .and_then(|stem| stem.to_str())
        .and_then(|stem| stem.parse::<SessionId>().ok())
        && named_id != session_id
    {
        return Err(SmithError::Session {
            code: smith::error::SessionCode::SessionIdMismatch,
            message: format!(
                "{} names session {named_id}, not {session_id}",
                path.display()
            ),
        });
    }
    Ok(())
}

fn reject_existing_target(path: &Path) -> Result<()> {
    match std::fs::symlink_metadata(path) {
        Ok(_) => Err(SmithError::Session {
            code: smith::error::SessionCode::SessionExists,
            message: format!("{} already exists", path.display()),
        }),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(io_error(smith::error::StorageCode::CreateStat, path)(error)),
    }
}

#[cfg(test)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum PublicationCheckpoint {
    TempCreated,
    PartialWrite,
    CompletedWrite,
    TempSynced,
    Published,
    ParentSynced,
    Cleanup,
}

#[cfg(test)]
type PublicationHook = Box<dyn FnMut(PublicationCheckpoint)>;

#[cfg(test)]
thread_local! {
    static PUBLICATION_HOOK: std::cell::RefCell<Option<PublicationHook>> = const { std::cell::RefCell::new(None) };
    static PARENT_SYNC_FAILURE: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

#[cfg(test)]
fn run_publication_hook(point: PublicationCheckpoint) {
    PUBLICATION_HOOK.with(|hook| {
        if let Some(hook) = hook.borrow_mut().as_mut() {
            hook(point);
        }
    });
}

macro_rules! publication_checkpoint {
    ($point:ident) => {{
        #[cfg(test)]
        run_publication_hook(PublicationCheckpoint::$point);
    }};
}

#[cfg(test)]
fn set_publication_hook(hook: impl FnMut(PublicationCheckpoint) + 'static) {
    PUBLICATION_HOOK.with(|slot| *slot.borrow_mut() = Some(Box::new(hook)));
}

#[cfg(test)]
fn clear_publication_hook() {
    PUBLICATION_HOOK.with(|slot| *slot.borrow_mut() = None);
}

#[cfg(test)]
fn set_parent_sync_failure(fail: bool) {
    PARENT_SYNC_FAILURE.with(|slot| slot.set(fail));
}

fn new_temp_file(path: &Path, operation: &str, id: SessionId) -> Result<(PathBuf, File)> {
    let name = path.file_name().ok_or_else(|| {
        storage_fault(
            smith::error::StorageCode::CreateTmp,
            path,
            format!("{} has no file name", path.display()),
        )
    })?;
    let mut last = path.to_path_buf();
    for attempt in 0..128 {
        let mut temporary = name.to_os_string();
        temporary.push(format!(".smith-{operation}-{id}-{attempt}.tmp"));
        let temporary = path.with_file_name(temporary);
        last.clone_from(&temporary);
        let mut options = OpenOptions::new();
        options.read(true).write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        match options.open(&temporary) {
            Ok(file) => return Ok((temporary, file)),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
            Err(error) => {
                return Err(io_error(smith::error::StorageCode::CreateTmp, &temporary)(
                    error,
                ));
            }
        }
    }
    Err(storage_fault(
        smith::error::StorageCode::CreateTmp,
        &last,
        format!(
            "could not allocate a unique temporary beside {}",
            path.display()
        ),
    ))
}

/// Append-only session writer holding exclusive access to its session.
///
/// Opening repairs a torn tail (truncating to the last complete frame
/// boundary) before appending, so acknowledged appends are always
/// reachable by the loader. The lock serializes writers against each other
/// and against repair.
#[derive(Debug)]
pub struct SessionWriter {
    file: File,
    config: Config,
    path: PathBuf,
    /// A failed write or sync may have appended an unreachable frame prefix.
    poisoned: bool,
    /// Last known entry boundary, used only for reconstructible hints.
    last: Option<(EntryId, u64)>,
    /// Hints since the last reconstructible selected-tip checkpoint.
    since_checkpoint: u16,
    /// Held for the writer's lifetime; released on drop.
    _lock: SessionLock,
}

impl SessionWriter {
    /// Open (creating parent directories) and lock the session file.
    ///
    /// # Errors
    ///
    /// Returns a configuration, locking, or input/output error.
    pub fn open<P: AsRef<Path>>(path: P, config: Config) -> Result<Self> {
        Self::open_with_recovery(path, config).map(|(writer, _)| writer)
    }

    /// Open and lock the session file, returning bounded damage/tail metadata.
    /// No frame payloads are retained by the durable open path.
    fn open_with_recovery<P: AsRef<Path>>(path: P, config: Config) -> Result<(Self, Recovery)> {
        // Win the lock before opening the data file, so a repair that is
        // publishing a replacement cannot hand out a stale descriptor.
        let lock = SessionLock::try_acquire(path.as_ref())?;
        Self::attach(lock.path().to_path_buf(), config, lock)
    }

    /// Take over an already-locked session file.
    fn attach(path: PathBuf, config: Config, lock: SessionLock) -> Result<(Self, Recovery)> {
        // Validation already names the offending field path.
        config.validate()?;
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
            .map_err(io_error(smith::error::StorageCode::StoreOpen, &path))?;
        // Drop a torn tail before appending so new frames stay reachable.
        // Complete but damaged frames are left in place and stay reported.
        if file
            .metadata()
            .map_err(io_error(smith::error::StorageCode::StoreStat, &path))?
            .len()
            == 0
        {
            file.write_all(&FILE_PROLOGUE)
                .map_err(io_error(smith::error::StorageCode::StoreWrite, &path))?;
            file.sync_data()
                .map_err(io_error(smith::error::StorageCode::StoreSync, &path))?;
        }
        let recovery = walk_file_bounded(&file, &path, config.frame_bytes_max)?;
        let (first, _) = SessionReader::open(&path)?.first(config.frame_bytes_max)?;
        if let Some(named_id) = path
            .file_stem()
            .and_then(|stem| stem.to_str())
            .and_then(|stem| stem.parse::<smith::id::SessionId>().ok())
            && let Some(Frame::Header { header }) = first
            && header.session_id != named_id
        {
            return Err(SmithError::Session {
                code: smith::error::SessionCode::SessionIdMismatch,
                message: format!(
                    "{} records session {}, not {named_id}",
                    path.display(),
                    header.session_id
                ),
            });
        }
        let len = file
            .metadata()
            .map_err(io_error(smith::error::StorageCode::StoreStat, &path))?
            .len();
        if recovery.good_end < len {
            let coordination = OpenOptions::new()
                .read(true)
                .write(true)
                .create(true)
                .truncate(false)
                .open(readers_path(&path))
                .map_err(io_error(
                    smith::error::StorageCode::StoreReaderLockOpen,
                    &readers_path(&path),
                ))?;
            fs2::FileExt::try_lock_exclusive(&coordination).map_err(io_error(
                smith::error::StorageCode::StoreReadersActive,
                &readers_path(&path),
            ))?;
            file.set_len(recovery.good_end)
                .map_err(io_error(smith::error::StorageCode::StoreTruncate, &path))?;
            file.sync_data()
                .map_err(io_error(smith::error::StorageCode::StoreSync, &path))?;
            fs2::FileExt::unlock(&coordination).map_err(io_error(
                smith::error::StorageCode::StoreReaderUnlock,
                &readers_path(&path),
            ))?;
        }
        let since_checkpoint = recovery_checkpoint_count(&file, &path, config.frame_bytes_max)?;
        file.seek(SeekFrom::Start(recovery.good_end))
            .map_err(io_error(smith::error::StorageCode::StoreSeek, &path))?;
        Ok((
            Self {
                file,
                config,
                path,
                poisoned: false,
                last: None,
                since_checkpoint,
                _lock: lock,
            },
            recovery,
        ))
    }

    /// Append one frame. The encoded size is checked against the configured
    /// limit before any bytes are written; a rejected frame leaves the file
    /// unchanged.
    ///
    /// # Errors
    ///
    /// Returns a framing error for oversized frames, or an input/output error.
    /// A write or sync error poisons this writer; later appends fail.
    pub fn append_frame(&mut self, frame: &Frame) -> Result<()> {
        let bytes = frame.encode(self.config.frame_bytes_max)?;
        self.write(&bytes)
    }

    /// Append one session record as an entry frame, framing its arena
    /// `content` bytes directly. Size limits apply as in
    /// [`SessionWriter::append_frame`].
    ///
    /// # Errors
    ///
    /// Returns a framing error for oversized frames, or an input/output error.
    /// A write or sync error poisons this writer; later appends fail.
    pub fn append_entry(&mut self, entry: &Entry, content: &[u8]) -> Result<()> {
        let bytes = encode_entry(entry, content, self.config.frame_bytes_max)?;
        self.write_entry_bytes(&bytes, entry.id, entry.parent)
    }

    #[expect(
        clippy::large_types_passed_by_value,
        reason = "Copy parent ID is consumed once to resolve an index hint"
    )]
    fn write_entry_bytes(
        &mut self,
        bytes: &[u8],
        id: EntryId,
        parent: Option<EntryId>,
    ) -> Result<()> {
        let parent_offset = parent.and_then(|id| {
            self.last
                .filter(|(last, _)| *last == id)
                .map(|(_, offset)| offset)
        });
        let offset = self
            .file
            .stream_position()
            .map_err(io_error(smith::error::StorageCode::StoreSeek, &self.path))?;
        self.write(bytes)?;
        self.last = Some((id, offset));
        if (parent.is_none() || parent_offset.is_some()) && self.config.frame_bytes_max >= 48 {
            self.write(&index_bytes(id, offset, parent, parent_offset))?;
        }
        self.since_checkpoint += 1;
        if self.since_checkpoint == 64 && self.config.frame_bytes_max >= 24 {
            self.append_frame(&Frame::Checkpoint { id, offset })?;
            self.since_checkpoint = 0;
        }
        Ok(())
    }

    /// Persist a branch-tip selection, including when no new entry was appended.
    ///
    /// # Errors
    /// Returns `ENTRY_NOT_FOUND` for an absent entry or a storage fault.
    pub fn select_entry(&mut self, id: EntryId) -> Result<()> {
        let (_, offset) = SessionReader::open(&self.path)?
            .find_entry(id, self.config.frame_bytes_max)?
            .ok_or_else(|| missing_entry(id))?;
        self.append_frame(&Frame::Selection { id })?;
        if self.config.frame_bytes_max >= 24 {
            self.append_frame(&Frame::Checkpoint { id, offset })?;
            self.since_checkpoint = 0;
        }
        self.last = Some((id, offset));
        Ok(())
    }

    fn write(&mut self, bytes: &[u8]) -> Result<()> {
        self.io_step(smith::error::StorageCode::StoreWrite, |file| {
            file.write_all(bytes)
        })?;
        self.io_step(smith::error::StorageCode::StoreSync, |file| {
            file.sync_data()
        })
    }

    fn io_step(
        &mut self,
        code: smith::error::StorageCode,
        step: impl FnOnce(&mut File) -> std::io::Result<()>,
    ) -> Result<()> {
        if self.poisoned {
            return Err(storage_fault(
                smith::error::StorageCode::StorePoisoned,
                &self.path,
                format!("{} may contain a partial append", self.path.display()),
            ));
        }
        match step(&mut self.file) {
            Ok(()) => Ok(()),
            Err(error) => {
                self.poisoned = true;
                Err(io_error(code, &self.path)(error))
            }
        }
    }

    #[cfg(test)]
    fn write_step(&mut self, step: impl FnOnce(&mut File) -> Result<()>) -> Result<()> {
        if self.poisoned {
            return Err(storage_fault(
                smith::error::StorageCode::StorePoisoned,
                &self.path,
                format!("{} may contain a partial append", self.path.display()),
            ));
        }
        match step(&mut self.file) {
            Ok(()) => Ok(()),
            Err(error) => {
                self.poisoned = true;
                Err(error)
            }
        }
    }

    /// Path of the underlying session file.
    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }
}

/// A session file opened for use.
#[derive(Debug)]
pub struct OpenSession {
    /// Restored, or newly created, session state.
    pub session: Session,
    /// Locked writer positioned after the last complete frame.
    pub writer: SessionWriter,
    /// Damage and torn-tail metadata found while loading; `frames` is empty
    /// on bounded durable open, not a second copy of session history.
    pub recovery: Recovery,
}

/// Open a session file, writing a header frame when the file is new.
///
/// Reopening restores the persisted session and branch IDs, so state
/// reconstructed from the same file is identical on every load. A file that
/// holds frames but no header cannot be attributed to a session and fails
/// explicitly instead of silently becoming a new session.
///
/// # Errors
///
/// Returns [`SmithError::Session`] when the file holds frames but no header,
/// or a locking, framing, or input/output error.
pub fn open_session(path: &Path, config: Config) -> Result<OpenSession> {
    let (mut writer, recovery) = SessionWriter::open_with_recovery(path, config)?;
    let reader = SessionReader::open(&writer.path)?;
    let session = if recovery.good_end > FILE_PROLOGUE.len() as u64 {
        reader.selected_session(writer.config.frame_bytes_max)?
    } else if recovery.damage.is_empty() {
        let session = path
            .file_stem()
            .and_then(|stem| stem.to_str())
            .and_then(|stem| stem.parse().ok())
            .map_or_else(Session::new, Session::new_with_id);
        writer.append_frame(&Frame::Header {
            header: session.header(),
        })?;
        session
    } else {
        return Err(SmithError::Session {
            code: smith::error::SessionCode::SessionHeaderMissing,
            message: format!("{} holds frames but no session header", path.display()),
        });
    };
    if let Some(id) = session.selected_id() {
        writer.last = reader
            .find_entry(id, writer.config.frame_bytes_max)?
            .map(|(_, offset)| (id, offset));
    }
    Ok(OpenSession {
        session,
        writer,
        recovery,
    })
}

/// Publish `session` as a new session file and open it for appending.
///
/// The file is written completely and then published by a no-replace hard
/// link, so an interrupted fork leaves no half-written session behind and an
/// existing path is never overwritten.
///
/// # Errors
///
/// Returns [`SmithError::Session`] when the target already holds a session,
/// or a locking, framing, or input/output error.
pub fn create_session(path: &Path, session: &Session, config: Config) -> Result<OpenSession> {
    config.validate()?;
    if session
        .active_branch()
        .entries()
        .first()
        .is_some_and(|root| root.parent.is_some())
    {
        return Err(session_fault(
            smith::error::SessionCode::SessionIncompleteContext,
            "effective context omits earlier ancestors; fork through SessionReader",
        ));
    }
    let lock = SessionLock::acquire(path)?;
    let path = lock.path().to_path_buf();
    validate_named_identity(&path, session.id())?;
    reject_existing_target(&path)?;
    let (temporary, mut out) = new_temp_file(&path, "create", session.id())?;
    publication_checkpoint!(TempCreated);
    let written = (|| {
        out.write_all(&FILE_PROLOGUE)
            .map_err(io_error(smith::error::StorageCode::CreateWrite, &temporary))?;
        publication_checkpoint!(PartialWrite);
        write_frame(
            &mut out,
            &Frame::Header {
                header: session.header(),
            },
            config.frame_bytes_max,
        )?;
        for entry in session.active_branch().entries() {
            let bytes = encode_entry(entry, session.bytes(entry), config.frame_bytes_max)?;
            out.write_all(&bytes)
                .map_err(io_error(smith::error::StorageCode::CreateWrite, &temporary))?;
        }
        if let Some(id) = session.selected_id()
            && session
                .active_branch()
                .entries()
                .last()
                .map(|entry| entry.id)
                != Some(id)
        {
            write_frame(&mut out, &Frame::Selection { id }, config.frame_bytes_max)?;
        }
        publication_checkpoint!(CompletedWrite);
        out.sync_all()
            .map_err(io_error(smith::error::StorageCode::CreateSync, &temporary))?;
        publication_checkpoint!(TempSynced);
        Ok(())
    })();
    drop(out);
    if let Err(error) = written {
        // Preserve the write failure; temporary-file removal is best-effort cleanup.
        let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
        return Err(error);
    }
    publish_no_clobber(&temporary, &path)?;
    publication_checkpoint!(Cleanup);
    let (writer, recovery) = SessionWriter::attach(path, config, lock)?;
    let session =
        SessionReader::open(writer.path())?.selected_session(writer.config.frame_bytes_max)?;
    Ok(OpenSession {
        session,
        writer,
        recovery,
    })
}

/// Visit frames within the file size observed at open, without retaining history.
///
/// A live incomplete tail is reported but never repaired by this read-only path.
/// Callers must inspect damage and torn tails before treating a scan as complete.
///
/// # Errors
/// Returns an input/output error or the visitor's error.
pub fn scan_session(
    path: &Path,
    frame_bytes_max: usize,
    visit: impl FnMut(Frame) -> Result<()>,
) -> Result<Recovery> {
    SessionReader::open(path)?.scan(frame_bytes_max, visit)
}

/// Whether the per-session exclusive writer lock is currently held.
///
/// # Errors
/// Returns an I/O error if the lock file cannot be probed.
pub fn writer_active(path: &Path) -> Result<bool> {
    let path = canonical_session_path(path)?;
    let file = match OpenOptions::new()
        .read(true)
        .write(true)
        .open(lock_path(&path))
    {
        Ok(file) => file,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(e) => {
            return Err(io_error(
                smith::error::StorageCode::StoreLockOpen,
                &lock_path(&path),
            )(e));
        }
    };
    match fs2::FileExt::try_lock_exclusive(&file) {
        Ok(()) => {
            fs2::FileExt::unlock(&file).map_err(io_error(
                smith::error::StorageCode::StoreUnlock,
                &lock_path(&path),
            ))?;
            Ok(false)
        }
        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(true),
        Err(e) => Err(io_error(
            smith::error::StorageCode::StoreLock,
            &lock_path(&path),
        )(e)),
    }
}

/// A pinned file generation with a shared maintenance lock, independent of writer ownership.
/// The snapshot length is captured at scan start; subsequent appends require a new scan.
#[derive(Debug)]
pub struct SessionReader {
    file: File,
    readers: File,
    snapshot_len: u64,
    writer_active: bool,
}

impl Drop for SessionReader {
    fn drop(&mut self) {
        // A fork shares this lock's open-file description until exec.  Unlock it
        // explicitly so a dropped snapshot cannot delay tail repair in that window.
        let _unlock_result: std::io::Result<()> = fs2::FileExt::unlock(&self.readers);
    }
}

struct SnapshotRead<'a> {
    file: &'a File,
    at: u64,
    end: u64,
}

impl Read for SnapshotRead<'_> {
    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
        let available = usize::try_from(self.end.saturating_sub(self.at))
            .unwrap_or(usize::MAX)
            .min(buffer.len());
        if available == 0 {
            return Ok(0);
        }
        #[cfg(unix)]
        let n = std::os::unix::fs::FileExt::read_at(self.file, &mut buffer[..available], self.at)?;
        #[cfg(windows)]
        let n =
            std::os::windows::fs::FileExt::seek_read(self.file, &mut buffer[..available], self.at)?;
        #[cfg(not(any(unix, windows)))]
        let n = {
            let mut clone = self.file.try_clone()?;
            clone.seek(SeekFrom::Start(self.at))?;
            clone.read(&mut buffer[..available])?
        };
        debug_assert!(
            n <= available,
            "snapshot reader must not advance beyond its validated byte window"
        );
        self.at += n as u64;
        Ok(n)
    }
}

#[cfg(test)]
thread_local! {
    static METADATA_PASSES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    static PROOF_VISITS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    static CHECKPOINT_HITS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}

// Fixed-width, offset-sorted structural proof. Scratch is unlinked before
// scanning; file size grows, resident memory does not. Only the sequential
// prologue walk can populate these records, never a magic-shaped payload.
struct Proof {
    file: File,
    path: PathBuf,
    len: u64,
    selected_checkpoint: Option<ProofEntry>,
    cancel: Option<CancelHandle>,
}

#[derive(Clone, Copy)]
struct ProofEntry {
    id: EntryId,
    parent: Option<EntryId>,
    offset: u64,
    parent_offset: Option<u64>,
    compaction: bool,
}

impl Proof {
    const WIDTH: u64 = 49;

    fn check_cancelled(&self) -> Result<()> {
        if self.cancel.as_ref().is_some_and(CancelHandle::is_cancelled) {
            return Err(SmithError::Cancelled);
        }
        Ok(())
    }

    fn get(&mut self, index: u64) -> Result<ProofEntry> {
        self.check_cancelled()?;
        #[cfg(test)]
        PROOF_VISITS.with(|visits| visits.set(visits.get() + 1));
        self.file
            .seek(SeekFrom::Start(index * Self::WIDTH))
            .map_err(io_error(
                smith::error::StorageCode::StoreProofSeek,
                &self.path,
            ))?;
        let mut bytes = [0; 49];
        self.file.read_exact(&mut bytes).map_err(io_error(
            smith::error::StorageCode::StoreProofRead,
            &self.path,
        ))?;
        let id =
            EntryId::from_u128(u128::from_le_bytes(bytes[..16].try_into().map_err(
                |_| session_fault(smith::error::SessionCode::StoreProof, "short id"),
            )?));
        let parent =
            u128::from_le_bytes(bytes[16..32].try_into().map_err(|_| {
                session_fault(smith::error::SessionCode::StoreProof, "short parent")
            })?);
        let offset =
            u64::from_le_bytes(bytes[32..40].try_into().map_err(|_| {
                session_fault(smith::error::SessionCode::StoreProof, "short offset")
            })?);
        let parent_offset = u64::from_le_bytes(
            bytes[40..48]
                .try_into()
                .map_err(|_| session_fault(smith::error::SessionCode::StoreProof, "short hint"))?,
        );
        Ok(ProofEntry {
            id,
            parent: (parent != 0).then(|| EntryId::from_u128(parent)),
            offset,
            parent_offset: (parent_offset != 0).then_some(parent_offset),
            compaction: bytes[48] != 0,
        })
    }

    fn push(&mut self, entry: &ProofEntry) -> Result<()> {
        self.check_cancelled()?;
        self.file
            .seek(SeekFrom::Start(self.len * Self::WIDTH))
            .map_err(io_error(
                smith::error::StorageCode::StoreProofSeek,
                &self.path,
            ))?;
        self.file
            .write_all(&entry.id.as_u128().to_le_bytes())
            .map_err(io_error(
                smith::error::StorageCode::StoreProofWrite,
                &self.path,
            ))?;
        self.file
            .write_all(&entry.parent.map_or(0, |id| id.as_u128()).to_le_bytes())
            .map_err(io_error(
                smith::error::StorageCode::StoreProofWrite,
                &self.path,
            ))?;
        self.file
            .write_all(&entry.offset.to_le_bytes())
            .map_err(io_error(
                smith::error::StorageCode::StoreProofWrite,
                &self.path,
            ))?;
        self.file
            .write_all(&entry.parent_offset.unwrap_or(0).to_le_bytes())
            .map_err(io_error(
                smith::error::StorageCode::StoreProofWrite,
                &self.path,
            ))?;
        self.file
            .write_all(&[u8::from(entry.compaction)])
            .map_err(io_error(
                smith::error::StorageCode::StoreProofWrite,
                &self.path,
            ))?;
        self.len += 1;
        Ok(())
    }

    fn set_hint(&mut self, at: u64) -> Result<()> {
        self.check_cancelled()?;
        self.file
            .seek(SeekFrom::Start((self.len - 1) * Self::WIDTH + 40))
            .map_err(io_error(
                smith::error::StorageCode::StoreProofSeek,
                &self.path,
            ))?;
        self.file.write_all(&at.to_le_bytes()).map_err(io_error(
            smith::error::StorageCode::StoreProofWrite,
            &self.path,
        ))
    }

    fn find(&mut self, id: EntryId) -> Result<Option<ProofEntry>> {
        let mut found = None;
        for i in 0..self.len {
            self.check_cancelled()?;
            let record = self.get(i)?;
            if record.id == id {
                if found.is_some() {
                    return Err(session_fault(
                        smith::error::SessionCode::SessionDuplicateId,
                        format!("duplicate entry {id}"),
                    ));
                }
                found = Some(record);
            }
        }
        Ok(found)
    }

    fn at_offset(&mut self, offset: u64) -> Result<Option<ProofEntry>> {
        let (mut start, mut end) = (0, self.len);
        while start < end {
            self.check_cancelled()?;
            let middle = start + (end - start) / 2;
            let record = self.get(middle)?;
            match record.offset.cmp(&offset) {
                std::cmp::Ordering::Less => start = middle + 1,
                std::cmp::Ordering::Greater => end = middle,
                std::cmp::Ordering::Equal => return Ok(Some(record)),
            }
        }
        Ok(None)
    }

    fn parent(&mut self, child: &ProofEntry) -> Result<Option<ProofEntry>> {
        self.check_cancelled()?;
        let Some(id) = child.parent else {
            return Ok(None);
        };
        // The hint is usable only if it names a structurally proven entry,
        // matches the parent identity, and precedes the child. Identity lookup also
        // rejects duplicate IDs, including outside the selected branch.
        let hinted = child
            .parent_offset
            .and_then(|at| (at < child.offset).then_some(at));
        let target = hinted.map(|at| self.at_offset(at)).transpose()?.flatten();
        let found = self.find(id)?.ok_or_else(|| missing_entry(id))?;
        let parent = target
            .filter(|candidate| candidate.id == id && candidate.offset == found.offset)
            .unwrap_or(found);
        if parent.offset >= child.offset {
            return Err(session_fault(
                smith::error::SessionCode::SessionParent,
                "parent does not precede child",
            ));
        }
        Ok(Some(parent))
    }
}

impl SessionReader {
    const fn snapshot_read(&self, at: u64) -> SnapshotRead<'_> {
        SnapshotRead {
            file: &self.file,
            at,
            end: self.snapshot_len,
        }
    }

    fn proof(
        &self,
        max: usize,
        cancel: Option<CancelHandle>,
    ) -> Result<(Proof, crate::session::SessionHeader, Option<EntryId>)> {
        let (path, file) = new_temp_file(
            &std::env::temp_dir().join("smith-proof"),
            "scan",
            SessionId::new(),
        )?;
        std::fs::remove_file(&path)
            .map_err(io_error(smith::error::StorageCode::StoreProofRemove, &path))?;
        let mut proof = Proof {
            file,
            path,
            len: 0,
            selected_checkpoint: None,
            cancel,
        };
        let mut header = None;
        let mut selected = None;
        let mut first = true;
        self.scan_metadata_positions(max, |offset, frame| {
            proof.check_cancelled()?;
            match frame {
                Frame::Header { header: h } if first => header = Some(h),
                Frame::Header { .. } => {
                    return Err(session_fault(
                        smith::error::SessionCode::SessionHeaderDuplicate,
                        "duplicate header",
                    ));
                }
                Frame::Known { entry } if header.is_some() => {
                    selected = Some(entry.id);
                    proof.selected_checkpoint = None;
                    proof.push(&ProofEntry {
                        id: entry.id,
                        parent: entry.parent,
                        offset,
                        parent_offset: None,
                        compaction: entry.kind
                            == (crate::session::EntryKind::Meta { compaction: true }),
                    })?;
                }
                Frame::Index {
                    id,
                    offset: target,
                    parent,
                    parent_offset,
                } if header.is_some() => {
                    if let Some(previous) = (proof.len > 0)
                        .then(|| proof.get(proof.len - 1))
                        .transpose()?
                        && previous.id == id
                        && previous.offset == target
                        && previous.parent == parent
                        && let Some(parent_at) = parent_offset.filter(|at| *at < target)
                    {
                        proof.set_hint(parent_at)?;
                    }
                }
                Frame::Selection { id } if header.is_some() => {
                    selected = Some(id);
                    proof.selected_checkpoint = None;
                }
                Frame::Checkpoint { id, offset: target } if header.is_some() => {
                    if selected == Some(id) && target < offset {
                        proof.selected_checkpoint =
                            proof.at_offset(target)?.filter(|entry| entry.id == id);
                    }
                }
                Frame::Unknown { .. } if header.is_some() => {}
                _ => {
                    return Err(session_fault(
                        smith::error::SessionCode::SessionHeaderMissing,
                        "missing first header",
                    ));
                }
            }
            first = false;
            Ok(())
        })?;
        Ok((
            proof,
            header.ok_or_else(|| {
                session_fault(
                    smith::error::SessionCode::SessionHeaderMissing,
                    "missing first header",
                )
            })?,
            selected,
        ))
    }

    fn scan_metadata_positions(
        &self,
        max: usize,
        mut visit: impl FnMut(u64, Frame) -> Result<()>,
    ) -> Result<Recovery> {
        #[cfg(test)]
        METADATA_PASSES.with(|passes| passes.set(passes.get() + 1));
        let mut reader = BufReader::with_capacity(64 * 1024, self.snapshot_read(0));
        read_prologue(&mut reader)?;
        visit_metadata_reporting(
            &mut reader,
            max,
            |offset, frame| visit(offset + FILE_PROLOGUE.len() as u64, frame),
            |_| Ok(()),
        )
        .map(file_offsets)
    }

    fn entry_at(&self, offset: u64, max: usize) -> Result<EntryFrame> {
        let mut file = self.snapshot_read(offset);
        let mut frame = None;
        let scan = visit_frames(&mut file, max, |found| {
            frame = Some(found);
            false
        })?;
        if let Some(fault) = scan.damage.into_iter().next() {
            return Err(fault);
        }
        match frame {
            Some(Frame::Known { entry }) => Ok(entry),
            _ => Err(session_fault(
                smith::error::SessionCode::SessionParent,
                "entry boundary no longer contains entry",
            )),
        }
    }

    /// Scan structural frame boundaries from the fixed prologue, reporting
    /// absolute offsets without retaining a per-file index.
    fn scan_positions(
        &self,
        max: usize,
        mut visit: impl FnMut(u64, Frame) -> Result<()>,
    ) -> Result<Recovery> {
        struct Counted<'a, R> {
            inner: R,
            bytes: &'a std::cell::Cell<u64>,
        }
        impl<R: Read> Read for Counted<'_, R> {
            fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
                let n = self.inner.read(buffer)?;
                self.bytes.set(self.bytes.get() + n as u64);
                Ok(n)
            }
        }
        let bytes = std::cell::Cell::new(0);
        let mut reader = Counted {
            inner: BufReader::with_capacity(64 * 1024, self.snapshot_read(0)),
            bytes: &bytes,
        };
        read_prologue(&mut reader)?;
        let offset = std::cell::Cell::new(FILE_PROLOGUE.len() as u64);
        let mut failure = None;
        let mut damage = None;
        let mut scan = visit_frames_reporting(
            &mut reader,
            max,
            |frame| {
                let current = offset.replace(bytes.get());
                match visit(current, frame) {
                    Ok(()) => true,
                    Err(error) => {
                        failure = Some(error);
                        false
                    }
                }
            },
            |fault| {
                offset.set(bytes.get());
                if damage.is_none() {
                    damage = Some(fault);
                }
                true
            },
        )?;
        if let Some(error) = failure {
            return Err(error);
        }
        if let Some(error) = damage {
            scan.damage.push(error);
        }
        Ok(file_offsets(scan))
    }

    /// Find an entry identity by walking authoritative entry frames at structural
    /// boundaries; embedded indexes are hints, never the source of identity.
    ///
    /// # Errors
    /// Returns storage faults or a duplicate identity error.
    pub fn find_entry(&self, id: EntryId, max: usize) -> Result<Option<(EntryFrame, u64)>> {
        let mut found = None;
        self.scan_metadata_positions(max, |offset, frame| {
            if let Frame::Known { entry } = frame
                && entry.id == id
            {
                if found.is_some() {
                    return Err(session_fault(
                        smith::error::SessionCode::SessionDuplicateId,
                        format!("duplicate entry {id}"),
                    ));
                }
                found = Some(offset);
            }
            Ok(())
        })?;
        found
            .map(|offset| self.entry_at(offset, max).map(|entry| (entry, offset)))
            .transpose()
    }

    /// Durable session and root-branch identity without materializing content.
    ///
    /// # Errors
    /// Returns a structural fault for an invalid or missing header.
    pub fn header(&self, max: usize) -> Result<crate::session::SessionHeader> {
        self.proof(max, None).map(|(_, header, _)| header)
    }

    /// Selected entry identity without materializing any content.
    ///
    /// # Errors
    /// Returns a structural fault for invalid selection or ancestry.
    pub fn selected_tip(&self, max: usize) -> Result<Option<EntryId>> {
        self.selected_tip_with_cancel(max, None)
    }

    /// Selected entry identity, checking cancellation at each bounded traversal boundary.
    ///
    /// # Errors
    /// Returns [`SmithError::Cancelled`] or a structural fault for invalid selection or ancestry.
    pub fn selected_tip_cancelled(
        &self,
        max: usize,
        cancel: &CancelHandle,
    ) -> Result<Option<EntryId>> {
        self.selected_tip_with_cancel(max, Some(cancel.clone()))
    }

    fn selected_tip_with_cancel(
        &self,
        max: usize,
        cancel: Option<CancelHandle>,
    ) -> Result<Option<EntryId>> {
        let (mut proof, _, selected) = self.proof(max, cancel)?;
        if let Some(id) = selected {
            proof.find(id)?.ok_or_else(|| missing_entry(id))?;
        }
        Ok(selected)
    }

    /// Selected effective context from latest compaction boundary to tip.
    /// Rootward ancestry is validated using a bounded, file-backed proof;
    /// only effective payloads become resident.
    ///
    /// # Errors
    /// Returns a fault for missing ancestry, invalid header or selection.
    pub fn selected_session(&self, max: usize) -> Result<Session> {
        self.selected_session_with_cancel(max, None)
    }

    fn selected_session_with_cancel(
        &self,
        max: usize,
        cancel: Option<CancelHandle>,
    ) -> Result<Session> {
        let (mut proof, header, selected) = self.proof(max, cancel)?;
        let (scratch_path, mut offsets) = scratch_file("context")?;
        let mut length = 0u64;
        if let Some(id) = selected {
            let mut cursor = proof.find(id)?.ok_or_else(|| missing_entry(id))?;
            if let Some(hint) = proof
                .selected_checkpoint
                .filter(|hint| hint.id == id && hint.offset == cursor.offset)
            {
                cursor = hint;
                #[cfg(test)]
                CHECKPOINT_HITS.with(|hits| hits.set(hits.get() + 1));
            }
            let mut keep = true;
            loop {
                proof.check_cancelled()?;
                if keep {
                    offsets
                        .write_all(&cursor.offset.to_le_bytes())
                        .map_err(io_error(
                            smith::error::StorageCode::StoreScratchWrite,
                            &scratch_path,
                        ))?;
                    length += 1;
                    if cursor.compaction {
                        keep = false;
                    }
                }
                match proof.parent(&cursor)? {
                    Some(parent) => cursor = parent,
                    None => break,
                }
            }
        }
        let mut session = Session::from_chain(header, Vec::new())?;
        for i in (0..length).rev() {
            proof.check_cancelled()?;
            offsets.seek(SeekFrom::Start(i * 8)).map_err(io_error(
                smith::error::StorageCode::StoreScratchSeek,
                &scratch_path,
            ))?;
            let mut bytes = [0; 8];
            offsets.read_exact(&mut bytes).map_err(io_error(
                smith::error::StorageCode::StoreScratchRead,
                &scratch_path,
            ))?;
            session.push_restored(self.entry_at(u64::from_le_bytes(bytes), max)?)?;
        }
        Ok(session)
    }

    /// Replay a rooted chain in chronological order with bounded retained
    /// payload. One metadata pass validates structural offsets into a
    /// file-backed proof; parent lookups scan fixed-width proof records.
    ///
    /// # Errors
    /// Returns a missing-entry, ancestry, storage, or callback fault.
    pub fn replay_at(
        &self,
        id: EntryId,
        max: usize,
        visit: impl FnMut(EntryFrame) -> Result<()>,
    ) -> Result<()> {
        self.replay_at_with_cancel(id, max, None, visit)
    }

    /// Replay a rooted chain, observing cancellation between bounded reads.
    ///
    /// # Errors
    /// Returns [`SmithError::Cancelled`], a storage fault, or a callback fault.
    pub fn replay_at_cancelled(
        &self,
        id: EntryId,
        max: usize,
        cancel: &CancelHandle,
        visit: impl FnMut(EntryFrame) -> Result<()>,
    ) -> Result<()> {
        self.replay_at_with_cancel(id, max, Some(cancel.clone()), visit)
    }

    fn replay_at_with_cancel(
        &self,
        id: EntryId,
        max: usize,
        cancel: Option<CancelHandle>,
        mut visit: impl FnMut(EntryFrame) -> Result<()>,
    ) -> Result<()> {
        // Spill only physical addresses, never payloads. Unlink immediately:
        // the open descriptor owns the scratch lifetime even on early errors.
        let (mut proof, _, _) = self.proof(max, cancel)?;
        let (scratch_path, mut scratch) = scratch_file("replay")?;
        let mut cursor = proof.find(id)?.ok_or_else(|| missing_entry(id))?;
        let mut depth = 0u64;
        loop {
            proof.check_cancelled()?;
            scratch
                .write_all(&cursor.offset.to_le_bytes())
                .map_err(io_error(
                    smith::error::StorageCode::StoreScratchWrite,
                    &scratch_path,
                ))?;
            depth += 1;
            match proof.parent(&cursor)? {
                Some(parent) => cursor = parent,
                None => break,
            }
        }
        for position in (0..depth).rev() {
            proof.check_cancelled()?;
            scratch
                .seek(SeekFrom::Start(position * 8))
                .map_err(io_error(
                    smith::error::StorageCode::StoreScratchSeek,
                    &scratch_path,
                ))?;
            let mut address = [0; 8];
            scratch.read_exact(&mut address).map_err(io_error(
                smith::error::StorageCode::StoreScratchRead,
                &scratch_path,
            ))?;
            visit(self.entry_at(u64::from_le_bytes(address), max)?)?;
        }
        Ok(())
    }

    /// Fork one structurally reachable chain to a new independently named file.
    ///
    /// # Errors
    /// Returns a missing-entry, ancestry, publication or storage fault.
    pub fn fork_at(&self, id: EntryId, path: &Path, config: Config) -> Result<OpenSession> {
        self.fork_at_with_cancel(id, path, config, None)
    }

    /// Fork one rooted chain, observing cancellation between bounded traversal work.
    ///
    /// # Errors
    /// Returns [`SmithError::Cancelled`], a storage fault, or a publication fault.
    pub fn fork_at_cancelled(
        &self,
        id: EntryId,
        path: &Path,
        config: Config,
        cancel: &CancelHandle,
    ) -> Result<OpenSession> {
        self.fork_at_with_cancel(id, path, config, Some(cancel.clone()))
    }

    fn fork_at_with_cancel(
        &self,
        id: EntryId,
        path: &Path,
        config: Config,
        cancel: Option<CancelHandle>,
    ) -> Result<OpenSession> {
        if cancel.as_ref().is_some_and(CancelHandle::is_cancelled) {
            return Err(SmithError::Cancelled);
        }
        config.validate()?;
        let source = path
            .file_stem()
            .and_then(|stem| stem.to_str())
            .and_then(|stem| stem.parse::<SessionId>().ok())
            .map_or_else(Session::new, Session::new_with_id);
        let lock = SessionLock::acquire(path)?;
        let target = lock.path().to_path_buf();
        validate_named_identity(&target, source.id())?;
        reject_existing_target(&target)?;
        let (temporary, mut out) = new_temp_file(&target, "fork", source.id())?;
        publication_checkpoint!(TempCreated);
        let written = (|| {
            out.write_all(&FILE_PROLOGUE)
                .map_err(io_error(smith::error::StorageCode::CreateWrite, &temporary))?;
            publication_checkpoint!(PartialWrite);
            write_frame(
                &mut out,
                &Frame::Header {
                    header: source.header(),
                },
                config.frame_bytes_max,
            )?;
            self.replay_at_with_cancel(id, config.frame_bytes_max, cancel.clone(), |entry| {
                write_frame(&mut out, &Frame::Known { entry }, config.frame_bytes_max)
            })?;
            publication_checkpoint!(CompletedWrite);
            out.sync_all()
                .map_err(io_error(smith::error::StorageCode::CreateSync, &temporary))?;
            publication_checkpoint!(TempSynced);
            Ok(())
        })();
        drop(out);
        if let Err(error) = written {
            // Preserve the write failure; temporary-file removal is best-effort cleanup.
            let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
            return Err(error);
        }
        publish_no_clobber(&temporary, &target)?;
        publication_checkpoint!(Cleanup);
        let (writer, recovery) = SessionWriter::attach(target, config, lock)?;
        let session = Self::open(writer.path())?
            .selected_session_with_cancel(writer.config.frame_bytes_max, cancel)?;
        Ok(OpenSession {
            session,
            writer,
            recovery,
        })
    }

    #[expect(
        clippy::large_types_passed_by_value,
        reason = "Copy IDs and offsets are compared once while validating an index record"
    )]
    fn verify_index(
        &self,
        last: Option<(EntryId, u64, Option<EntryId>)>,
        id: EntryId,
        target: u64,
        parent: Option<EntryId>,
        parent_offset: Option<u64>,
        max: usize,
    ) -> Result<()> {
        if last != Some((id, target, parent)) {
            return Err(session_fault(
                smith::error::SessionCode::SessionIndex,
                "index target is not preceding entry boundary",
            ));
        }
        if let Some(parent) = parent {
            let (_, actual) = self
                .find_entry(parent, max)?
                .ok_or_else(|| missing_entry(parent))?;
            if parent_offset != Some(actual) || actual >= target {
                return Err(session_fault(
                    smith::error::SessionCode::SessionIndex,
                    "index parent offset or identity mismatch",
                ));
            }
        } else if parent_offset.is_some() {
            return Err(session_fault(
                smith::error::SessionCode::SessionIndex,
                "root index has parent offset",
            ));
        }
        Ok(())
    }

    /// Check every frame and content body in the pinned snapshot, including
    /// unrelated branches and all embedded location hints.
    ///
    /// # Errors
    /// Reports damaged frames, invalid ancestry/offsets, or a torn tail.
    pub fn verify_full(&self, max: usize) -> Result<()> {
        let mut first = true;
        let mut last_entry = None;
        let scan = self.scan_positions(max, |offset, frame| {
            match frame {
                Frame::Header { .. } if first => {}
                Frame::Header { .. } => {
                    return Err(session_fault(
                        smith::error::SessionCode::SessionHeaderDuplicate,
                        "duplicate header",
                    ));
                }
                Frame::Known { entry } if !first => {
                    // Full verification rejects duplicate identities even when the
                    // selected branch never references either copy.
                    self.find_entry(entry.id, max)?
                        .ok_or_else(|| missing_entry(entry.id))?;
                    let content = crate::content_codec::EntryContentView::decode(&entry.content)?;
                    if content.kind() != entry.kind {
                        return Err(session_fault(
                            smith::error::SessionCode::EntryKindMismatch,
                            format!("entry {} kind mismatch", entry.id),
                        ));
                    }
                    if let Some(parent) = entry.parent {
                        let (_, parent_offset) = self
                            .find_entry(parent, max)?
                            .ok_or_else(|| missing_entry(parent))?;
                        if parent_offset >= offset {
                            return Err(session_fault(
                                smith::error::SessionCode::SessionParent,
                                "parent does not precede child",
                            ));
                        }
                    }
                    last_entry = Some((entry.id, offset, entry.parent));
                }
                Frame::Index {
                    id,
                    offset: target,
                    parent,
                    parent_offset,
                } if !first => {
                    self.verify_index(last_entry, id, target, parent, parent_offset, max)?;
                    last_entry = None;
                }
                Frame::Selection { id } if !first => {
                    self.find_entry(id, max)?.ok_or_else(|| missing_entry(id))?;
                    last_entry = None;
                }
                Frame::Checkpoint { id, offset: target } if !first => {
                    if self
                        .find_entry(id, max)?
                        .is_none_or(|(_, actual)| actual != target)
                        || target >= offset
                    {
                        return Err(session_fault(
                            smith::error::SessionCode::SessionCheckpoint,
                            "checkpoint target is not an earlier entry boundary",
                        ));
                    }
                    last_entry = None;
                }
                Frame::Unknown { .. } if !first => last_entry = None,
                _ => {
                    return Err(session_fault(
                        smith::error::SessionCode::SessionHeaderMissing,
                        "missing first header",
                    ));
                }
            }
            first = false;
            Ok(())
        })?;
        if first {
            return Err(session_fault(
                smith::error::SessionCode::SessionHeaderMissing,
                "missing first header",
            ));
        }
        if let Some(fault) = scan.damage.into_iter().next() {
            return Err(fault);
        }
        if scan.torn_tail_at.is_some() {
            return Err(session_fault(
                smith::error::SessionCode::SessionTorn,
                "incomplete tail",
            ));
        }
        Ok(())
    }

    /// Open without taking the writer lock or modifying session content.
    ///
    /// # Errors
    /// Returns an input/output error for missing data or failed coordination.
    pub fn open(path: &Path) -> Result<Self> {
        Self::open_with_cancel(path, None)
    }

    /// Open a reader snapshot, observing cancellation while maintenance holds the reader lock.
    ///
    /// # Errors
    /// Returns [`SmithError::Cancelled`] or an input/output error for failed coordination.
    pub fn open_cancelled(path: &Path, cancel: &CancelHandle) -> Result<Self> {
        Self::open_with_cancel(path, Some(cancel))
    }

    fn open_with_cancel(path: &Path, cancel: Option<&CancelHandle>) -> Result<Self> {
        let path = canonical_session_path(path)?;
        let readers = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(readers_path(&path))
            .map_err(io_error(
                smith::error::StorageCode::StoreReaderLockOpen,
                &readers_path(&path),
            ))?;
        if let Some(cancel) = cancel {
            loop {
                if cancel.is_cancelled() {
                    return Err(SmithError::Cancelled);
                }
                match fs2::FileExt::try_lock_shared(&readers) {
                    Ok(()) => break,
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                        std::thread::sleep(std::time::Duration::from_millis(10));
                    }
                    Err(error) => {
                        return Err(io_error(
                            smith::error::StorageCode::StoreReaderLock,
                            &readers_path(&path),
                        )(error));
                    }
                }
            }
        } else {
            fs2::FileExt::lock_shared(&readers).map_err(io_error(
                smith::error::StorageCode::StoreReaderLock,
                &readers_path(&path),
            ))?;
        }
        let file =
            File::open(&path).map_err(io_error(smith::error::StorageCode::StoreOpen, &path))?;
        let snapshot_len = file
            .metadata()
            .map_err(io_error(smith::error::StorageCode::StoreStat, &path))?
            .len();
        let writer_active = writer_active(&path)?;
        Ok(Self {
            file,
            readers,
            snapshot_len,
            writer_active,
        })
    }

    /// Visit a consistent file-size snapshot with per-frame bounded allocation.
    ///
    /// # Errors
    /// Returns an I/O error or the visitor's error.
    pub fn scan(
        &self,
        frame_bytes_max: usize,
        mut visit: impl FnMut(Frame) -> Result<()>,
    ) -> Result<Recovery> {
        let mut reader = BufReader::with_capacity(64 * 1024, self.snapshot_read(0));
        read_prologue(&mut reader)?;
        let mut failure = None;
        let scan = visit_frames(&mut reader, frame_bytes_max, |frame| match visit(frame) {
            Ok(()) => true,
            Err(error) => {
                failure = Some(error);
                false
            }
        });
        let scan = file_offsets(scan?);
        if let Some(error) = failure {
            return Err(error);
        }
        Ok(scan)
    }

    /// Decode the first frame only, without reading the history.
    ///
    /// # Errors
    /// Returns an I/O error when the file cannot be read.
    pub fn first(&self, frame_bytes_max: usize) -> Result<(Option<Frame>, Recovery)> {
        let first_limit = u64::try_from(frame_bytes_max)
            .unwrap_or(u64::MAX)
            .saturating_add(
                u64::try_from(
                    FILE_PROLOGUE.len() + crate::frame::HEADER_LEN + crate::frame::BODY_PREFIX_LEN,
                )
                .unwrap_or(u64::MAX),
            )
            .min(self.snapshot_len);
        let mut reader = BufReader::new(self.snapshot_read(0).take(first_limit));
        read_prologue(&mut reader)?;
        let mut first = None;
        let scan = visit_frames(&mut reader, frame_bytes_max, |frame| {
            first = Some(frame);
            false
        })?;
        Ok((first, file_offsets(scan)))
    }

    /// Whether a writer owned this session when this reader acquired its snapshot.
    #[must_use]
    pub const fn writer_active(&self) -> bool {
        self.writer_active
    }

    /// File size of the pinned generation at reader acquisition.
    #[must_use]
    pub const fn file_size(&self) -> u64 {
        self.snapshot_len
    }

    /// Scan a snapshot while reporting damage through `report_damage`.
    ///
    /// Unlike [`SessionReader::scan`], this method retains no damage, keeping
    /// callers that report or reject faults incrementally bounded.
    ///
    /// # Errors
    /// Returns an I/O error, visitor error, or damage-reporting error.
    pub fn scan_reporting(
        &self,
        frame_bytes_max: usize,
        mut visit: impl FnMut(Frame) -> Result<bool>,
        mut report_damage: impl FnMut(SmithError) -> Result<()>,
    ) -> Result<Recovery> {
        let mut reader = BufReader::with_capacity(64 * 1024, self.snapshot_read(0));
        read_prologue(&mut reader)?;
        let failure = std::cell::RefCell::new(None);
        let scan = visit_frames_reporting(
            &mut reader,
            frame_bytes_max,
            |frame| match visit(frame) {
                Ok(keep_going) => keep_going,
                Err(error) => {
                    *failure.borrow_mut() = Some(error);
                    false
                }
            },
            |damage| match report_damage(shift_damage(damage)) {
                Ok(()) => true,
                Err(error) => {
                    *failure.borrow_mut() = Some(error);
                    false
                }
            },
        )?;
        if let Some(error) = failure.into_inner() {
            return Err(error);
        }
        Ok(file_offsets(scan))
    }
}

/// Repair a session file by republishing only its complete frames.
///
/// Damaged frames are dropped; [`Recovery::damage_count`] counts all faults
/// while [`Recovery::damage`] retains only the first report.
/// An explicit selection whose entry is dropped fails before publication.
/// The original content stays intact if any step fails before publication.
/// Repair takes the session lock, so it waits for a live writer to finish
/// instead of replacing the file underneath it.
///
/// # Errors
///
/// Returns a locking, framing, or input/output error.
pub fn repair_session(path: &Path, frame_bytes_max: usize) -> Result<Recovery> {
    let lock = SessionLock::acquire(path)?;
    let path = lock.path();
    let _coordination = maintenance_lock(path)?;
    let (temporary, mut out) = new_temp_file(path, "repair", SessionId::new())?;
    publication_checkpoint!(TempCreated);
    let written = (|| {
        out.write_all(&FILE_PROLOGUE)
            .map_err(io_error(smith::error::StorageCode::RepairWrite, &temporary))?;
        publication_checkpoint!(PartialWrite);
        let mut source = BufReader::with_capacity(
            64 * 1024,
            File::open(path).map_err(io_error(smith::error::StorageCode::StoreOpen, path))?,
        );
        read_prologue(&mut source)?;
        let mut last = None;
        let mut count = 0u16;
        let mut selection = None;
        let mut has_header = false;
        let mut first_damage = None;
        let mut failure = None;
        let mut recovery = visit_frames_reporting(
            &mut source,
            frame_bytes_max,
            |frame| {
                let result = (|| {
                    match frame {
                        Frame::Index { .. } | Frame::Checkpoint { .. } => {}
                        Frame::Selection { id } => selection = Some(id),
                        Frame::Header { header } => {
                            has_header = true;
                            write_frame(&mut out, &Frame::Header { header }, frame_bytes_max)?;
                        }
                        Frame::Known { entry } => {
                            repair_entry(
                                &mut out,
                                path,
                                entry,
                                frame_bytes_max,
                                has_header,
                                &mut last,
                                &mut count,
                            )?;
                        }
                        other @ Frame::Unknown { .. } => {
                            write_frame(&mut out, &other, frame_bytes_max)?;
                        }
                    }
                    Ok(())
                })();
                if let Err(error) = result {
                    failure = Some(error);
                    false
                } else {
                    true
                }
            },
            |damage| {
                if first_damage.is_none() {
                    first_damage = Some(shift_damage(damage));
                }
                true
            },
        )?;
        if let Some(error) = failure {
            return Err(error);
        }
        finish_repair_selection(&temporary, &mut out, selection.as_ref(), frame_bytes_max)?;
        if let Some(damage) = first_damage {
            recovery.damage.push(damage);
        }
        publication_checkpoint!(CompletedWrite);
        out.sync_all()
            .map_err(io_error(smith::error::StorageCode::RepairSync, &temporary))?;
        publication_checkpoint!(TempSynced);
        Ok(file_offsets(recovery))
    })();
    drop(out);
    let recovery = match written {
        Ok(recovery) => recovery,
        Err(error) => {
            // Preserve the recovery failure; temporary-file removal is best-effort cleanup.
            let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
            return Err(error);
        }
    };
    if let Err(error) = std::fs::rename(&temporary, path) {
        // Preserve the publish failure; temporary-file removal is best-effort cleanup.
        let _cleanup_result: std::io::Result<()> = std::fs::remove_file(&temporary);
        return Err(io_error(smith::error::StorageCode::RepairPublish, path)(
            error,
        ));
    }
    publication_checkpoint!(Published);
    sync_parent(path)?;
    publication_checkpoint!(ParentSynced);
    // `rename` removed the temporary name as it published the replacement.
    publication_checkpoint!(Cleanup);
    drop(lock);
    Ok(recovery)
}

fn repair_entry(
    out: &mut File,
    path: &Path,
    entry: EntryFrame,
    max: usize,
    has_header: bool,
    last: &mut Option<(EntryId, u64)>,
    count: &mut u16,
) -> Result<()> {
    let offset = out
        .stream_position()
        .map_err(io_error(smith::error::StorageCode::RepairSeek, path))?;
    let id = entry.id;
    let parent = entry.parent;
    write_frame(out, &Frame::Known { entry }, max)?;
    let parent_offset =
        parent.and_then(|parent| last.filter(|(id, _)| *id == parent).map(|(_, at)| at));
    *last = Some((id, offset));
    if has_header && (parent.is_none() || parent_offset.is_some()) && max >= 48 {
        out.write_all(&index_bytes(id, offset, parent, parent_offset))
            .map_err(io_error(smith::error::StorageCode::RepairWrite, path))?;
    }
    *count += 1;
    if has_header && *count == 64 && max >= 24 {
        write_frame(out, &Frame::Checkpoint { id, offset }, max)?;
        *count = 0;
    }
    Ok(())
}

fn maintenance_lock(path: &Path) -> Result<File> {
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(readers_path(path))
        .map_err(io_error(
            smith::error::StorageCode::StoreReaderLockOpen,
            &readers_path(path),
        ))?;
    fs2::FileExt::try_lock_exclusive(&file).map_err(io_error(
        smith::error::StorageCode::StoreReadersActive,
        &readers_path(path),
    ))?;
    Ok(file)
}

fn finish_repair_selection(
    path: &Path,
    out: &mut File,
    selection: Option<&EntryId>,
    max: usize,
) -> Result<()> {
    let Some(&id) = selection else {
        return Ok(());
    };
    let mut source = BufReader::new(
        File::open(path).map_err(io_error(smith::error::StorageCode::RepairOpen, path))?,
    );
    read_prologue(&mut source)?;
    let mut found = None;
    visit_metadata_reporting(
        &mut source,
        max,
        |at, frame| {
            if let Frame::Known { entry } = frame
                && entry.id == id
            {
                found = Some(at + FILE_PROLOGUE.len() as u64);
            }
            Ok(())
        },
        |_| Ok(()),
    )?;
    let offset = found.ok_or_else(|| missing_entry(id))?;
    write_frame(out, &Frame::Selection { id }, max)?;
    if max >= 24 {
        write_frame(out, &Frame::Checkpoint { id, offset }, max)?;
    }
    Ok(())
}

fn recovery_checkpoint_count(file: &File, path: &Path, max: usize) -> Result<u16> {
    let mut reader = BufReader::new(
        file.try_clone()
            .map_err(io_error(smith::error::StorageCode::StoreOpen, path))?,
    );
    reader
        .seek(SeekFrom::Start(0))
        .map_err(io_error(smith::error::StorageCode::StoreSeek, path))?;
    read_prologue(&mut reader)?;
    let (path, file) = scratch_file("cadence")?;
    let mut proof = Proof {
        file,
        path,
        len: 0,
        selected_checkpoint: None,
        cancel: None,
    };
    let mut count = 0;
    visit_metadata_reporting(
        &mut reader,
        max,
        |offset, frame| {
            match frame {
                Frame::Known { entry } => {
                    proof.push(&ProofEntry {
                        id: entry.id,
                        parent: entry.parent,
                        offset: offset + FILE_PROLOGUE.len() as u64,
                        parent_offset: None,
                        compaction: false,
                    })?;
                    count = (count + 1) % 64;
                }
                Frame::Checkpoint { id, offset: at }
                    if proof.at_offset(at)?.is_some_and(|entry| entry.id == id) =>
                {
                    count = 0;
                }
                _ => {}
            }
            Ok(())
        },
        |_| Ok(()),
    )?;
    Ok(count)
}

/// Walk an already-open file from its start using the shared frame reader.
///
/// # Errors
///
/// Returns an input/output error when the file cannot be re-read.
fn walk_file_bounded(file: &File, path: &Path, frame_bytes_max: usize) -> Result<Recovery> {
    let mut handle = file
        .try_clone()
        .map_err(io_error(smith::error::StorageCode::StoreOpen, path))?;
    handle
        .seek(SeekFrom::Start(0))
        .map_err(io_error(smith::error::StorageCode::StoreSeek, path))?;
    let mut reader = BufReader::with_capacity(64 * 1024, handle);
    read_prologue(&mut reader)?;
    let mut damage = None;
    let mut recovery = visit_frames_reporting(
        &mut reader,
        frame_bytes_max,
        |_| true,
        |fault| {
            if damage.is_none() {
                damage = Some(fault);
            }
            true
        },
    )
    .map(file_offsets)?;
    if let Some(fault) = damage {
        recovery.damage.push(shift_damage(fault));
    }
    Ok(recovery)
}

fn read_prologue(reader: &mut impl Read) -> Result<()> {
    let mut magic = [0; FILE_PROLOGUE.len()];
    reader.read_exact(&mut magic).map_err(io_error(
        smith::error::StorageCode::SessionPrologue,
        Path::new("<session>"),
    ))?;
    if magic != FILE_PROLOGUE {
        return Err(SmithError::Session {
            code: smith::error::SessionCode::SessionPrologue,
            message: "invalid session format or version; expected SMH3 prologue".into(),
        });
    }
    Ok(())
}

fn file_offsets(mut recovery: Recovery) -> Recovery {
    recovery.good_end += FILE_PROLOGUE.len() as u64;
    recovery.torn_tail_at = recovery
        .torn_tail_at
        .map(|n| n + FILE_PROLOGUE.len() as u64);
    for damage in &mut recovery.damage {
        if let SmithError::Frame {
            offset: Some(n), ..
        } = damage
        {
            *n += FILE_PROLOGUE.len() as u64;
        }
    }
    recovery
}

const fn shift_damage(mut damage: SmithError) -> SmithError {
    if let SmithError::Frame {
        offset: Some(n), ..
    } = &mut damage
    {
        *n += FILE_PROLOGUE.len() as u64;
    }
    damage
}

/// Rename a synced temporary file into an absent target without any fallback.
///
/// `renamore::rename_exclusive` is `renameat2(RENAME_NOREPLACE)` on Linux,
/// `renamex_np(RENAME_EXCL)` on macOS, and `MoveFileExW` without replace flags
/// on Windows. Unsupported filesystems fail rather than weakening no-clobber.
fn publish_no_clobber(temporary: &Path, target: &Path) -> Result<()> {
    renamore::rename_exclusive(temporary, target)
        .map_err(io_error(smith::error::StorageCode::CreatePublish, target))?;
    publication_checkpoint!(Published);
    sync_parent(target)?;
    publication_checkpoint!(ParentSynced);
    Ok(())
}

fn sync_parent(path: &Path) -> Result<()> {
    let parent = path
        .parent()
        .filter(|p| !p.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));
    #[cfg(test)]
    if PARENT_SYNC_FAILURE.with(std::cell::Cell::get) {
        return Err(published_parent_sync_error(
            parent,
            "injected parent directory sync failure",
        ));
    }
    sync_directory(parent).map_err(|error| published_parent_sync_error(parent, error))
}

fn published_parent_sync_error(path: &Path, error: impl std::fmt::Display) -> SmithError {
    SmithError::Storage {
        code: smith::error::StorageCode::StoreDirSync,
        operation: smith::error::StorageOperation::SyncParentDirectoryAfterPublication,
        path: path.to_path_buf(),
        message: error.to_string(),
        publication: Some(smith::error::PublicationState::PublishedDurabilityUncertain),
    }
}

#[cfg(not(windows))]
fn sync_directory(path: &Path) -> std::io::Result<()> {
    File::open(path)?.sync_all()
}

#[cfg(windows)]
fn sync_directory(path: &Path) -> std::io::Result<()> {
    use std::os::windows::fs::OpenOptionsExt;

    // FILE_FLAG_BACKUP_SEMANTICS permits opening a directory handle safely.
    OpenOptions::new()
        .read(true)
        .custom_flags(0x0200_0000)
        .open(path)?
        .sync_all()
}

fn session_fault(code: smith::error::SessionCode, message: impl Into<String>) -> SmithError {
    SmithError::Session {
        code,
        message: message.into(),
    }
}

fn missing_entry(id: EntryId) -> SmithError {
    session_fault(
        smith::error::SessionCode::EntryNotFound,
        format!("entry {id} not found"),
    )
}

fn io_error(code: smith::error::StorageCode, path: &Path) -> impl Fn(std::io::Error) -> SmithError {
    storage_error(code, storage_operation(code), path)
}

const fn storage_operation(code: smith::error::StorageCode) -> smith::error::StorageOperation {
    use smith::error::{StorageCode, StorageOperation};
    match code {
        StorageCode::StoreCanonicalize => StorageOperation::Canonicalize,
        StorageCode::StoreStat | StorageCode::CreateStat => StorageOperation::Stat,
        StorageCode::StoreOpen
        | StorageCode::StoreLockOpen
        | StorageCode::StoreReaderLockOpen
        | StorageCode::RepairOpen => StorageOperation::Open,
        StorageCode::StoreRead
        | StorageCode::StoreProofRead
        | StorageCode::StoreScratchRead
        | StorageCode::FrameRead
        | StorageCode::SessionPrologue
        | StorageCode::FrameBoundary
        | StorageCode::AuthRead => StorageOperation::Read,
        StorageCode::StoreWrite
        | StorageCode::StoreProofWrite
        | StorageCode::StoreScratchWrite
        | StorageCode::CreateWrite
        | StorageCode::RepairWrite
        | StorageCode::StorePoisoned => StorageOperation::Write,
        StorageCode::StoreSeek
        | StorageCode::StoreProofSeek
        | StorageCode::StoreScratchSeek
        | StorageCode::RepairSeek => StorageOperation::Seek,
        StorageCode::StoreSync
        | StorageCode::CreateSync
        | StorageCode::RepairSync
        | StorageCode::StoreDirSync => StorageOperation::Sync,
        StorageCode::StoreTruncate => StorageOperation::Truncate,
        StorageCode::StoreMkdir | StorageCode::RuntimeDir => StorageOperation::CreateDirectory,
        StorageCode::AuthStore => StorageOperation::AuthStoreOperation,
        StorageCode::StoreLock
        | StorageCode::StoreBusy
        | StorageCode::StoreReaderLock
        | StorageCode::StoreReadersActive => StorageOperation::Lock,
        StorageCode::StoreUnlock | StorageCode::StoreReaderUnlock => StorageOperation::Unlock,
        StorageCode::StoreProofRemove
        | StorageCode::StoreScratchRemove
        | StorageCode::CreateTempRemove => StorageOperation::Remove,
        StorageCode::CreateTmp => StorageOperation::CreateTemporaryFile,
        StorageCode::CreatePublish | StorageCode::RepairPublish => StorageOperation::Publish,
    }
}

fn storage_fault(
    code: smith::error::StorageCode,
    path: &Path,
    message: impl Into<String>,
) -> SmithError {
    SmithError::Storage {
        code,
        operation: storage_operation(code),
        path: path.to_path_buf(),
        message: message.into(),
        publication: None,
    }
}

fn storage_error(
    code: smith::error::StorageCode,
    operation: smith::error::StorageOperation,
    path: &Path,
) -> impl Fn(std::io::Error) -> SmithError + '_ {
    move |e| SmithError::Storage {
        code,
        operation,
        path: path.to_path_buf(),
        message: e.to_string(),
        publication: None,
    }
}

#[cfg(test)]
#[expect(
    clippy::unwrap_used,
    clippy::panic,
    reason = "tests may panic on invariant violations"
)]
mod tests {
    use super::*;
    use crate::session::COMPACTION_KIND;
    use crate::session::{EntryContent, EntryFrame, EntryKind, Session};
    use smith::message::{ContentBlock, Message, Role};
    use smith::structured::{StructuredData, ToolArguments};
    use smith_alloc::{Scope, measure};
    use std::sync::Mutex;

    static MEASURING: Mutex<()> = Mutex::new(());

    fn tmp_path(name: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!("smith_store_tests_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        dir.join(name)
    }

    fn message(text: &str) -> EntryContent {
        EntryContent::Message(Message::with_text(Role::User, text))
    }

    fn entry(text: &str) -> EntryFrame {
        EntryFrame::new(None, &message(text)).unwrap()
    }

    fn text(entry: &EntryFrame) -> Option<String> {
        match EntryContent::decode(&entry.content).unwrap() {
            EntryContent::Message(m) => match &m.blocks[0] {
                ContentBlock::Text(t) => Some(t.clone()),
                _ => None,
            },
            _ => None,
        }
    }

    fn frame(text: &str) -> Frame {
        Frame::Known { entry: entry(text) }
    }

    fn file_bytes(frames: impl IntoIterator<Item = u8>) -> Vec<u8> {
        FILE_PROLOGUE.into_iter().chain(frames).collect()
    }

    /// Eager frame collection belongs only in recovery assertions; durable
    /// product paths use `SessionReader`'s bounded readers directly.
    fn scanned_frames(path: &Path, frame_bytes_max: usize) -> Recovery {
        let reader = SessionReader::open(path).unwrap();
        let mut frames = Vec::new();
        let mut recovery = reader
            .scan(frame_bytes_max, |frame| {
                frames.push(frame);
                Ok(())
            })
            .unwrap();
        recovery.frames = frames;
        recovery
    }

    #[test]
    fn durable_selection_replays_and_forks_a_sibling_without_loading_unrelated_content() {
        let path = tmp_path("durable-selection.smh");
        let fork_path = tmp_path("durable-selection-fork.smh");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&fork_path);
        let cfg = Config::default_valid();
        let root = entry("root");
        let left = EntryFrame::new(Some(root.id), &message("left")).unwrap();
        let right = EntryFrame::new(Some(root.id), &message("right")).unwrap();
        {
            let mut opened = open_session(&path, cfg.clone()).unwrap();
            for entry in [&root, &left, &right] {
                opened
                    .writer
                    .append_frame(&Frame::Known {
                        entry: entry.clone(),
                    })
                    .unwrap();
            }
            opened.writer.select_entry(left.id).unwrap();
        }
        let reopened = open_session(&path, cfg.clone()).unwrap();
        assert_eq!(reopened.session.selected_id(), Some(left.id));
        assert_eq!(reopened.session.active_branch().entries().len(), 2);
        assert_eq!(
            reopened
                .session
                .content(&reopened.session.active_branch().entries()[1])
                .unwrap(),
            EntryContent::decode(&left.content).unwrap()
        );
        drop(reopened);
        let reader = SessionReader::open(&path).unwrap();
        let fork = reader.fork_at(right.id, &fork_path, cfg).unwrap();
        assert_eq!(fork.session.selected_id(), Some(right.id));
        assert_ne!(
            fork.session.id(),
            SessionReader::open(&path)
                .unwrap()
                .selected_session(Config::default_valid().frame_bytes_max)
                .unwrap()
                .id()
        );
        assert_eq!(
            fork.session
                .content(&fork.session.active_branch().entries()[1])
                .unwrap(),
            EntryContent::decode(&right.content).unwrap()
        );
        drop(fork);
        drop(reader);
        let _ = std::fs::remove_file(path);
        let _ = std::fs::remove_file(fork_path);
    }

    #[test]
    fn full_verification_detects_duplicate_ids_outside_selected_branch() {
        let path = tmp_path("duplicate-unselected.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let unselected = entry("one");
        let mut duplicate = entry("two");
        duplicate.id = unselected.id;
        for entry in [unselected, duplicate] {
            opened.writer.append_frame(&Frame::Known { entry }).unwrap();
        }
        drop(opened);
        let reader = SessionReader::open(&path).unwrap();
        assert_eq!(
            reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
            "SESSION_DUPLICATE_ID"
        );
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn missing_session_file_reports_storage_operation_path_and_source() {
        let path = tmp_path("missing-storage-error.smh");
        let _ = std::fs::remove_file(&path);
        let error = SessionReader::open(&path).unwrap_err();
        match error {
            SmithError::Storage {
                code,
                operation,
                path: actual,
                message,
                ..
            } => {
                assert_eq!(code, smith::error::StorageCode::StoreOpen);
                assert_eq!(operation.as_str(), "open");
                assert_eq!(actual, path);
                assert!(!message.is_empty());
            }
            other => panic!("expected storage error, got {other:?}"),
        }
        let _ = std::fs::remove_file(readers_path(&path));
    }

    #[test]
    fn missing_selected_entry_fails_open_without_retargeting_a_branch() {
        let path = tmp_path("missing-selected.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        opened
            .writer
            .append_frame(&Frame::Known {
                entry: entry("valid"),
            })
            .unwrap();
        opened
            .writer
            .append_frame(&Frame::Selection { id: EntryId::new() })
            .unwrap();
        drop(opened);
        assert_eq!(
            open_session(&path, cfg).unwrap_err().code(),
            "ENTRY_NOT_FOUND"
        );
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn damaged_index_does_not_hide_following_entries() {
        let path = tmp_path("damaged-index.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let session = Session::new();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        let mut bytes = FILE_PROLOGUE.to_vec();
        bytes.extend(
            Frame::Header {
                header: session.header(),
            }
            .encode(cfg.frame_bytes_max)
            .unwrap(),
        );
        bytes.extend(
            Frame::Known {
                entry: root.clone(),
            }
            .encode(cfg.frame_bytes_max)
            .unwrap(),
        );
        let mut invalid = Frame::Index {
            id: root.id,
            offset: 8,
            parent: None,
            parent_offset: None,
        }
        .encode(cfg.frame_bytes_max)
        .unwrap();
        invalid[crate::frame::HEADER_LEN + 4] ^= 0x80;
        bytes.extend(invalid);
        bytes.extend(
            Frame::Known {
                entry: child.clone(),
            }
            .encode(cfg.frame_bytes_max)
            .unwrap(),
        );
        std::fs::write(&path, bytes).unwrap();
        let reader = SessionReader::open(&path).unwrap();
        assert_eq!(
            reader
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .selected_id(),
            Some(child.id)
        );
        assert_eq!(
            reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
            "FRAME_CHECKSUM"
        );
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn repair_discards_stale_offsets_without_losing_authoritative_entries() {
        let path = tmp_path("repair-index-offset.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let record = *opened.session.append(&message("kept")).unwrap();
        opened
            .writer
            .append_entry(&record, opened.session.bytes(&record))
            .unwrap();
        let selected = record.id;
        drop(opened);
        let original = std::fs::read(&path).unwrap();
        let header_end = FILE_PROLOGUE.len()
            + Frame::Header {
                header: SessionReader::open(&path)
                    .unwrap()
                    .selected_session(cfg.frame_bytes_max)
                    .unwrap()
                    .header(),
            }
            .encode(cfg.frame_bytes_max)
            .unwrap()
            .len();
        let mut bytes = original[..header_end].to_vec();
        let mut damaged = Frame::Unknown {
            kind: 99,
            version: 1,
            payload: b"bad".to_vec(),
        }
        .encode(cfg.frame_bytes_max)
        .unwrap();
        damaged[crate::frame::HEADER_LEN + 4] ^= 1;
        bytes.extend(damaged);
        bytes.extend(&original[header_end..]);
        std::fs::write(&path, bytes).unwrap();
        repair_session(&path, cfg.frame_bytes_max).unwrap();
        let reader = SessionReader::open(&path).unwrap();
        reader.verify_full(cfg.frame_bytes_max).unwrap();
        assert_eq!(
            reader
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .selected_id(),
            Some(selected)
        );
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn reopen_keeps_only_effective_payload_after_compaction() {
        let path = tmp_path("effective-compaction.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let before = *opened
            .session
            .append(&message(&"x".repeat(400_000)))
            .unwrap();
        opened
            .writer
            .append_entry(&before, opened.session.bytes(&before))
            .unwrap();
        let compact = EntryContent::Meta {
            kind: COMPACTION_KIND.into(),
            detail: StructuredData::from_serializable(&serde_json::json!({"summary": ""})).unwrap(),
        };
        let boundary = *opened.session.append(&compact).unwrap();
        opened
            .writer
            .append_entry(&boundary, opened.session.bytes(&boundary))
            .unwrap();
        let after = *opened.session.append(&message("after")).unwrap();
        opened
            .writer
            .append_entry(&after, opened.session.bytes(&after))
            .unwrap();
        drop(opened);
        let reopened = open_session(&path, cfg).unwrap();
        assert_eq!(reopened.session.selected_id(), Some(after.id));
        assert_eq!(reopened.session.compaction_boundary(), Some(boundary.id));
        assert_eq!(reopened.session.active_branch().entries().len(), 2);
        assert!(reopened.session.bytes(&before).is_empty());
        drop(reopened);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn replay_visits_file_metadata_once_for_linear_history() {
        let path = tmp_path("replay-pass-count.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let mut parent = None;
        for _ in 0..96 {
            let e = EntryFrame::new(parent, &message("x")).unwrap();
            parent = Some(e.id);
            opened
                .writer
                .append_frame(&Frame::Known { entry: e })
                .unwrap();
        }
        drop(opened);
        let reader = SessionReader::open(&path).unwrap();
        METADATA_PASSES.with(|passes| passes.set(0));
        let mut visited = 0;
        reader
            .replay_at(parent.unwrap(), cfg.frame_bytes_max, |_| {
                visited += 1;
                Ok(())
            })
            .unwrap();
        assert_eq!(visited, 96);
        METADATA_PASSES
            .with(|passes| assert!(passes.get() <= 2, "{} metadata passes", passes.get()));
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn effective_context_cannot_publish_a_broken_rooted_fork() {
        let path = tmp_path("partial-source.smh");
        let target = tmp_path("partial-target.smh");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&target);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let first = *opened.session.append(&message("before")).unwrap();
        opened
            .writer
            .append_entry(&first, opened.session.bytes(&first))
            .unwrap();
        let compact = EntryContent::Meta {
            kind: COMPACTION_KIND.into(),
            detail: StructuredData::from_serializable(&serde_json::json!({"summary": ""})).unwrap(),
        };
        let boundary = *opened.session.append(&compact).unwrap();
        opened
            .writer
            .append_entry(&boundary, opened.session.bytes(&boundary))
            .unwrap();
        drop(opened);
        let resumed = open_session(&path, cfg.clone()).unwrap();
        assert_eq!(
            create_session(&target, &resumed.session, cfg)
                .unwrap_err()
                .code(),
            "SESSION_INCOMPLETE_CONTEXT"
        );
        assert!(!target.exists());
        drop(resumed);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn snapshot_reads_pinned_inode_after_uncoordinated_replacement() {
        let path = tmp_path("pinned-generation.smh");
        let replacement = tmp_path("pinned-generation-replacement.smh");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&replacement);
        let cfg = Config::default_valid();
        let mut first = open_session(&path, cfg.clone()).unwrap();
        let root = entry("old");
        first
            .writer
            .append_frame(&Frame::Known {
                entry: root.clone(),
            })
            .unwrap();
        let old_id = first.session.id();
        drop(first);
        let snapshot = SessionReader::open(&path).unwrap();
        let newer = open_session(&replacement, cfg.clone()).unwrap();
        let new_id = newer.session.id();
        drop(newer);
        std::fs::rename(&replacement, &path).unwrap();
        assert_eq!(
            snapshot.selected_session(cfg.frame_bytes_max).unwrap().id(),
            old_id
        );
        assert_eq!(
            snapshot
                .find_entry(root.id, cfg.frame_bytes_max)
                .unwrap()
                .unwrap()
                .0
                .id,
            root.id
        );
        assert_eq!(
            SessionReader::open(&path)
                .unwrap()
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .id(),
            new_id
        );
        drop(snapshot);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn replay_and_fork_visit_growth_is_quadratic_not_cubic() {
        fn visits(depth: usize) -> (usize, usize) {
            let path = tmp_path(&format!("growth-{depth}.smh"));
            let target = tmp_path(&format!("growth-fork-{depth}.smh"));
            let _ = std::fs::remove_file(&path);
            let _ = std::fs::remove_file(&target);
            let cfg = Config::default_valid();
            let mut opened = open_session(&path, cfg.clone()).unwrap();
            let mut parent = None;
            for _ in 0..depth {
                let entry = EntryFrame::new(parent, &message("x")).unwrap();
                parent = Some(entry.id);
                opened.writer.append_frame(&Frame::Known { entry }).unwrap();
            }
            drop(opened);
            let reader = SessionReader::open(&path).unwrap();
            PROOF_VISITS.with(|visits| visits.set(0));
            reader
                .replay_at(parent.unwrap(), cfg.frame_bytes_max, |_| Ok(()))
                .unwrap();
            let replay = PROOF_VISITS.with(std::cell::Cell::get);
            PROOF_VISITS.with(|visits| visits.set(0));
            let fork = reader.fork_at(parent.unwrap(), &target, cfg).unwrap();
            assert_eq!(fork.session.selected_id(), parent);
            let fork_visits = PROOF_VISITS.with(std::cell::Cell::get);
            drop(fork);
            drop(reader);
            let _ = std::fs::remove_file(path);
            let _ = std::fs::remove_file(target);
            (replay, fork_visits)
        }
        let small = visits(64);
        let large = visits(128);
        assert!(
            large.0 <= 5 * small.0,
            "replay proof visits: {small:?} -> {large:?}"
        );
        assert!(
            large.1 <= 5 * small.1,
            "fork proof visits: {small:?} -> {large:?}"
        );
    }

    #[test]
    fn create_keeps_selection_of_existing_ancestor() {
        let path = tmp_path("create-ancestor-selection.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        let source = Session::from_frames(&[
            Frame::Header {
                header: Session::new().header(),
            },
            Frame::Known {
                entry: root.clone(),
            },
            Frame::Known { entry: child },
            Frame::Selection { id: root.id },
        ])
        .unwrap();
        let opened = create_session(&path, &source, cfg.clone()).unwrap();
        assert_eq!(opened.session.selected_id(), Some(root.id));
        drop(opened);
        let reopened = open_session(&path, cfg).unwrap();
        assert_eq!(reopened.session.selected_id(), Some(root.id));
        drop(reopened);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn valid_checkpoint_locates_selected_ancestor() {
        let path = tmp_path("selected-checkpoint-hint.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        for record in [&root, &child] {
            opened
                .writer
                .append_frame(&Frame::Known {
                    entry: record.clone(),
                })
                .unwrap();
        }
        opened.writer.select_entry(root.id).unwrap();
        drop(opened);
        CHECKPOINT_HITS.with(|hits| hits.set(0));
        let reader = SessionReader::open(&path).unwrap();
        assert_eq!(
            reader
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .selected_id(),
            Some(root.id)
        );
        CHECKPOINT_HITS.with(|hits| assert_eq!(hits.get(), 1));
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn invalid_checkpoint_does_not_reset_writer_cadence() {
        let path = tmp_path("invalid-cadence.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        for _ in 0..63 {
            let record = *opened.session.append(&message("turn")).unwrap();
            opened
                .writer
                .append_entry(&record, opened.session.bytes(&record))
                .unwrap();
        }
        opened
            .writer
            .append_frame(&Frame::Checkpoint {
                id: EntryId::new(),
                offset: 123,
            })
            .unwrap();
        drop(opened);
        let mut reopened = open_session(&path, cfg.clone()).unwrap();
        let record = *reopened.session.append(&message("last")).unwrap();
        reopened
            .writer
            .append_entry(&record, reopened.session.bytes(&record))
            .unwrap();
        drop(reopened);
        let mut count = 0;
        SessionReader::open(&path)
            .unwrap()
            .scan(cfg.frame_bytes_max, |frame| {
                if matches!(frame, Frame::Checkpoint { .. }) {
                    count += 1;
                }
                Ok(())
            })
            .unwrap();
        assert_eq!(count, 2);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn writer_publishes_reconstructible_checkpoints_at_the_cadence() {
        let path = tmp_path("checkpoint-cadence.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        for _ in 0..63 {
            let record = *opened.session.append(&message("turn")).unwrap();
            opened
                .writer
                .append_entry(&record, opened.session.bytes(&record))
                .unwrap();
        }
        drop(opened);
        let mut reopened = open_session(&path, cfg.clone()).unwrap();
        let record = *reopened.session.append(&message("turn")).unwrap();
        reopened
            .writer
            .append_entry(&record, reopened.session.bytes(&record))
            .unwrap();
        let selected = reopened.session.selected_id();
        drop(reopened);
        let reader = SessionReader::open(&path).unwrap();
        let mut checkpoints = 0;
        reader
            .scan(cfg.frame_bytes_max, |frame| {
                if matches!(frame, Frame::Checkpoint { .. }) {
                    checkpoints += 1;
                }
                Ok(())
            })
            .unwrap();
        assert_eq!(checkpoints, 1);
        reader.verify_full(cfg.frame_bytes_max).unwrap();
        assert_eq!(
            reader
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .selected_id(),
            selected
        );
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn forged_checkpoint_is_ignored_for_resume_but_fails_full_verification() {
        let path = tmp_path("forged-checkpoint.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        opened
            .writer
            .append_frame(&Frame::Known {
                entry: root.clone(),
            })
            .unwrap();
        opened
            .writer
            .append_frame(&Frame::Checkpoint {
                id: root.id,
                offset: FILE_PROLOGUE.len() as u64 + 1,
            })
            .unwrap();
        opened
            .writer
            .append_frame(&Frame::Known {
                entry: child.clone(),
            })
            .unwrap();
        drop(opened);
        let reader = SessionReader::open(&path).unwrap();
        assert_eq!(
            reader
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .selected_id(),
            Some(child.id)
        );
        assert_eq!(
            reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
            "SESSION_CHECKPOINT"
        );
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn forged_index_never_hides_later_entries_and_full_verification_rejects_it() {
        let path = tmp_path("forged-index.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        opened
            .writer
            .append_frame(&Frame::Known {
                entry: root.clone(),
            })
            .unwrap();
        opened
            .writer
            .append_frame(&Frame::Index {
                id: root.id,
                offset: FILE_PROLOGUE.len() as u64 + 1,
                parent: None,
                parent_offset: None,
            })
            .unwrap();
        opened
            .writer
            .append_frame(&Frame::Known {
                entry: child.clone(),
            })
            .unwrap();
        drop(opened);
        let reader = SessionReader::open(&path).unwrap();
        assert_eq!(
            reader
                .selected_session(cfg.frame_bytes_max)
                .unwrap()
                .selected_id(),
            Some(child.id)
        );
        assert_eq!(
            reader.verify_full(cfg.frame_bytes_max).unwrap_err().code(),
            "SESSION_INDEX"
        );
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn create_validates_config_before_creating_session_files() {
        let path = tmp_path("config-before-effects/nested/session.smh");
        let _ = std::fs::remove_dir_all(path.parent().unwrap().parent().unwrap());
        let mut config = Config::default_valid();
        config.frame_bytes_max = 0;
        assert!(create_session(&path, &Session::new(), config).is_err());
        assert!(!path.parent().unwrap().exists());
    }

    #[test]
    fn create_rejects_named_identity_mismatch_before_publication() {
        let session = Session::new();
        let path = tmp_path(&format!("{}.smh", smith::id::SessionId::new()));
        let _ = std::fs::remove_file(&path);
        assert_eq!(
            create_session(&path, &session, Config::default_valid())
                .unwrap_err()
                .code(),
            "SESSION_ID_MISMATCH"
        );
        assert!(!path.exists());
        let _ = std::fs::remove_file(lock_path(&path));
    }

    #[cfg(unix)]
    #[test]
    fn create_temp_symlink_cannot_truncate_its_target() {
        let path = tmp_path("create-temp-link.smh");
        let temporary = path.with_extension("smith-create.tmp");
        let victim = tmp_path("create-temp-victim");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&temporary);
        std::fs::write(&victim, b"do not truncate").unwrap();
        std::os::unix::fs::symlink(&victim, &temporary).unwrap();
        create_session(&path, &Session::new(), Config::default_valid()).unwrap();
        assert_eq!(std::fs::read(&victim).unwrap(), b"do not truncate");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&temporary);
        let _ = std::fs::remove_file(&victim);
    }

    #[cfg(unix)]
    #[test]
    fn repair_temp_symlink_cannot_truncate_its_target() {
        let path = tmp_path("repair-temp-link.smh");
        let temporary = path.with_extension("smith-repair.tmp");
        let victim = tmp_path("repair-temp-victim");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&temporary);
        std::fs::write(&path, file_bytes(frame("intact").encode(1 << 20).unwrap())).unwrap();
        std::fs::write(&victim, b"do not truncate").unwrap();
        std::os::unix::fs::symlink(&victim, &temporary).unwrap();
        repair_session(&path, Config::default_valid().frame_bytes_max).unwrap();
        assert_eq!(std::fs::read(&victim).unwrap(), b"do not truncate");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&temporary);
        let _ = std::fs::remove_file(&victim);
    }

    fn text_of(frame: &Frame) -> String {
        match frame {
            Frame::Known { entry } => text(entry).unwrap(),
            other => panic!("unexpected frame {other:?}"),
        }
    }

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

    #[test]
    fn reporting_scan_does_not_retain_streamed_damage() {
        let path = tmp_path("reporting-damage.smh");
        let _ = std::fs::remove_file(&path);
        let mut malformed = frame("bad").encode(1 << 20).unwrap();
        malformed[crate::frame::HEADER_LEN + 4] ^= 0xff;
        std::fs::write(&path, file_bytes(malformed.repeat(128))).unwrap();
        let reader = SessionReader::open(&path).unwrap();
        let mut reported = 0;
        let recovery = reader
            .scan_reporting(
                Config::default_valid().frame_bytes_max,
                |_| Ok(true),
                |_| {
                    reported += 1;
                    Ok(())
                },
            )
            .unwrap();
        assert_eq!(reported, 128);
        assert!(recovery.damage.is_empty());
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn reader_snapshot_keeps_initial_writer_ownership_and_length() {
        let path = tmp_path("reader-snapshot.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
        writer.append_frame(&frame("before")).unwrap();
        let reader = SessionReader::open(&path).unwrap();
        assert!(reader.writer_active());
        let snapshot_len = reader.file_size();
        writer.append_frame(&frame("after")).unwrap();
        drop(writer);
        assert_eq!(reader.file_size(), snapshot_len);
        let mut frames = 0;
        reader
            .scan(cfg.frame_bytes_max, |_| {
                frames += 1;
                Ok(())
            })
            .unwrap();
        assert_eq!(frames, 1);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn streaming_scan_visits_entries_without_retaining_history() {
        // SMH-SPEC-SPEC0001: Sessions, Working sets; a snapshot never follows live appends.
        let path = tmp_path("stream_scan.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
        for _ in 0..128 {
            writer.append_frame(&frame("entry")).unwrap();
        }
        let mut count = 0;
        let scan = scan_session(&path, cfg.frame_bytes_max, |frame| {
            assert!(matches!(frame, Frame::Known { .. }));
            count += 1;
            Ok(())
        })
        .unwrap();
        assert_eq!(count, 128);
        assert!(scan.damage.is_empty());
        assert_eq!(scan.torn_tail_at, None);
        let mut corrupt = File::options().append(true).open(&path).unwrap();
        corrupt.write_all(&[0, 1]).unwrap();
        let scan = scan_session(&path, cfg.frame_bytes_max, |_| Ok(())).unwrap();
        assert!(scan.torn_tail_at.is_some());
        drop(writer);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn append_load_round_trip_preserves_entries() {
        let path = tmp_path("round_trip.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let first = entry("one");
        let second = EntryFrame::new(Some(first.id), &message("two")).unwrap();
        {
            let mut opened = open_session(&path, cfg.clone()).unwrap();
            opened
                .writer
                .append_frame(&Frame::Known {
                    entry: first.clone(),
                })
                .unwrap();
            opened
                .writer
                .append_frame(&Frame::Known {
                    entry: second.clone(),
                })
                .unwrap();
        }
        let session = SessionReader::open(&path)
            .unwrap()
            .selected_session(cfg.frame_bytes_max)
            .unwrap();
        assert_eq!(
            session
                .active_branch()
                .entries()
                .iter()
                .map(|e| e.id)
                .collect::<Vec<_>>(),
            vec![first.id, second.id]
        );
        assert_eq!(session.selected_id(), Some(second.id));
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn successful_io_error_context_setup_does_not_allocate() {
        let path = Path::new("/tmp/smith-store-allocation-check");
        let _serial = MEASURING
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let (_, cost) = measure(Scope::IoRequest, || {
            let error = io_error(smith::error::StorageCode::StoreWrite, path);
            drop(std::hint::black_box(error));
        });
        assert_eq!(cost.allocations, 0);
        assert_eq!(cost.bytes, 0);
    }

    #[test]
    fn successful_write_does_not_allocate_error_context() {
        let path = tmp_path("write-allocation-check.smh");
        let _ = std::fs::remove_file(&path);
        let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
        let bytes = frame("allocation-check")
            .encode(Config::default_valid().frame_bytes_max)
            .unwrap();
        let _serial = MEASURING
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let (result, cost) = measure(Scope::IoRequest, || writer.write(&bytes));
        result.unwrap();
        assert_eq!(cost.allocations, 0);
        assert_eq!(cost.bytes, 0);
        drop(writer);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn torn_tail_is_truncated_on_open_and_appends_stay_visible() {
        let path = tmp_path("torn_tail.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        {
            let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
            w.append_frame(&frame("kept")).unwrap();
        }
        // Simulate a crash mid-append: partial header + payload bytes.
        let good = Frame::Known {
            entry: entry("kept"),
        }
        .encode(cfg.frame_bytes_max)
        .unwrap();
        let mut torn = good;
        torn.extend_from_slice(&[0x10, 0x00, 0x00, 0x00, 0xAA, 0xBB]);
        std::fs::write(&path, file_bytes(torn)).unwrap();
        {
            // Open repairs the tail, then a new append must be loadable.
            let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
            w.append_frame(&frame("after")).unwrap();
        }
        let frames = scanned_frames(&path, cfg.frame_bytes_max).frames;
        let texts: Vec<String> = frames
            .iter()
            .filter_map(|f| match f {
                Frame::Known { entry } => text(entry),
                _ => None,
            })
            .collect();
        assert_eq!(texts, vec!["kept".to_string(), "after".to_string()]);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn loader_stops_at_torn_tail_without_header_fabrication() {
        let path = tmp_path("loader_torn.smh");
        let cfg = Config::default_valid();
        let good = frame("a").encode(cfg.frame_bytes_max).unwrap();
        let mut bytes = good;
        bytes.extend_from_slice(&[0xFF, 0xFF]); // partial header
        std::fs::write(&path, file_bytes(bytes)).unwrap();
        let frames = scanned_frames(&path, cfg.frame_bytes_max).frames;
        assert_eq!(frames.len(), 1);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn reopening_a_multi_frame_session_keeps_every_frame() {
        let path = tmp_path("reopen_multi.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        {
            let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
            w.append_frame(&frame("one")).unwrap();
            w.append_frame(&frame("two")).unwrap();
        }
        {
            // Reopening must not mistake a later frame boundary for a torn tail.
            let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
            w.append_frame(&frame("three")).unwrap();
        }
        let recovery = scanned_frames(&path, cfg.frame_bytes_max);
        assert_eq!(recovery.frames.len(), 3);
        assert!(recovery.damage.is_empty());
        assert_eq!(recovery.torn_tail_at, None);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn oversized_complete_frame_is_skipped_and_following_frames_reachable() {
        let path = tmp_path("oversized_skip.smh");
        // declared = 5000 > limit 1024, payload fully present, then a valid frame.
        let mut bytes = Frame::Unknown {
            kind: 9,
            version: 1,
            payload: vec![0xAB; 5000],
        }
        .encode(6000)
        .unwrap();
        bytes.extend_from_slice(&frame("tail").encode(1 << 20).unwrap());
        std::fs::write(&path, file_bytes(bytes)).unwrap();
        let recovery = scanned_frames(&path, 1024);
        assert_eq!(recovery.frames.len(), 1);
        assert!(matches!(recovery.frames[0], Frame::Known { .. }));
        assert_eq!(recovery.damage[0].code(), "FRAME_TOO_LARGE");
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn repair_reports_all_faults_without_retaining_their_frames() {
        let path = tmp_path("repair-multiple-faults.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut bytes = FILE_PROLOGUE.to_vec();
        for _ in 0..2 {
            let mut broken = frame("damaged").encode(cfg.frame_bytes_max).unwrap();
            broken[crate::frame::HEADER_LEN + 4] ^= 1;
            bytes.extend(broken);
        }
        std::fs::write(&path, bytes).unwrap();
        let recovery = repair_session(&path, cfg.frame_bytes_max).unwrap();
        assert_eq!(recovery.damage_count, 2);
        assert_eq!(recovery.damage.len(), 1);
        assert!(recovery.frames.is_empty());
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn repair_streams_unknown_frames_and_keeps_final_selection() {
        let path = tmp_path("repair-stream-selection.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        let unknown = Frame::Unknown {
            kind: 0xfabc,
            version: 17,
            payload: vec![42; 1024],
        };
        let mut opened = open_session(&path, cfg.clone()).unwrap();
        for entry in [&root, &child] {
            opened
                .writer
                .append_frame(&Frame::Known {
                    entry: entry.clone(),
                })
                .unwrap();
        }
        opened.writer.append_frame(&unknown).unwrap();
        opened.writer.select_entry(root.id).unwrap();
        drop(opened);
        let recovery = repair_session(&path, cfg.frame_bytes_max).unwrap();
        assert!(
            recovery.frames.is_empty(),
            "repair must not retain historical frames"
        );
        let mut seen_unknown = false;
        SessionReader::open(&path)
            .unwrap()
            .scan(cfg.frame_bytes_max, |frame| {
                if frame == unknown {
                    seen_unknown = true;
                }
                Ok(())
            })
            .unwrap();
        assert!(seen_unknown);
        let reopened = open_session(&path, cfg).unwrap();
        assert_eq!(reopened.session.selected_id(), Some(root.id));
        drop(reopened);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn repair_rejects_damaged_explicit_selection_without_publication() {
        let dir = tmp_path("repair-selected-damaged");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("session.smh");
        let cfg = Config::default_valid();
        let session = Session::new();
        let root = entry("root");
        let selected = EntryFrame::new(Some(root.id), &message("selected")).unwrap();
        let mut bytes = FILE_PROLOGUE.to_vec();
        for frame in [
            Frame::Header {
                header: session.header(),
            },
            Frame::Known {
                entry: root.clone(),
            },
            Frame::Known {
                entry: selected.clone(),
            },
            Frame::Selection { id: selected.id },
        ] {
            bytes.extend(frame.encode(cfg.frame_bytes_max).unwrap());
        }
        let selected_start = FILE_PROLOGUE.len()
            + Frame::Header {
                header: session.header(),
            }
            .encode(cfg.frame_bytes_max)
            .unwrap()
            .len()
            + Frame::Known { entry: root }
                .encode(cfg.frame_bytes_max)
                .unwrap()
                .len();
        let selected_checksum = selected_start + crate::frame::HEADER_LEN + 4;
        bytes[selected_checksum] ^= 1;
        std::fs::write(&path, &bytes).unwrap();
        let before_temps = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(std::result::Result::ok)
            .filter(|entry| {
                entry
                    .file_name()
                    .to_string_lossy()
                    .starts_with("session.smh.smith-repair-")
            })
            .count();

        let error = repair_session(&path, cfg.frame_bytes_max).unwrap_err();

        assert_eq!(error.code(), "ENTRY_NOT_FOUND");
        assert_eq!(std::fs::read(&path).unwrap(), bytes);
        let after_temps = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(std::result::Result::ok)
            .filter(|entry| {
                entry
                    .file_name()
                    .to_string_lossy()
                    .starts_with("session.smh.smith-repair-")
            })
            .count();
        assert_eq!(after_temps, before_temps);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn repair_republishes_file_without_torn_tail() {
        let path = tmp_path("repair.smh");
        let cfg = Config::default_valid();
        let good = frame("a").encode(cfg.frame_bytes_max).unwrap();
        let mut bytes = good;
        bytes.extend_from_slice(&frame("b").encode(cfg.frame_bytes_max).unwrap()[..6]);
        std::fs::write(&path, file_bytes(bytes)).unwrap();
        let before = std::fs::metadata(&path).unwrap().len();
        let recovery = repair_session(&path, cfg.frame_bytes_max).unwrap();
        assert!(recovery.frames.is_empty());
        assert_eq!(recovery.torn_tail_at, Some(before - 6));
        let after = std::fs::metadata(&path).unwrap().len();
        assert!(after < before);
        assert_eq!(scanned_frames(&path, cfg.frame_bytes_max).frames.len(), 1);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn writer_rejects_oversized_frame_without_touching_file() {
        let path = tmp_path("oversize_reject.smh");
        let _ = std::fs::remove_file(&path);
        let mut cfg = Config::default_valid();
        cfg.frame_bytes_max = 4096;
        {
            let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
            w.append_frame(&frame("small")).unwrap();
            let big = EntryFrame::new(
                None,
                &EntryContent::Message({
                    let mut m = Message::with_text(Role::User, "x");
                    m.add_block(ContentBlock::text("y".repeat(8192)));
                    m
                }),
            )
            .unwrap();
            let error = w.append_frame(&Frame::Known { entry: big }).unwrap_err();
            assert_eq!(error.code(), "FRAME_TOO_LARGE");
        }
        assert_eq!(scanned_frames(&path, cfg.frame_bytes_max).frames.len(), 1);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn session_identity_mismatch_does_not_mutate_source() {
        // SMH-SPEC-SPEC0001: Sessions, identity independent of offsets and filenames.
        let original = Session::new();
        let path = tmp_path(&format!("{}.smh", original.id()));
        let renamed = tmp_path(&format!("{}.smh", smith::id::SessionId::new()));
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&renamed);
        drop(open_session(&path, Config::default_valid()).unwrap());
        std::fs::rename(&path, &renamed).unwrap();
        let before = std::fs::read(&renamed).unwrap();
        assert_eq!(
            open_session(&renamed, Config::default_valid())
                .unwrap_err()
                .code(),
            "SESSION_ID_MISMATCH"
        );
        assert_eq!(std::fs::read(&renamed).unwrap(), before);
        let _ = std::fs::remove_file(renamed);
    }

    #[test]
    fn observer_prevents_implicit_tail_repair() {
        // SMH-SPEC-SPEC0001: Sessions, maintenance cannot invalidate a pinned reader.
        let path = tmp_path("observer_torn.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
        writer.append_frame(&frame("intact")).unwrap();
        drop(writer);
        let mut tail = File::options().append(true).open(&path).unwrap();
        tail.write_all(&[0, 1]).unwrap();
        let before = std::fs::read(&path).unwrap();
        let observer = SessionReader::open(&path).unwrap();
        assert_eq!(
            SessionWriter::open(&path, cfg).unwrap_err().code(),
            "STORE_READERS_ACTIVE"
        );
        assert_eq!(std::fs::read(&path).unwrap(), before);
        drop(observer);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn observer_coexists_with_writer_but_blocks_replacement() {
        // SMH-SPEC-SPEC0001: Sessions, Access ownership and repair coordination.
        let path = tmp_path("observer_lock.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
        writer.append_frame(&frame("one")).unwrap();
        let observer = SessionReader::open(&path).unwrap();
        assert_eq!(
            observer
                .scan(cfg.frame_bytes_max, |_| Ok(()))
                .unwrap()
                .good_end,
            std::fs::metadata(&path).unwrap().len()
        );
        assert_eq!(
            SessionWriter::open(&path, cfg.clone()).unwrap_err().code(),
            "STORE_BUSY"
        );
        drop(writer);
        assert_eq!(
            repair_session(&path, cfg.frame_bytes_max)
                .unwrap_err()
                .code(),
            "STORE_READERS_ACTIVE"
        );
        drop(observer);
        repair_session(&path, cfg.frame_bytes_max).unwrap();
        let _ = std::fs::remove_file(path);
    }

    #[cfg(unix)]
    #[test]
    fn symlinked_data_and_parent_paths_share_one_writer_lock() {
        let dir = tmp_path("symlinked-writer-lock");
        let real_dir = dir.join("real");
        let parent_alias = dir.join("parent-alias");
        let data_alias = dir.join("data-alias.smh");
        let real_path = real_dir.join("session.smh");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&real_dir).unwrap();
        std::os::unix::fs::symlink(&real_dir, &parent_alias).unwrap();

        let writer = SessionWriter::open(&real_path, Config::default_valid()).unwrap();
        std::os::unix::fs::symlink(&real_path, &data_alias).unwrap();
        for alias in [&parent_alias.join("session.smh"), &data_alias] {
            assert_eq!(
                SessionWriter::open(alias, Config::default_valid())
                    .unwrap_err()
                    .code(),
                "STORE_BUSY"
            );
        }
        drop(writer);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[cfg(unix)]
    #[test]
    fn multiply_linked_session_data_is_rejected() {
        let dir = tmp_path("multiply-linked-session");
        let path = dir.join("session.smh");
        let alias = dir.join("alias.smh");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        drop(SessionWriter::open(&path, Config::default_valid()).unwrap());
        std::fs::hard_link(&path, &alias).unwrap();

        assert_eq!(
            SessionWriter::open(&path, Config::default_valid())
                .unwrap_err()
                .code(),
            "SESSION_AMBIGUOUS_ALIAS"
        );
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn failed_write_poisoning_rejects_later_appends() {
        let path = tmp_path("poisoned-writer.smh");
        let _ = std::fs::remove_file(&path);
        let mut writer = SessionWriter::open(&path, Config::default_valid()).unwrap();
        let healthy = writer.file.try_clone().unwrap();
        writer.file = File::open(path.parent().unwrap()).unwrap();
        assert_eq!(
            writer.append_frame(&frame("partial")).unwrap_err().code(),
            "STORE_WRITE"
        );
        writer.file = healthy;
        assert_eq!(
            writer.append_frame(&frame("later")).unwrap_err().code(),
            "STORE_POISONED"
        );
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn fault_injected_partial_write_and_sync_failure_poison_writers() {
        let partial_path = tmp_path("poisoned-partial-writer.smh");
        let sync_path = tmp_path("poisoned-sync-writer.smh");
        let _ = std::fs::remove_file(&partial_path);
        let _ = std::fs::remove_file(&sync_path);
        let bytes = frame("pending")
            .encode(Config::default_valid().frame_bytes_max)
            .unwrap();

        let mut partial = SessionWriter::open(&partial_path, Config::default_valid()).unwrap();
        assert_eq!(
            partial
                .write_step(|file| {
                    file.write_all(&bytes[..bytes.len() / 2]).map_err(io_error(
                        smith::error::StorageCode::StoreWrite,
                        &partial_path,
                    ))?;
                    Err(storage_fault(
                        smith::error::StorageCode::StoreWrite,
                        &partial_path,
                        "injected partial write failure",
                    ))
                })
                .unwrap_err()
                .code(),
            "STORE_WRITE"
        );
        assert_eq!(
            partial.append_frame(&frame("later")).unwrap_err().code(),
            "STORE_POISONED"
        );
        drop(partial);

        let mut sync = SessionWriter::open(&sync_path, Config::default_valid()).unwrap();
        assert_eq!(
            sync.write_step(|file| {
                file.write_all(&bytes)
                    .map_err(io_error(smith::error::StorageCode::StoreWrite, &sync_path))?;
                Err(storage_fault(
                    smith::error::StorageCode::StoreSync,
                    &sync_path,
                    "injected sync failure",
                ))
            })
            .unwrap_err()
            .code(),
            "STORE_SYNC"
        );
        assert_eq!(
            sync.append_frame(&frame("later")).unwrap_err().code(),
            "STORE_POISONED"
        );
        drop(sync);
        let _ = std::fs::remove_file(partial_path);
        let _ = std::fs::remove_file(sync_path);
    }

    #[test]
    fn lock_serializes_second_writer_handle() {
        let path = tmp_path("locked.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let writer = SessionWriter::open(&path, cfg).unwrap();
        // While the first writer holds the session, an independent handle
        // cannot take it: writers cannot interleave frames.
        let probe = OpenOptions::new()
            .read(true)
            .write(true)
            .open(lock_path(&path))
            .unwrap();
        assert!(fs2::FileExt::try_lock_exclusive(&probe).is_err());
        drop(writer);
        assert!(fs2::FileExt::try_lock_exclusive(&probe).is_ok());
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn repair_waits_for_a_live_writer_and_keeps_its_entries() {
        let path = tmp_path("repair_race.smh");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(lock_path(&path));
        let cfg = Config::default_valid();
        let mut writer = SessionWriter::open(&path, cfg.clone()).unwrap();
        writer.append_frame(&frame("one")).unwrap();

        let repair_path = path.clone();
        let limit = cfg.frame_bytes_max;
        let (entered, repair_started) = std::sync::mpsc::sync_channel(1);
        let repair = std::thread::spawn(move || {
            entered.send(()).unwrap();
            repair_session(&repair_path, limit).unwrap()
        });

        // synchronize: the repair has started; the writer's lock holds it
        // until the drop, so an append now is still committed to the file
        // the repair will republish.
        repair_started
            .recv_timeout(std::time::Duration::from_secs(10))
            .unwrap();
        writer.append_frame(&frame("two")).unwrap();
        drop(writer);

        let recovery = repair.join().unwrap();
        assert!(recovery.frames.is_empty());
        let loaded = scanned_frames(&path, cfg.frame_bytes_max);
        let texts: Vec<String> = loaded.frames.iter().map(text_of).collect();
        assert_eq!(texts, vec!["one".to_string(), "two".to_string()]);
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(lock_path(&path));
    }

    #[test]
    fn unknown_frames_survive_write_and_load() {
        let path = tmp_path("unknown_frames.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let unknown = Frame::Unknown {
            kind: 0x4242,
            version: 3,
            payload: vec![1, 2, 3, 4, 5, 6],
        };
        {
            let mut w = SessionWriter::open(&path, cfg.clone()).unwrap();
            w.append_frame(&unknown).unwrap();
            w.append_frame(&frame("known")).unwrap();
        }
        let frames = scanned_frames(&path, cfg.frame_bytes_max).frames;
        assert_eq!(frames.len(), 2);
        match &frames[0] {
            Frame::Unknown {
                kind,
                version,
                payload,
            } => {
                assert_eq!(payload, &vec![1, 2, 3, 4, 5, 6]);
                assert_eq!((*kind, *version), (0x4242, 3));
            }
            other => panic!("expected unknown frame, got {other:?}"),
        }
        match &frames[1] {
            Frame::Known { entry } => assert_eq!(text(entry).as_deref(), Some("known")),
            other => panic!("expected known frame, got {other:?}"),
        }
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn end_to_end_tool_activity_persists_and_reconstructs() {
        let dir = tmp_path("e2e");
        std::fs::create_dir_all(&dir).unwrap();
        let workdir = dir.join("work");
        std::fs::create_dir_all(&workdir).unwrap();
        let session_path = dir.join("e2e.smh");
        let _ = std::fs::remove_file(&session_path);
        let cfg = Config::default_valid();
        let opened_id;
        {
            let opened = open_session(&session_path, cfg.clone()).unwrap();
            opened_id = opened.session.id();
            let mut ts = crate::tools::ToolSession::with_session(
                workdir.clone(),
                opened.session,
                opened.writer,
            );
            ts.invoke(
                "write",
                &ToolArguments::from_serializable(
                    &serde_json::json!({"path": "out/hello.txt", "content": "hi"}),
                )
                .unwrap(),
            )
            .unwrap();
            ts.invoke(
                "read",
                &ToolArguments::from_serializable(&serde_json::json!({"path": "out/hello.txt"}))
                    .unwrap(),
            )
            .unwrap();
        }
        let restored = SessionReader::open(&session_path)
            .unwrap()
            .selected_session(cfg.frame_bytes_max)
            .unwrap();
        assert_eq!(restored.id(), opened_id);
        assert_eq!(restored.active_branch().entries().len(), 6);
        // Tool call/result/completion triples stay adjacent and ordered.
        let kinds: Vec<&str> = restored
            .active_branch()
            .entries()
            .iter()
            .map(|e| match e.kind {
                EntryKind::ToolCall { .. } => "call",
                EntryKind::ToolResult { .. } => "result",
                _ => "other",
            })
            .collect();
        assert_eq!(
            kinds,
            vec!["call", "result", "other", "call", "result", "other"]
        );
        // The file effect actually happened under the invocation workdir.
        assert_eq!(
            std::fs::read_to_string(workdir.join("out/hello.txt")).unwrap(),
            "hi"
        );
        let _ = std::fs::remove_file(&session_path);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn reopened_session_keeps_its_identity_and_replays_identically() {
        let path = tmp_path("identity.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let created = {
            let mut opened = open_session(&path, cfg.clone()).unwrap();
            opened.writer.append_frame(&frame("one")).unwrap();
            opened.session
        };
        let first = SessionReader::open(&path).unwrap();
        let second = SessionReader::open(&path).unwrap();
        let first_header = first.header(cfg.frame_bytes_max).unwrap();
        let second_header = second.header(cfg.frame_bytes_max).unwrap();
        assert_eq!(first_header.session_id, created.id());
        assert_eq!(first_header.branch_id, created.active_branch().id());
        assert_eq!(second_header.session_id, first_header.session_id);

        // Replay evidence is identical across independent durable readers.
        let records = |reader: &SessionReader| {
            let mut records = Vec::new();
            crate::trace::replay(reader, cfg.frame_bytes_max, |record| {
                records.push(record);
                Ok(())
            })
            .unwrap();
            records
        };
        assert_eq!(records(&first), records(&second));
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn forking_publishes_a_new_session_file_and_leaves_the_source_intact() {
        let dir = tmp_path("fork_dir");
        std::fs::create_dir_all(&dir).unwrap();
        let source_path = dir.join("source.smh");
        let fork_path = dir.join("fork.smh");
        let cfg = Config::default_valid();

        let (source_session, boundary_id, leaf_id) = {
            let mut opened = open_session(&source_path, cfg.clone()).unwrap();
            let root = *opened.session.append(&message("root")).unwrap();
            opened
                .writer
                .append_entry(&root, opened.session.bytes(&root))
                .unwrap();
            let boundary = *opened
                .session
                .append(&EntryContent::Meta {
                    kind: COMPACTION_KIND.to_string(),
                    detail: StructuredData::from_serializable(
                        &serde_json::json!({"summary": "", "kept": 1}),
                    )
                    .unwrap(),
                })
                .unwrap();
            opened
                .writer
                .append_entry(&boundary, opened.session.bytes(&boundary))
                .unwrap();
            let tip = *opened.session.append(&message("after")).unwrap();
            opened
                .writer
                .append_entry(&tip, opened.session.bytes(&tip))
                .unwrap();
            assert_ne!(root.id, tip.id);
            (opened.session, boundary.id, tip.id)
        };

        let source_reader = SessionReader::open(&source_path).unwrap();
        let fork_id = {
            let forked = source_reader
                .fork_at(boundary_id, &fork_path, cfg.clone())
                .unwrap();
            assert_ne!(forked.session.id(), source_session.id());
            assert_eq!(forked.session.selected_id(), Some(boundary_id));
            assert_eq!(forked.session.compaction_boundary(), Some(boundary_id));
            assert!(forked.recovery.damage.is_empty());
            forked.session.id()
        };

        // The fork is durable: reopening restores the same identity and leaf.
        {
            let reopened = open_session(&fork_path, cfg.clone()).unwrap();
            assert_eq!(reopened.session.id(), fork_id);
            assert_eq!(reopened.session.selected_id(), Some(boundary_id));
            assert_eq!(reopened.session.active_branch().entries().len(), 1);
            assert_eq!(reopened.session.compaction_boundary(), Some(boundary_id));
            // Full historical chain remains on disk, including pre-compaction content.
            let mut history = Vec::new();
            SessionReader::open(&fork_path)
                .unwrap()
                .replay_at(boundary_id, cfg.frame_bytes_max, |entry| {
                    history.push(entry);
                    Ok(())
                })
                .unwrap();
            assert_eq!(history.len(), 2);
            assert_eq!(
                history[0].content,
                source_session.bytes(&source_session.active_branch().entries()[0])
            );
        }

        // The source keeps its own identity, tip, and entry count.
        {
            let source_again = open_session(&source_path, cfg.clone()).unwrap();
            assert_eq!(source_again.session.id(), source_session.id());
            assert_eq!(source_again.session.selected_id(), Some(leaf_id));
            assert_eq!(source_again.session.active_branch().entries().len(), 2);
            let mut count = 0;
            SessionReader::open(&source_path)
                .unwrap()
                .replay_at(leaf_id, cfg.frame_bytes_max, |_| {
                    count += 1;
                    Ok(())
                })
                .unwrap();
            assert_eq!(count, 3);
        }

        // A populated target is refused instead of overwritten.
        let error = source_reader
            .fork_at(boundary_id, &fork_path, cfg)
            .unwrap_err();
        assert_eq!(error.code(), "SESSION_EXISTS");
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn legacy_session_without_prologue_is_rejected_without_repair() {
        let path = tmp_path("legacy-no-prologue.smh");
        let bytes = frame("legacy").encode(1 << 20).unwrap();
        std::fs::write(&path, &bytes).unwrap();
        assert_eq!(
            open_session(&path, Config::default_valid())
                .unwrap_err()
                .code(),
            "SESSION_PROLOGUE"
        );
        assert_eq!(std::fs::read(&path).unwrap(), bytes);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn session_file_without_header_is_rejected() {
        let path = tmp_path("headerless.smh");
        let cfg = Config::default_valid();
        std::fs::write(
            &path,
            file_bytes(frame("orphan").encode(cfg.frame_bytes_max).unwrap()),
        )
        .unwrap();
        let error = open_session(&path, cfg).unwrap_err();
        assert_eq!(error.code(), "SESSION_HEADER_MISSING");
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn cancellation_stops_replay_at_entry_boundary_and_leaves_no_fork() {
        let path = tmp_path("cancel-replay.smh");
        let fork = tmp_path("cancel-replay-fork.smh");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&fork);
        let config = Config::default_valid();
        let mut opened = open_session(&path, config.clone()).unwrap();
        let root = entry("root");
        let child = EntryFrame::new(Some(root.id), &message("child")).unwrap();
        for entry in [&root, &child] {
            opened
                .writer
                .append_frame(&Frame::Known {
                    entry: entry.clone(),
                })
                .unwrap();
        }
        drop(opened);
        let reader = SessionReader::open(&path).unwrap();
        let cancel = CancelHandle::new();
        let mut visited = 0;
        let error = reader
            .replay_at_cancelled(child.id, config.frame_bytes_max, &cancel, |_| {
                visited += 1;
                cancel.cancel();
                Ok(())
            })
            .unwrap_err();
        assert_eq!(error.code(), "CANCELLED");
        assert_eq!(visited, 1);
        assert_eq!(
            reader
                .fork_at_cancelled(child.id, &fork, config, &cancel)
                .unwrap_err()
                .code(),
            "CANCELLED"
        );
        assert!(!fork.exists());
        drop(reader);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn reader_drop_releases_an_inherited_snapshot_lock() {
        let path = tmp_path("reader-drop-unlocks-inherited-lock.smh");
        let _ = std::fs::remove_file(&path);
        let opened = open_session(&path, Config::default_valid()).unwrap();
        drop(opened);
        let reader = SessionReader::open(&path).unwrap();
        // A fork inherits this open-file description.  Closing only this process's
        // descriptor must still release the maintenance lock before the child execs.
        let inherited_readers = reader.readers.try_clone().unwrap();
        drop(reader);
        let coordination = maintenance_lock(&path).unwrap();
        drop(coordination);
        drop(inherited_readers);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn checkpoint_tail_snapshots_reopen_to_the_last_authoritative_entry() {
        let path = tmp_path("checkpoint-tail-snapshots.smh");
        let _ = std::fs::remove_file(&path);
        let cfg = Config::default_valid();
        let (before_checkpoint, selected, boundary, ancestry) = {
            let mut opened = open_session(&path, cfg.clone()).unwrap();
            let root = *opened.session.append(&message("root")).unwrap();
            opened
                .writer
                .append_entry(&root, opened.session.bytes(&root))
                .unwrap();
            let boundary = *opened
                .session
                .append(&EntryContent::Meta {
                    kind: COMPACTION_KIND.to_string(),
                    detail: StructuredData::from_serializable(
                        &serde_json::json!({"summary": "", "kept": 1}),
                    )
                    .unwrap(),
                })
                .unwrap();
            opened
                .writer
                .append_entry(&boundary, opened.session.bytes(&boundary))
                .unwrap();
            let mut ancestry = vec![root.id, boundary.id];
            for _ in 0..61 {
                let record = *opened.session.append(&message("turn")).unwrap();
                opened
                    .writer
                    .append_entry(&record, opened.session.bytes(&record))
                    .unwrap();
                ancestry.push(record.id);
            }
            let record = *opened.session.append(&message("selected")).unwrap();
            opened
                .writer
                .append_entry(&record, opened.session.bytes(&record))
                .unwrap();
            ancestry.push(record.id);
            (
                std::fs::read(&path).unwrap(),
                record.id,
                boundary.id,
                ancestry,
            )
        };
        let checkpoint = Frame::Checkpoint {
            id: selected,
            offset: 0,
        }
        .encode(cfg.frame_bytes_max)
        .unwrap();
        let checkpoint_at = before_checkpoint.len() - checkpoint.len();
        for end in checkpoint_at..before_checkpoint.len() {
            std::fs::write(&path, &before_checkpoint[..end]).unwrap();
            if end > checkpoint_at {
                assert_eq!(
                    SessionReader::open(&path)
                        .unwrap()
                        .scan(cfg.frame_bytes_max, |_| Ok(()))
                        .unwrap()
                        .torn_tail_at,
                    Some(u64::try_from(checkpoint_at).unwrap())
                );
            }
            let reopened = open_session(&path, cfg.clone()).unwrap();
            assert_eq!(reopened.session.selected_id(), Some(selected));
            assert_eq!(reopened.session.compaction_boundary(), Some(boundary));
            drop(reopened);
            let mut restored_ancestry = Vec::new();
            SessionReader::open(&path)
                .unwrap()
                .replay_at(selected, cfg.frame_bytes_max, |entry| {
                    restored_ancestry.push(entry.id);
                    Ok(())
                })
                .unwrap();
            assert_eq!(restored_ancestry, ancestry);
            assert_eq!(
                SessionReader::open(&path)
                    .unwrap()
                    .scan(cfg.frame_bytes_max, |_| Ok(()))
                    .unwrap()
                    .torn_tail_at,
                None
            );
        }
        let _ = std::fs::remove_file(path);
    }

    fn publication_fixture(path: &Path, cfg: &Config) -> (Vec<u8>, [EntryId; 3]) {
        let mut opened = open_session(path, cfg.clone()).unwrap();
        let root = *opened.session.append(&message("root")).unwrap();
        opened
            .writer
            .append_entry(&root, opened.session.bytes(&root))
            .unwrap();
        let boundary = *opened
            .session
            .append(&EntryContent::Meta {
                kind: COMPACTION_KIND.to_string(),
                detail: StructuredData::from_serializable(
                    &serde_json::json!({"summary": "", "kept": 1}),
                )
                .unwrap(),
            })
            .unwrap();
        opened
            .writer
            .append_entry(&boundary, opened.session.bytes(&boundary))
            .unwrap();
        let leaf = *opened.session.append(&message("leaf")).unwrap();
        opened
            .writer
            .append_entry(&leaf, opened.session.bytes(&leaf))
            .unwrap();
        opened
            .writer
            .append_frame(&Frame::Unknown {
                kind: 99,
                version: 1,
                payload: b"preserve me".to_vec(),
            })
            .unwrap();
        drop(opened);
        (
            std::fs::read(path).unwrap(),
            [root.id, boundary.id, leaf.id],
        )
    }

    fn assert_complete_publication(path: &Path, cfg: &Config, ids: [EntryId; 3], unknown: bool) {
        let reader = SessionReader::open(path).unwrap();
        reader.verify_full(cfg.frame_bytes_max).unwrap();
        let session = reader.selected_session(cfg.frame_bytes_max).unwrap();
        assert_eq!(session.selected_id(), Some(ids[2]));
        assert_eq!(session.compaction_boundary(), Some(ids[1]));
        let mut ancestry = Vec::new();
        reader
            .replay_at(ids[2], cfg.frame_bytes_max, |entry| {
                ancestry.push(entry.id);
                Ok(())
            })
            .unwrap();
        assert_eq!(ancestry, ids);
        assert_eq!(
            scanned_frames(path, cfg.frame_bytes_max)
                .frames
                .iter()
                .any(|frame| matches!(frame, Frame::Unknown { kind: 99, .. })),
            unknown
        );
    }

    fn publication_temps(dir: &Path, operation: &str) -> Vec<PathBuf> {
        std::fs::read_dir(dir)
            .unwrap()
            .map(|entry| entry.unwrap().path())
            .filter(|path| {
                path.file_name()
                    .and_then(|name| name.to_str())
                    .is_some_and(|name| name.contains(&format!(".smith-{operation}-")))
            })
            .collect()
    }

    #[test]
    fn fork_published_checkpoint_reopens_complete_history() {
        let dir = tmp_path("fork-hard-link-interval");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let source = dir.join("source.smh");
        let target = dir.join("fork.smh");
        let cfg = Config::default_valid();
        let (_, ids) = publication_fixture(&source, &cfg);
        let observed = std::rc::Rc::new(std::cell::Cell::new(false));
        let observed_at_hook = std::rc::Rc::clone(&observed);
        let target_at_hook = target.clone();
        set_publication_hook(move |point| {
            if point == PublicationCheckpoint::Published {
                observed_at_hook.set(true);
                assert!(SessionReader::open(&target_at_hook).is_ok());
            }
        });
        SessionReader::open(&source)
            .unwrap()
            .fork_at(ids[2], &target, cfg)
            .unwrap();
        clear_publication_hook();
        assert!(observed.get());
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn fork_and_repair_publication_interruptions_never_expose_partial_sessions() {
        use std::panic::{AssertUnwindSafe, catch_unwind, panic_any};

        let cfg = Config::default_valid();
        let points = [
            PublicationCheckpoint::TempCreated,
            PublicationCheckpoint::PartialWrite,
            PublicationCheckpoint::CompletedWrite,
            PublicationCheckpoint::TempSynced,
            PublicationCheckpoint::Published,
            PublicationCheckpoint::ParentSynced,
            PublicationCheckpoint::Cleanup,
        ];
        for point in points {
            let dir = tmp_path(&format!("publication-interrupt-{point:?}"));
            let _ = std::fs::remove_dir_all(&dir);
            std::fs::create_dir_all(&dir).unwrap();
            let source = dir.join("source.smh");
            let target = dir.join("fork.smh");
            let (before, ids) = publication_fixture(&source, &cfg);
            let reader = SessionReader::open(&source).unwrap();
            set_publication_hook(move |actual| {
                if actual == point {
                    // Panic simulates abrupt control-flow loss: fork's error cleanup is not run.
                    panic_any(actual);
                }
            });
            assert!(
                catch_unwind(AssertUnwindSafe(|| reader.fork_at(
                    ids[2],
                    &target,
                    cfg.clone()
                )))
                .is_err()
            );
            clear_publication_hook();
            assert_eq!(std::fs::read(&source).unwrap(), before);
            assert_complete_publication(&source, &cfg, ids, true);
            if matches!(
                point,
                PublicationCheckpoint::TempCreated
                    | PublicationCheckpoint::PartialWrite
                    | PublicationCheckpoint::CompletedWrite
                    | PublicationCheckpoint::TempSynced
            ) {
                assert!(!target.exists());
                assert!(SessionReader::open(&target).is_err());
                assert_eq!(publication_temps(&dir, "fork").len(), 1);
            } else {
                assert_complete_publication(&target, &cfg, ids, false);
            }
            drop(reader);
            let _ = std::fs::remove_dir_all(dir);
        }

        for point in points {
            let dir = tmp_path(&format!("repair-publication-interrupt-{point:?}"));
            let _ = std::fs::remove_dir_all(&dir);
            std::fs::create_dir_all(&dir).unwrap();
            let source = dir.join("source.smh");
            let (before, ids) = publication_fixture(&source, &cfg);
            set_publication_hook(move |actual| {
                if actual == point {
                    // Panic simulates abrupt control-flow loss: repair's error cleanup is not run.
                    panic_any(actual);
                }
            });
            assert!(
                catch_unwind(AssertUnwindSafe(|| repair_session(
                    &source,
                    cfg.frame_bytes_max
                )))
                .is_err()
            );
            clear_publication_hook();
            if matches!(
                point,
                PublicationCheckpoint::TempCreated
                    | PublicationCheckpoint::PartialWrite
                    | PublicationCheckpoint::CompletedWrite
                    | PublicationCheckpoint::TempSynced
            ) {
                assert_eq!(std::fs::read(&source).unwrap(), before);
                assert_complete_publication(&source, &cfg, ids, true);
                assert_eq!(publication_temps(&dir, "repair").len(), 1);
            } else {
                assert_complete_publication(&source, &cfg, ids, true);
                assert!(publication_temps(&dir, "repair").is_empty());
            }
            let _ = std::fs::remove_dir_all(dir);
        }
    }

    #[test]
    fn fork_publish_collision_preserves_target_and_temporary_source() {
        let dir = tmp_path("fork-publish-failure");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let source = dir.join("source.smh");
        let target = dir.join("fork.smh");
        let cfg = Config::default_valid();
        let (before, ids) = publication_fixture(&source, &cfg);
        let occupied = target.clone();
        set_publication_hook(move |point| {
            if point == PublicationCheckpoint::TempSynced {
                std::fs::write(&occupied, b"occupied during publish").unwrap();
            }
        });
        let error = SessionReader::open(&source)
            .unwrap()
            .fork_at(ids[2], &target, cfg)
            .unwrap_err();
        clear_publication_hook();
        match error {
            SmithError::Storage {
                code,
                operation,
                path,
                message,
                publication,
            } => {
                assert_eq!(code, smith::error::StorageCode::CreatePublish);
                assert_eq!(publication, None);
                assert_eq!(operation.as_str(), "publish");
                assert_eq!(path, target);
                assert!(!message.is_empty());
            }
            other => panic!("expected publication storage error, got {other:?}"),
        }
        assert_eq!(std::fs::read(&source).unwrap(), before);
        assert_eq!(std::fs::read(&target).unwrap(), b"occupied during publish");
        assert_eq!(publication_temps(&dir, "fork").len(), 1);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn concurrent_publishers_choose_exactly_one_winner_without_overwrite() {
        use std::sync::{Arc, Barrier};

        let dir = tmp_path("concurrent-no-clobber-publication");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let target = dir.join("target.smh");
        let barrier = Arc::new(Barrier::new(4));
        let mut workers = Vec::new();
        for index in 0..4 {
            let temp = dir.join(format!("temp-{index}.smh"));
            let bytes = format!("publisher-{index}").into_bytes();
            let mut file = File::create(&temp).unwrap();
            file.write_all(&bytes).unwrap();
            file.sync_all().unwrap();
            let target = target.clone();
            let barrier = Arc::clone(&barrier);
            workers.push(std::thread::spawn(move || {
                barrier.wait();
                (bytes, temp.clone(), publish_no_clobber(&temp, &target))
            }));
        }
        let results: Vec<_> = workers
            .into_iter()
            .map(|worker| worker.join().unwrap())
            .collect();
        let winners: Vec<_> = results
            .iter()
            .filter(|(_, _, result)| result.is_ok())
            .collect();
        assert_eq!(winners.len(), 1);
        assert_eq!(std::fs::read(&target).unwrap(), winners[0].0);
        for (_, temp, result) in results {
            assert_eq!(temp.exists(), result.is_err());
        }
        let _ = std::fs::remove_dir_all(dir);
    }

    #[cfg(unix)]
    #[test]
    fn no_clobber_publication_never_leaves_an_alias() {
        use std::os::unix::fs::MetadataExt;

        let dir = tmp_path("no-clobber-publication-nlink");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let temp = dir.join("temp.smh");
        let target = dir.join("target.smh");
        std::fs::write(&temp, b"new").unwrap();
        publish_no_clobber(&temp, &target).unwrap();
        assert!(!temp.exists());
        assert_eq!(std::fs::metadata(&target).unwrap().nlink(), 1);

        std::fs::write(&temp, b"retry").unwrap();
        let before = std::fs::read(&target).unwrap();
        assert!(publish_no_clobber(&temp, &target).is_err());
        assert_eq!(std::fs::read(&target).unwrap(), before);
        assert_eq!(std::fs::metadata(&temp).unwrap().nlink(), 1);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn parent_sync_failure_reports_published_durability_uncertain_on_wire() {
        let dir = tmp_path("fork-parent-sync-failure");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let source = dir.join("source.smh");
        let target = dir.join("fork.smh");
        let cfg = Config::default_valid();
        let (before, ids) = publication_fixture(&source, &cfg);
        set_parent_sync_failure(true);
        let error = SessionReader::open(&source)
            .unwrap()
            .fork_at(ids[2], &target, cfg.clone())
            .unwrap_err();
        set_parent_sync_failure(false);
        assert_eq!(std::fs::read(&source).unwrap(), before);
        assert_complete_publication(&target, &cfg, ids, false);
        assert_eq!(
            serde_json::to_string(&error).unwrap(),
            format!(
                r#"{{"Storage":{{"code":"STORE_DIR_SYNC","operation":"sync parent directory after publication","path":"{}","message":"injected parent directory sync failure","publication":"published-durability-uncertain"}}}}"#,
                dir.display()
            )
        );
        match error {
            SmithError::Storage { publication, .. } => assert_eq!(
                publication,
                Some(smith::error::PublicationState::PublishedDurabilityUncertain)
            ),
            other => panic!("expected published storage error, got {other:?}"),
        }
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn temporary_name_exhaustion_reports_final_attempt_without_mutation_or_leak() {
        let dir = tmp_path("temporary-name-exhaustion");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("session.smh");
        let id = SessionId::from_u128(7);
        for attempt in 0..128 {
            std::fs::create_dir(
                path.with_file_name(format!("session.smh.smith-create-{id}-{attempt}.tmp")),
            )
            .unwrap();
        }
        let before = std::fs::read_dir(&dir).unwrap().count();
        let error = new_temp_file(&path, "create", id).unwrap_err();
        match error {
            SmithError::Storage {
                code,
                operation,
                path: attempted,
                message,
                ..
            } => {
                assert_eq!(code, smith::error::StorageCode::CreateTmp);
                assert_eq!(operation.as_str(), "create temporary file");
                assert_eq!(
                    attempted,
                    path.with_file_name(format!("session.smh.smith-create-{id}-127.tmp"))
                );
                assert!(!message.is_empty());
            }
            other => panic!("expected storage error, got {other:?}"),
        }
        assert!(!path.exists());
        assert_eq!(std::fs::read_dir(&dir).unwrap().count(), before);
        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn header_len_matches_protocol() {
        assert_eq!(crate::frame::HEADER_LEN, 12);
        assert_eq!(crate::frame::BODY_PREFIX_LEN, 8);
    }
}