//! P12 experimental TurkFS-v1 recovery extension for disposable disks only.
//! Two fixed single-sector files (inodes 2/4), a K395-style allocation bitmap
//! at LBA26, an extension marker at LBA27 and two redo slots at LBA28/29.
//! No journal recycling, general writer, torn-sector/physical power-loss
//! model or authenticated storage; production-write stays blocked.
use crate::{
    journal::{self, Journal, Pointer, Record, Replay},
    turkfs, virtio_ring,
};

pub const ROOT_LBA: u64 = 17;
pub const INODE_LBA: u64 = 18;
pub const BITMAP_LBA: u64 = 26;
pub const CONTROL_LBA: u64 = 27;
pub const SLOTS: [u64; 2] = [28, 29];
pub const INODES: [u32; 2] = [2, 4];
const NAMES: [&[u8]; 2] = [b"KERNEL", b"TAIL"];
/// First LBA of the experiment's allocation pool; lower blocks are fixture-owned.
pub const POOL_START: u64 = 30;
pub const MARKER: &[u8; 8] = b"P12FS002";
pub const WRITER_STEPS: u32 = 20;
pub const REPLAY_STEPS: u32 = 4;
pub type Block = [u8; 512];

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Error {
    /// Device/transport failure, including a backend ENOSPC reported as IOERR.
    Io,
    Format,
    Scope,
    Journal(journal::Error),
    Conflict,
    Full,
    NoSpace,
    ReplayRequired,
    DeviceMode,
    Parameters,
}
impl From<journal::Error> for Error {
    fn from(e: journal::Error) -> Self {
        match e {
            journal::Error::Conflict => Self::Conflict,
            other => Self::Journal(other),
        }
    }
}
impl Error {
    pub fn label(self) -> &'static str {
        match self {
            Self::Io => "EIO",
            Self::Format => "FS-FORMAT",
            Self::Scope => "EXPERIMENT-SCOPE",
            Self::Journal(_) => "JOURNAL-INTEGRITY",
            Self::Conflict => "JOURNAL-CONFLICT",
            Self::Full => "JOURNAL-FULL",
            Self::NoSpace => "ENOSPC",
            Self::ReplayRequired => "REPLAY-REQUIRED",
            Self::DeviceMode => "DEVICE-MODE",
            Self::Parameters => "PARAMETERS",
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Step {
    DataWrite,
    DataFlush,
    RecordWrite,
    RecordFlush,
    CommitWrite,
    CommitFlush,
    InodeWrite,
    InodeFlush,
    BitmapWrite,
    BitmapFlush,
    ReplayInodeWrite,
    ReplayInodeFlush,
    ReplayBitmapWrite,
    ReplayBitmapFlush,
}
impl Step {
    pub fn label(self) -> &'static str {
        match self {
            Self::DataWrite => "DATA-WRITE",
            Self::DataFlush => "DATA-FLUSH",
            Self::RecordWrite => "RECORD-WRITE",
            Self::RecordFlush => "RECORD-FLUSH",
            Self::CommitWrite => "COMMIT-WRITE",
            Self::CommitFlush => "COMMIT-FLUSH",
            Self::InodeWrite => "INODE-WRITE",
            Self::InodeFlush => "INODE-FLUSH",
            Self::BitmapWrite => "BITMAP-WRITE",
            Self::BitmapFlush => "BITMAP-FLUSH",
            Self::ReplayInodeWrite => "REPLAY-INODE-WRITE",
            Self::ReplayInodeFlush => "REPLAY-INODE-FLUSH",
            Self::ReplayBitmapWrite => "REPLAY-BITMAP-WRITE",
            Self::ReplayBitmapFlush => "REPLAY-BITMAP-FLUSH",
        }
    }
}

pub trait BlockIo {
    fn read(&mut self, lba: u64, out: &mut Block) -> Result<(), Error>;
    fn write(&mut self, lba: u64, bytes: &Block) -> Result<(), Error>;
    fn flush(&mut self) -> Result<(), Error>;
    /// Called only after the device completed the preceding request.
    fn checkpoint(&mut self, step: u32, label: Step);
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Mode {
    Write,
    Replay,
    Read,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Params {
    pub mode: Mode,
    /// Zero means run to completion; otherwise park after this checkpoint.
    pub pause: u32,
}

/// Host-controlled per-boot parameters: exactly `turkos.p12=<mode>` with an
/// optional `,pause=<1..=20 writer | 1..=4 replay>`. Anything else rejects.
pub fn parse_params(args: &[u8]) -> Result<Params, Error> {
    let rest = args.strip_prefix(b"turkos.p12=").ok_or(Error::Parameters)?;
    let mut parts = rest.split(|b| *b == b',');
    let mode = match parts.next() {
        Some(b"write") => Mode::Write,
        Some(b"replay") => Mode::Replay,
        Some(b"read") => Mode::Read,
        _ => return Err(Error::Parameters),
    };
    let mut pause = 0u32;
    if let Some(part) = parts.next() {
        let digits = part.strip_prefix(b"pause=").ok_or(Error::Parameters)?;
        if digits.is_empty() || digits.len() > 2 || digits[0] == b'0' {
            return Err(Error::Parameters);
        }
        for digit in digits {
            if !digit.is_ascii_digit() {
                return Err(Error::Parameters);
            }
            pause = pause * 10 + u32::from(digit - b'0');
        }
        let limit = match mode {
            Mode::Write => WRITER_STEPS,
            Mode::Replay => REPLAY_STEPS,
            Mode::Read => 0,
        };
        if pause > limit {
            return Err(Error::Parameters);
        }
    }
    if parts.next().is_some() {
        return Err(Error::Parameters);
    }
    Ok(Params { mode, pause })
}

/// FDT v17 `/chosen/bootargs` (QEMU `-append`). One property, NUL-terminated.
pub fn chosen_bootargs(blob: &[u8]) -> Result<&[u8], Error> {
    let be = |bytes: &[u8], at: usize| -> Result<usize, Error> {
        let field = bytes
            .get(at..at.checked_add(4).ok_or(Error::Parameters)?)
            .ok_or(Error::Parameters)?;
        Ok(u32::from_be_bytes(field.try_into().unwrap()) as usize)
    };
    let span = |start: usize, len: usize| {
        blob.get(start..start.checked_add(len).ok_or(Error::Parameters)?)
            .ok_or(Error::Parameters)
    };
    if be(blob, 0)? != 0xd00d_feed || be(blob, 4)? > blob.len() || be(blob, 20)? < 17 {
        return Err(Error::Parameters);
    }
    let structure = span(be(blob, 8)?, be(blob, 36)?)?;
    let strings = span(be(blob, 12)?, be(blob, 32)?)?;
    let nul = |bytes: &[u8]| -> Result<usize, Error> {
        bytes.iter().position(|b| *b == 0).ok_or(Error::Parameters)
    };
    let align = |n: usize| n.checked_add(3).map(|n| n & !3).ok_or(Error::Parameters);
    let (mut cursor, mut depth, mut chosen, mut found) = (0usize, 0usize, false, None);
    loop {
        let token = be(structure, cursor)?;
        cursor += 4;
        match token {
            1 => {
                let rest = structure.get(cursor..).ok_or(Error::Parameters)?;
                let len = nul(rest)?;
                depth += 1;
                if depth == 2 {
                    chosen = &rest[..len] == b"chosen";
                }
                cursor = align(cursor + len + 1)?;
            }
            2 => {
                if depth == 0 {
                    return Err(Error::Parameters);
                }
                if depth == 2 {
                    chosen = false;
                }
                depth -= 1;
            }
            3 => {
                let len = be(structure, cursor)?;
                let name = strings
                    .get(be(structure, cursor + 4)?..)
                    .ok_or(Error::Parameters)?;
                let name = &name[..nul(name)?];
                cursor += 8;
                let value = structure
                    .get(cursor..cursor.checked_add(len).ok_or(Error::Parameters)?)
                    .ok_or(Error::Parameters)?;
                cursor = align(cursor + len)?;
                if depth == 2 && chosen && name == b"bootargs" && found.replace(value).is_some() {
                    return Err(Error::Parameters);
                }
            }
            4 => {}
            9 if depth == 0 => break,
            _ => return Err(Error::Parameters),
        }
    }
    let value = found.ok_or(Error::Parameters)?;
    let value = value.strip_suffix(&[0]).ok_or(Error::Parameters)?;
    if value.contains(&0) {
        return Err(Error::Parameters);
    }
    Ok(value)
}

fn u32_at(b: &Block, at: usize) -> u32 {
    u32::from_le_bytes(b[at..at + 4].try_into().unwrap())
}
fn u64_at(b: &Block, at: usize) -> u64 {
    u64::from_le_bytes(b[at..at + 8].try_into().unwrap())
}
fn inode_offset(inode: u32) -> usize {
    (inode as usize - 1) * 32
}
pub fn allocated(bitmap: &Block, lba: u64) -> bool {
    bitmap[(lba / 8) as usize] & (1 << (lba % 8)) != 0
}
fn set_allocated(bitmap: &mut Block, lba: u64, value: bool) {
    let bit = 1u8 << (lba % 8);
    if value {
        bitmap[(lba / 8) as usize] |= bit;
    } else {
        bitmap[(lba / 8) as usize] &= !bit;
    }
}
fn fixed_metadata(lba: u64) -> bool {
    lba <= INODE_LBA || lba == BITMAP_LBA || lba == CONTROL_LBA || SLOTS.contains(&lba)
}
fn set_pointer(table: &mut Block, index: usize, lba: u64, generation: u64) {
    let at = inode_offset(INODES[index]);
    table[at + 16..at + 24].copy_from_slice(&lba.to_le_bytes());
    table[at + 24..at + 32].copy_from_slice(&generation.to_le_bytes());
}
pub fn pattern(index: usize) -> Block {
    core::array::from_fn(|i| (i as u8) ^ if index == 0 { 0xa6 } else { 0x39 })
}

/// Per-block owner classification of the CURRENT on-disk image. An allocated
/// block nobody owns and no committed record explains is `unknown`: it is
/// reported and never freed or reused, even under ENOSPC.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct Ownership {
    pub allocated: u32,
    pub owned: u32,
    /// Allocated, unreferenced old block of a committed record (replay frees).
    pub released: u32,
    /// New block of a committed record that is referenced or allocated but not both.
    pub claimed: u32,
    pub unknown: u32,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Plan {
    pub empty: u32,
    pub incomplete: u32,
    pub committed: u32,
    pub apply: u32,
    pub already: u32,
    pub bitmap_changes: u32,
    pub ownership: Ownership,
    /// Metadata after replay; equal to the current blocks when nothing is due.
    pub inode: Block,
    pub bitmap: Block,
    /// Allocated blocks of unknown ownership in the current image.
    pub unknown: Block,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplayOutcome {
    pub applied: u32,
    pub already: u32,
    pub writes: u32,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FileState {
    pub inode: u32,
    pub lba: u64,
    pub generation: u64,
    pub sha256: [u8; 32],
}

pub struct Mount {
    block_count: u64,
    geometry: turkfs::BlockGeometry,
    root: turkfs::RootDirectory,
    inode: Block,
    bitmap: Block,
    slots: [Block; 2],
}

impl Mount {
    /// Reads every metadata block from the device; nothing is cached across
    /// mounts, so a fresh `open` observes the disk rather than prior state.
    pub fn open(io: &mut impl BlockIo, capacity_sectors: u64) -> Result<Self, Error> {
        let mut block = [0; 512];
        io.read(turkfs::SUPERBLOCK_LBA, &mut block)?;
        turkfs::require_superblock_signature(&block).map_err(|_| Error::Format)?;
        turkfs::require_format_version(&block).map_err(|_| Error::Format)?;
        let bytes = capacity_sectors.checked_mul(512).ok_or(Error::Format)?;
        let geometry = turkfs::require_block_geometry(&block, bytes).map_err(|_| Error::Format)?;
        let (root_lba, count) =
            turkfs::root_directory_location(&block, geometry).map_err(|_| Error::Format)?;
        let inode_lba =
            turkfs::inode_table_location(&block, geometry, root_lba).map_err(|_| Error::Format)?;
        if root_lba != ROOT_LBA
            || inode_lba != INODE_LBA
            || count != 4
            || geometry.block_count < POOL_START + 2
            || geometry.block_count > 4096
        {
            return Err(Error::Scope);
        }
        io.read(ROOT_LBA, &mut block)?;
        let root =
            turkfs::parse_root_directory(ROOT_LBA, count, &block).map_err(|_| Error::Format)?;
        let mut inode = [0; 512];
        io.read(INODE_LBA, &mut inode)?;
        let mut bitmap = [0; 512];
        io.read(BITMAP_LBA, &mut bitmap)?;
        io.read(CONTROL_LBA, &mut block)?;
        if &block[..8] != MARKER
            || block[8..12] != 2u32.to_le_bytes()
            || block[12..].iter().any(|b| *b != 0)
        {
            return Err(Error::Scope);
        }
        let mut slots = [[0; 512]; 2];
        for (slot, lba) in slots.iter_mut().zip(SLOTS) {
            io.read(lba, slot)?;
        }
        let mount = Self {
            block_count: geometry.block_count,
            geometry,
            root,
            inode,
            bitmap,
            slots,
        };
        if (mount.block_count..4096).any(|lba| allocated(&mount.bitmap, lba)) {
            return Err(Error::Format);
        }
        for index in 0..2 {
            mount.file(&mount.inode, index)?;
        }
        Ok(mount)
    }

    fn file(&self, table: &Block, index: usize) -> Result<(u64, u64), Error> {
        if self.root.find(NAMES[index]).map(|e| e.inode) != Some(INODES[index]) {
            return Err(Error::Scope);
        }
        let file = turkfs::bounded_single_block_file(
            &self.root,
            table,
            NAMES[index],
            self.geometry,
            INODE_LBA,
        )
        .map_err(|_| Error::Format)?;
        let generation = u64_at(table, inode_offset(INODES[index]) + 24);
        if file.size != 512 || generation == 0 || fixed_metadata(file.lba) {
            return Err(Error::Scope);
        }
        Ok((file.lba, generation))
    }

    /// Number of inode extents and fixed metadata roles naming `lba`.
    fn owners(&self, table: &Block, lba: u64) -> Result<u32, Error> {
        let mut owners = fixed_metadata(lba) as u32;
        for entry in &self.root.entries[..self.root.count] {
            let at = inode_offset(entry.inode);
            if u32_at(table, at) != entry.inode || u32_at(table, at + 4) != 0 {
                return Err(Error::Format);
            }
            let size = u64_at(table, at + 8);
            if size == 0 {
                continue;
            }
            if size > 1024 {
                return Err(Error::Scope);
            }
            let start = u64_at(table, at + 16);
            let end = start.checked_add(size.div_ceil(512)).ok_or(Error::Format)?;
            if start <= INODE_LBA || end > self.block_count {
                return Err(Error::Format);
            }
            owners += (start..end).contains(&lba) as u32;
        }
        Ok(owners)
    }

    /// Validates the whole journal and computes the full redo target before
    /// any caller may write; a bad second record cannot leave a partial replay.
    pub fn plan(&self, io: &mut impl BlockIo) -> Result<Plan, Error> {
        let mut journal = Journal::<2>::new();
        let (mut empty, mut incomplete) = (0, 0);
        for raw in &self.slots {
            if raw.iter().all(|b| *b == 0) {
                empty += 1;
                continue;
            }
            match Record::decode(raw) {
                Ok(record) => journal.push(record)?,
                Err(journal::Error::Uncommitted) => incomplete += 1,
                Err(e) => return Err(e.into()),
            }
        }
        let mut inode = self.inode;
        let (mut apply, mut already) = (0, 0);
        for record in journal.records() {
            let index = INODES
                .iter()
                .position(|i| *i == record.inode)
                .ok_or(Error::Scope)?;
            for block in [record.old_block, record.new_block] {
                if fixed_metadata(block) || block >= self.block_count {
                    return Err(Error::Scope);
                }
            }
            let (lba, generation) = self.file(&self.inode, index)?;
            let mut data = [0; 512];
            io.read(record.new_block, &mut data)?;
            let current = Pointer {
                generation,
                block: lba,
            };
            match record.replay(current, virtio_ring::sector_sha256(&data))? {
                Replay::Apply(next) => {
                    set_pointer(&mut inode, index, next.block, next.generation);
                    apply += 1;
                }
                Replay::AlreadyApplied => already += 1,
            }
        }
        let mut bitmap = self.bitmap;
        for record in journal.records() {
            set_allocated(&mut bitmap, record.new_block, true);
            if self.owners(&inode, record.old_block)? != 0 {
                return Err(Error::Conflict);
            }
            set_allocated(&mut bitmap, record.old_block, false);
        }
        let mut ownership = Ownership::default();
        let mut unknown = [0; 512];
        for lba in 0..self.block_count {
            let current = self.owners(&self.inode, lba)?;
            let target = self.owners(&inode, lba)?;
            // Target state: one owner per block, every owned block allocated.
            if target > 1 || (target == 1 && !allocated(&bitmap, lba)) {
                return Err(Error::Format);
            }
            let named_old = journal.records().any(|r| r.old_block == lba);
            let named_new = journal.records().any(|r| r.new_block == lba);
            let is_allocated = allocated(&self.bitmap, lba);
            ownership.allocated += is_allocated as u32;
            match (is_allocated, current) {
                (_, 2..) => return Err(Error::Format),
                (true, 1) => ownership.owned += 1,
                (true, 0) if named_old => ownership.released += 1,
                (true, 0) if named_new => ownership.claimed += 1,
                (true, 0) => {
                    ownership.unknown += 1;
                    set_allocated(&mut unknown, lba, true);
                }
                (false, 1) if named_new => ownership.claimed += 1,
                (false, 1) => return Err(Error::Format),
                (false, _) => {}
            }
        }
        let bitmap_changes = (0..self.block_count)
            .filter(|lba| allocated(&bitmap, *lba) != allocated(&self.bitmap, *lba))
            .count() as u32;
        Ok(Plan {
            empty,
            incomplete,
            committed: journal.len() as u32,
            apply,
            already,
            bitmap_changes,
            ownership,
            inode,
            bitmap,
            unknown,
        })
    }

    pub fn files(&self, io: &mut impl BlockIo) -> Result<[FileState; 2], Error> {
        let mut files = [FileState {
            inode: 0,
            lba: 0,
            generation: 0,
            sha256: [0; 32],
        }; 2];
        for (index, file) in files.iter_mut().enumerate() {
            let (lba, generation) = self.file(&self.inode, index)?;
            let mut data = [0; 512];
            io.read(lba, &mut data)?;
            *file = FileState {
                inode: INODES[index],
                lba,
                generation,
                sha256: virtio_ring::sector_sha256(&data),
            };
        }
        Ok(files)
    }

    /// Unknown blocks of an earlier plan whose bit is clear in THIS mount.
    pub fn unknown_freed(&self, before: &Plan) -> u32 {
        (0..self.block_count)
            .filter(|lba| allocated(&before.unknown, *lba) && !allocated(&self.bitmap, *lba))
            .count() as u32
    }

    /// Two-inode transaction, one inode after the other. Journal slots and
    /// pool blocks for BOTH inodes are reserved before the first mutation.
    pub fn write_pair(&mut self, io: &mut impl BlockIo) -> Result<(), Error> {
        let plan = self.plan(io)?;
        if plan.inode != self.inode || plan.bitmap != self.bitmap {
            return Err(Error::ReplayRequired);
        }
        if self.slots.iter().any(|slot| slot.iter().any(|b| *b != 0)) {
            return Err(Error::Full);
        }
        // The scratch map only proves space; each inode persists its own bit
        // so a cut never leaves the other inode's block allocated-unowned.
        let mut scratch = self.bitmap;
        let mut fresh = [0u64; 2];
        for block in fresh.iter_mut() {
            *block = turkfs::reserve_first_free_block(&mut scratch, POOL_START, self.block_count)
                .map_err(|e| match e {
                turkfs::Error::NoSpace => Error::NoSpace,
                _ => Error::Format,
            })?;
        }
        for (index, new_block) in fresh.into_iter().enumerate() {
            let (old_block, generation) = self.file(&self.inode, index)?;
            let data = pattern(index);
            let record = Record {
                generation: generation.checked_add(1).ok_or(Error::Scope)?,
                inode: INODES[index],
                old_block,
                new_block,
                data_hash: virtio_ring::sector_sha256(&data),
            };
            let base = index as u32 * 10;
            io.write(new_block, &data)?;
            io.checkpoint(base + 1, Step::DataWrite);
            io.flush()?;
            io.checkpoint(base + 2, Step::DataFlush);
            io.write(SLOTS[index], &record.encode(false)?)?;
            io.checkpoint(base + 3, Step::RecordWrite);
            io.flush()?;
            io.checkpoint(base + 4, Step::RecordFlush);
            let committed = record.encode(true)?;
            io.write(SLOTS[index], &committed)?;
            self.slots[index] = committed;
            io.checkpoint(base + 5, Step::CommitWrite);
            io.flush()?;
            io.checkpoint(base + 6, Step::CommitFlush);
            let mut inode = self.inode;
            set_pointer(&mut inode, index, new_block, record.generation);
            io.write(INODE_LBA, &inode)?;
            self.inode = inode;
            io.checkpoint(base + 7, Step::InodeWrite);
            io.flush()?;
            io.checkpoint(base + 8, Step::InodeFlush);
            if self.owners(&self.inode, old_block)? != 0 {
                return Err(Error::Conflict);
            }
            let mut bitmap = self.bitmap;
            set_allocated(&mut bitmap, new_block, true);
            set_allocated(&mut bitmap, old_block, false);
            io.write(BITMAP_LBA, &bitmap)?;
            self.bitmap = bitmap;
            io.checkpoint(base + 9, Step::BitmapWrite);
            io.flush()?;
            io.checkpoint(base + 10, Step::BitmapFlush);
        }
        Ok(())
    }

    /// Redo committed records. Writes only blocks that differ from the target,
    /// so a second replay of the same disk issues no write at all.
    pub fn replay(&mut self, io: &mut impl BlockIo) -> Result<ReplayOutcome, Error> {
        let plan = self.plan(io)?;
        let mut writes = 0;
        if plan.inode != self.inode {
            io.write(INODE_LBA, &plan.inode)?;
            self.inode = plan.inode;
            io.checkpoint(1, Step::ReplayInodeWrite);
            io.flush()?;
            io.checkpoint(2, Step::ReplayInodeFlush);
            writes += 1;
        }
        if plan.bitmap != self.bitmap {
            io.write(BITMAP_LBA, &plan.bitmap)?;
            self.bitmap = plan.bitmap;
            io.checkpoint(3, Step::ReplayBitmapWrite);
            io.flush()?;
            io.checkpoint(4, Step::ReplayBitmapFlush);
            writes += 1;
        }
        Ok(ReplayOutcome {
            applied: plan.apply,
            already: plan.already,
            writes,
        })
    }
}

#[cfg(test)]
mod tests {
    extern crate std;
    use super::*;
    use std::{vec, vec::Vec};

    const SECTORS: usize = 2048;
    #[derive(Clone)]
    struct Mem {
        disk: Vec<u8>,
        ops: Vec<(char, u64)>,
        checkpoints: Vec<(u32, Step, Vec<u8>)>,
        fail_mutation: Option<usize>,
        mutations: usize,
        after_failure: usize,
        failed: bool,
    }
    impl Mem {
        fn new(disk: Vec<u8>) -> Self {
            Self {
                disk,
                ops: vec![],
                checkpoints: vec![],
                fail_mutation: None,
                mutations: 0,
                after_failure: 0,
                failed: false,
            }
        }
        fn mutate(&mut self) -> Result<(), Error> {
            if self.failed {
                self.after_failure += 1;
            }
            let n = self.mutations;
            self.mutations += 1;
            if self.fail_mutation == Some(n) {
                self.failed = true;
                return Err(Error::Io);
            }
            Ok(())
        }
    }
    impl BlockIo for Mem {
        fn read(&mut self, lba: u64, out: &mut Block) -> Result<(), Error> {
            if self.failed {
                self.after_failure += 1;
            }
            out.copy_from_slice(&self.disk[lba as usize * 512..lba as usize * 512 + 512]);
            Ok(())
        }
        fn write(&mut self, lba: u64, bytes: &Block) -> Result<(), Error> {
            self.mutate()?;
            self.ops.push(('W', lba));
            self.disk[lba as usize * 512..lba as usize * 512 + 512].copy_from_slice(bytes);
            Ok(())
        }
        fn flush(&mut self) -> Result<(), Error> {
            self.mutate()?;
            self.ops.push(('F', 0));
            Ok(())
        }
        fn checkpoint(&mut self, step: u32, label: Step) {
            self.checkpoints.push((step, label, self.disk.clone()));
        }
    }

    fn put32(d: &mut [u8], at: usize, v: u32) {
        d[at..at + 4].copy_from_slice(&v.to_le_bytes());
    }
    fn put64(d: &mut [u8], at: usize, v: u64) {
        d[at..at + 8].copy_from_slice(&v.to_le_bytes());
    }
    /// Same layout as scripts/journal-disk-fixture.mjs (no queue-test prefix).
    fn fixture() -> Vec<u8> {
        let mut d = vec![0u8; SECTORS * 512];
        let sb = 16 * 512;
        d[sb..sb + 8].copy_from_slice(b"TURKFS\0\0");
        put32(&mut d, sb + 8, 1);
        put32(&mut d, sb + 12, 512);
        put64(&mut d, sb + 16, SECTORS as u64);
        put64(&mut d, sb + 24, 17);
        put32(&mut d, sb + 32, 4);
        put64(&mut d, sb + 36, 18);
        let names = [(1u32, "README"), (2, "KERNEL"), (3, "MULTI"), (4, "TAIL")];
        for (i, (inode, name)) in names.into_iter().enumerate() {
            let e = 17 * 512 + i * 32;
            d[e] = 1;
            d[e + 1] = name.len() as u8;
            put32(&mut d, e + 4, inode);
            d[e + 8..e + 8 + name.len()].copy_from_slice(name.as_bytes());
        }
        let inodes = [
            (1u32, 0u64, 0u64, 0u64),
            (2, 512, 19, 1),
            (3, 1024, 20, 0),
            (4, 512, 22, 1),
        ];
        for (inode, size, lba, generation) in inodes {
            let at = 18 * 512 + (inode as usize - 1) * 32;
            put32(&mut d, at, inode);
            put64(&mut d, at + 8, size);
            put64(&mut d, at + 16, lba);
            put64(&mut d, at + 24, generation);
        }
        for (lba, xor) in [
            (19usize, 0x5a),
            (20, 0x3c),
            (21, 0xc3),
            (22, 0x7e),
            (23, 0x42),
        ] {
            for i in 0..512 {
                d[lba * 512 + i] = (i as u8) ^ xor;
            }
        }
        for lba in (0..=22).chain(26..=29) {
            d[26 * 512 + lba / 8] |= 1 << (lba % 8);
        }
        d[27 * 512..27 * 512 + 8].copy_from_slice(MARKER);
        put32(&mut d, 27 * 512 + 8, 2);
        d
    }
    fn bit(d: &[u8], lba: u64) -> bool {
        d[26 * 512 + (lba / 8) as usize] & (1 << (lba % 8)) != 0
    }
    fn files(d: &[u8]) -> [FileState; 2] {
        let mut m = Mem::new(d.to_vec());
        Mount::open(&mut m, SECTORS as u64)
            .unwrap()
            .files(&mut m)
            .unwrap()
    }
    fn plan(d: &[u8]) -> Plan {
        let mut m = Mem::new(d.to_vec());
        Mount::open(&mut m, SECTORS as u64)
            .unwrap()
            .plan(&mut m)
            .unwrap()
    }
    fn old_hashes() -> [[u8; 32]; 2] {
        files(&fixture()).map(|f| f.sha256)
    }
    fn new_hashes() -> [[u8; 32]; 2] {
        [0, 1].map(|i| virtio_ring::sector_sha256(&pattern(i)))
    }
    fn replay(d: &[u8]) -> (Mem, Result<ReplayOutcome, Error>) {
        let mut m = Mem::new(d.to_vec());
        let r = Mount::open(&mut m, SECTORS as u64).and_then(|mut x| x.replay(&mut m));
        (m, r)
    }
    fn written() -> Mem {
        let mut m = Mem::new(fixture());
        Mount::open(&mut m, SECTORS as u64)
            .unwrap()
            .write_pair(&mut m)
            .unwrap();
        m
    }

    #[test]
    fn p12_k543_writer_order_is_data_flush_record_commit_inode_bitmap_per_inode() {
        let m = written();
        let mut expected = vec![];
        for (slot, fresh) in [(28, 30), (29, 31)] {
            expected.extend([
                ('W', fresh),
                ('F', 0),
                ('W', slot),
                ('F', 0),
                ('W', slot),
                ('F', 0),
            ]);
            expected.extend([('W', 18), ('F', 0), ('W', 26), ('F', 0)]);
        }
        assert_eq!(m.ops, expected);
        for (i, (n, label, _)) in m.checkpoints.iter().enumerate() {
            assert_eq!(*n as usize, i + 1);
            let kind = if i % 2 == 0 { "-WRITE" } else { "-FLUSH" };
            assert!(label.label().ends_with(kind), "{label:?}");
        }
        assert_eq!(files(&m.disk).map(|f| f.sha256), new_hashes());
        assert!(bit(&m.disk, 30) && bit(&m.disk, 31) && !bit(&m.disk, 19) && !bit(&m.disk, 22));
    }

    #[test]
    fn p12_k544_full_or_one_free_slot_journal_rejects_before_any_write() {
        let full = written();
        let mut m = Mem::new(full.disk.clone());
        let mut mount = Mount::open(&mut m, SECTORS as u64).unwrap();
        assert_eq!(mount.write_pair(&mut m), Err(Error::Full));
        assert!(m.ops.is_empty());
        assert_eq!(m.disk, full.disk);
        // One slot holds an older, already-applied record: the two-inode
        // transaction cannot fit and must not commit inode 2 alone.
        let mut d = fixture();
        let older = Record {
            generation: 1,
            inode: 4,
            old_block: 23,
            new_block: 22,
            data_hash: virtio_ring::sector_sha256(d[22 * 512..23 * 512].try_into().unwrap()),
        };
        d[29 * 512..30 * 512].copy_from_slice(&older.encode(true).unwrap());
        assert_eq!((plan(&d).already, plan(&d).bitmap_changes), (1, 0));
        let mut m = Mem::new(d.clone());
        let mut mount = Mount::open(&mut m, SECTORS as u64).unwrap();
        assert_eq!(mount.write_pair(&mut m), Err(Error::Full));
        assert!(m.ops.is_empty());
        assert_eq!(m.disk, d);
    }

    #[test]
    fn p12_k545_replay_changes_only_the_disposable_copy() {
        let m = written();
        let (_, _, cut) = &m.checkpoints[5]; // inode 2 COMMIT-FLUSH
        let source = cut.clone();
        let (copy, outcome) = replay(&source);
        let expected = ReplayOutcome {
            applied: 1,
            already: 0,
            writes: 2,
        };
        assert_eq!(outcome, Ok(expected));
        assert_eq!(&source, cut);
        assert_ne!(copy.disk, source);
        let f = files(&copy.disk);
        assert_eq!(
            [f[0].sha256, f[1].sha256],
            [new_hashes()[0], old_hashes()[1]]
        );
        assert_eq!((f[0].lba, f[0].generation), (30, 2));
    }

    #[test]
    fn p12_k546_second_replay_writes_nothing_and_never_advances_generation() {
        for (_, _, cut) in written().checkpoints {
            let (first, _) = replay(&cut);
            let (second, outcome) = replay(&first.disk);
            assert_eq!(outcome.unwrap().writes, 0);
            assert!(second.ops.is_empty());
            assert_eq!(second.disk, first.disk);
            assert!(files(&second.disk).iter().all(|f| f.generation <= 2));
        }
    }

    #[test]
    fn p12_k547_cut_before_commit_keeps_old_file_and_hash() {
        let m = written();
        for (step, label, cut) in &m.checkpoints {
            let (r, _) = replay(cut);
            let index = if *step <= 10 { 0 } else { 1 };
            let local = step - index as u32 * 10;
            let expected = if local < 5 {
                old_hashes()
            } else {
                new_hashes()
            };
            assert_eq!(
                files(&r.disk)[index].sha256,
                expected[index],
                "{step} {label:?}"
            );
        }
        let flushed = &m.checkpoints[3].2; // uncommitted record flushed
        let p = plan(flushed);
        assert_eq!((p.incomplete, p.committed, p.apply), (1, 0, 0));
        let (r, outcome) = replay(flushed);
        assert_eq!(outcome.unwrap().writes, 0);
        assert!(r.ops.is_empty());
        let mut damaged = flushed.clone();
        damaged[28 * 512 + 80] ^= 1;
        assert!(matches!(replay(&damaged).1, Err(Error::Journal(_))));
    }

    #[test]
    fn p12_k548_corrupt_record_or_data_hash_is_explicit_error_without_writes() {
        let cut = written().checkpoints[5].2.clone();
        let mut checksum = cut.clone();
        checksum[28 * 512 + 40] ^= 1;
        let mut data = cut.clone();
        data[30 * 512] ^= 1;
        for (bytes, error) in [
            (checksum, Error::Journal(journal::Error::Checksum)),
            (data, Error::Journal(journal::Error::DataHash)),
        ] {
            let (m, result) = replay(&bytes);
            assert_eq!(result, Err(error));
            assert!(m.ops.is_empty());
            assert_eq!(m.disk, bytes);
        }
        let mut conflict = cut;
        conflict[18 * 512 + 32 + 24] = 7; // generation is no longer the predecessor
        let (m, result) = replay(&conflict);
        assert_eq!(result, Err(Error::Conflict));
        assert!(m.ops.is_empty());
    }

    #[test]
    fn p12_k549_unknown_allocated_block_is_counted_and_never_freed() {
        let mut d = fixture();
        d[26 * 512 + 40 / 8] |= 1 << (40 % 8);
        d[40 * 512..41 * 512].fill(0xee);
        let before = plan(&d);
        assert_eq!(
            (before.ownership.unknown, before.ownership.allocated),
            (1, 28)
        );
        let mut m = Mem::new(d);
        Mount::open(&mut m, SECTORS as u64)
            .unwrap()
            .write_pair(&mut m)
            .unwrap();
        let (r, _) = replay(&m.disk);
        let mut mm = Mem::new(r.disk.clone());
        assert_eq!(
            Mount::open(&mut mm, SECTORS as u64)
                .unwrap()
                .unknown_freed(&before),
            0
        );
        assert!(bit(&r.disk, 40) && r.disk[40 * 512..41 * 512].iter().all(|b| *b == 0xee));
        // A full pool reports ENOSPC; unknown blocks are not reclaimed for it.
        let mut full = fixture();
        for lba in POOL_START..SECTORS as u64 {
            full[26 * 512 + (lba / 8) as usize] |= 1 << (lba % 8);
        }
        assert_eq!(
            plan(&full).ownership.unknown,
            SECTORS as u32 - POOL_START as u32
        );
        let mut m = Mem::new(full.clone());
        let mut mount = Mount::open(&mut m, SECTORS as u64).unwrap();
        assert_eq!(mount.write_pair(&mut m), Err(Error::NoSpace));
        assert!(m.ops.is_empty());
        // Writer cut between inode and bitmap: old block is journal-released.
        let o = plan(&written().checkpoints[7].2).ownership;
        assert_eq!((o.released, o.claimed, o.unknown), (1, 1, 0));
    }

    #[test]
    fn p12_k550_two_inodes_have_independent_ordering_and_hash_boundaries() {
        let mut seen = std::collections::BTreeSet::new();
        for (step, _, cut) in written().checkpoints {
            let (r, _) = replay(&cut);
            let f = files(&r.disk);
            let state = [0, 1].map(|i| f[i].sha256 == new_hashes()[i]);
            assert_eq!(state, [step >= 5, step >= 15], "{step}");
            seen.insert(state);
        }
        assert!(seen.contains(&[true, false]) && seen.contains(&[true, true]));
        assert!(old_hashes()[0] != old_hashes()[1] && new_hashes()[0] != new_hashes()[1]);
    }

    #[test]
    fn p12_k551_every_writer_checkpoint_snapshot_recovers_consistently() {
        let m = written();
        assert_eq!(m.checkpoints.len(), WRITER_STEPS as usize);
        for (_, _, cut) in &m.checkpoints {
            let (r, outcome) = replay(cut);
            outcome.unwrap();
            let p = plan(&r.disk);
            assert_eq!((p.apply, p.bitmap_changes), (0, 0));
            let o = p.ownership;
            assert_eq!((o.released, o.claimed, o.unknown), (0, 0, 0));
        }
    }

    #[test]
    fn p12_k551_bootargs_and_fdt_chosen_are_strict() {
        let write = Params {
            mode: Mode::Write,
            pause: 20,
        };
        assert_eq!(parse_params(b"turkos.p12=write,pause=20"), Ok(write));
        let read = Params {
            mode: Mode::Read,
            pause: 0,
        };
        assert_eq!(parse_params(b"turkos.p12=read"), Ok(read));
        for bad in [
            &b""[..],
            b"turkos.p12=write,pause=21",
            b"turkos.p12=replay,pause=5",
            b"turkos.p12=read,pause=1",
            b"turkos.p12=write,pause=0",
            b"turkos.p12=write,pause=05",
            b"turkos.p12=write,pause=",
            b"turkos.p12=write,pause=3,x",
            b"turkos.p12=format",
            b" turkos.p12=write",
        ] {
            assert_eq!(parse_params(bad), Err(Error::Parameters));
        }
        // Minimal FDT: / { <node> { bootargs = "..." } }.
        fn fdt(args: &[u8], node: &[u8]) -> Vec<u8> {
            let mut s = vec![];
            let w = |s: &mut Vec<u8>, v: u32| s.extend(v.to_be_bytes());
            let pad = |s: &mut Vec<u8>| {
                while s.len() % 4 != 0 {
                    s.push(0)
                }
            };
            w(&mut s, 1);
            s.extend([0, 0, 0, 0]);
            w(&mut s, 1);
            s.extend(node);
            s.push(0);
            pad(&mut s);
            w(&mut s, 3);
            w(&mut s, args.len() as u32 + 1);
            w(&mut s, 0);
            s.extend(args);
            s.push(0);
            pad(&mut s);
            w(&mut s, 2);
            w(&mut s, 2);
            w(&mut s, 9);
            let strings = b"bootargs\0";
            let mut b = vec![0u8; 40];
            let total = 40 + s.len() + strings.len();
            for (at, v) in [
                (0, 0xd00d_feedu32),
                (4, total as u32),
                (8, 40),
                (12, 40 + s.len() as u32),
                (20, 17),
                (24, 16),
                (32, strings.len() as u32),
                (36, s.len() as u32),
            ] {
                b[at..at + 4].copy_from_slice(&v.to_be_bytes());
            }
            b.extend(s);
            b.extend(strings);
            b
        }
        let blob = fdt(b"turkos.p12=replay,pause=4", b"chosen");
        assert_eq!(
            chosen_bootargs(&blob),
            Ok(&b"turkos.p12=replay,pause=4"[..])
        );
        assert_eq!(
            chosen_bootargs(&fdt(b"x", b"memory")),
            Err(Error::Parameters)
        );
        assert_eq!(
            chosen_bootargs(&blob[..blob.len() - 12]),
            Err(Error::Parameters)
        );
        let mut bad = blob.clone();
        bad[0] = 0;
        assert_eq!(chosen_bootargs(&bad), Err(Error::Parameters));
    }

    #[test]
    fn p12_k553_interrupted_replay_reopens_to_the_same_result() {
        for (_, _, cut) in written().checkpoints {
            let (full, _) = replay(&cut);
            for (_, _, partial) in full.checkpoints.clone() {
                let (again, outcome) = replay(&partial);
                outcome.unwrap();
                assert_eq!(again.disk, full.disk);
            }
        }
    }

    #[test]
    fn p12_k554_failed_request_stops_io_and_leaves_old_or_new_state() {
        let total = written().mutations;
        assert_eq!(total, 20);
        for fail in 0..total {
            let mut m = Mem::new(fixture());
            m.fail_mutation = Some(fail);
            let mut mount = Mount::open(&mut m, SECTORS as u64).unwrap();
            assert_eq!(mount.write_pair(&mut m), Err(Error::Io));
            assert_eq!(m.after_failure, 0, "request issued after failure {fail}");
            let (r, outcome) = replay(&m.disk);
            outcome.unwrap();
            let f = files(&r.disk);
            for index in 0..2 {
                // COMMIT-WRITE is mutation 4 (inode 2) or 14 (inode 4).
                let committed = fail > 4 + index * 10;
                let expected = if committed {
                    new_hashes()
                } else {
                    old_hashes()
                };
                assert_eq!(
                    f[index].sha256, expected[index],
                    "fail {fail} index {index}"
                );
            }
        }
    }

    // Internal pre-audit inputs for the reviewer's questions. They are not
    // the independent review and deliberately carry no contract K number.
    /// Snapshot-free device for large matrices; counts every mutation.
    struct Lean {
        disk: Vec<u8>,
        mutations: usize,
    }
    impl BlockIo for Lean {
        fn read(&mut self, lba: u64, out: &mut Block) -> Result<(), Error> {
            out.copy_from_slice(&self.disk[lba as usize * 512..][..512]);
            Ok(())
        }
        fn write(&mut self, lba: u64, bytes: &Block) -> Result<(), Error> {
            self.mutations += 1;
            self.disk[lba as usize * 512..][..512].copy_from_slice(bytes);
            Ok(())
        }
        fn flush(&mut self) -> Result<(), Error> {
            self.mutations += 1;
            Ok(())
        }
        fn checkpoint(&mut self, _: u32, _: Step) {}
    }
    fn slot_committed(d: &[u8], index: usize) -> bool {
        Record::decode(&d[SLOTS[index] as usize * 512..][..512]).is_ok()
    }

    #[test]
    fn preaudit_torn_sector_at_every_writer_write_is_old_new_or_refused() {
        // Torn 512-byte write: a byte prefix (or suffix) of the target sector
        // is new, the rest still old. Physical tearing is not claimed solved.
        let m = written();
        let (old, new) = (files(&fixture()), files(&m.disk));
        let mut cuts: Vec<usize> = (0..=512).step_by(8).collect();
        cuts.extend([1, 4, 12, 28, 81, 111, 113, 503, 505, 508, 511]);
        let (mut refused, mut recovered) = (0, 0);
        for (n, (_, label, after)) in m.checkpoints.iter().enumerate() {
            let ('W', lba) = m.ops[n] else { continue };
            let before = if n == 0 {
                fixture()
            } else {
                m.checkpoints[n - 1].2.clone()
            };
            let at = lba as usize * 512;
            for &cut in &cuts {
                for new_first in [true, false] {
                    let span = if new_first {
                        at..at + cut
                    } else {
                        at + cut..at + 512
                    };
                    let mut torn = before.clone();
                    torn[span.clone()].copy_from_slice(&after[span]);
                    let committed = [0, 1].map(|i| slot_committed(&torn, i));
                    let mut io = Lean {
                        disk: torn,
                        mutations: 0,
                    };
                    let why = std::format!("{label:?} cut={cut} new_first={new_first}");
                    let result =
                        Mount::open(&mut io, SECTORS as u64).and_then(|mut x| x.replay(&mut io));
                    if result.is_err() {
                        // Fail closed: refused before any mutation of the disk.
                        assert_eq!(io.mutations, 0, "{why}");
                        refused += 1;
                        continue;
                    }
                    recovered += 1;
                    let mount = Mount::open(&mut io, SECTORS as u64).unwrap();
                    let p = mount.plan(&mut io).unwrap();
                    let o = p.ownership;
                    assert_eq!((p.apply, p.bitmap_changes), (0, 0), "{why}");
                    assert_eq!((o.released, o.claimed, o.unknown), (0, 0, 0), "{why}");
                    let now = mount.files(&mut io).unwrap();
                    for i in 0..2 {
                        // New state exactly when the full commit marker survived.
                        let expected = if committed[i] { new[i] } else { old[i] };
                        assert_eq!(now[i], expected, "{why} inode index {i}");
                    }
                }
            }
        }
        // Both outcomes occur: e.g. a torn record sector refuses the mount.
        assert!(refused > 0 && recovered > 0, "{refused} {recovered}");
    }

    #[test]
    fn preaudit_commit_marker_outside_digest_relies_on_never_reusing_a_slot() {
        let r = |inode, old_block, new_block| Record {
            generation: 2,
            inode,
            old_block,
            new_block,
            data_hash: [7; 32],
        };
        let stale = r(2, 19, 30).encode(true).unwrap();
        let body = r(4, 22, 31).encode(false).unwrap();
        // SHA-256 covers 0..80 and 112..504 only. Over a committed sector, a
        // torn rewrite keeping the old tail would publish the new body.
        let mut torn = body;
        torn[504..].copy_from_slice(&stale[504..]);
        assert_eq!(Record::decode(&torn), Ok(r(4, 22, 31)));
        // Hence the writer refuses any non-zero slot, even an incomplete one.
        let mut d = fixture();
        d[28 * 512..29 * 512].copy_from_slice(&r(2, 19, 30).encode(false).unwrap());
        assert_eq!((plan(&d).incomplete, plan(&d).committed), (1, 0));
        let mut m = Mem::new(d.clone());
        let mut mount = Mount::open(&mut m, SECTORS as u64).unwrap();
        assert_eq!(mount.write_pair(&mut m), Err(Error::Full));
        assert!(m.ops.is_empty());
        assert_eq!(m.disk, d);
    }

    #[test]
    fn preaudit_resealed_field_tricks_and_marker_damage_never_commit() {
        use sha2::{Digest, Sha256};
        let seal = |mut b: Block| {
            let mut h = Sha256::new();
            h.update(&b[..80]);
            h.update(&b[112..504]);
            b[80..112].copy_from_slice(&h.finalize());
            b
        };
        let r = Record {
            generation: 2,
            inode: 2,
            old_block: 19,
            new_block: 30,
            data_hash: [7; 32],
        };
        let good = r.encode(true).unwrap();
        assert_eq!(seal(good), good);
        // A valid digest does not authorize any header or reserved field.
        let field = |at: usize, bytes: &[u8]| {
            let mut b = good;
            b[at..at + bytes.len()].copy_from_slice(bytes);
            Record::decode(&seal(b))
        };
        for length in [0u32, 511, 513, u32::MAX] {
            assert_eq!(
                field(12, &length.to_le_bytes()),
                Err(journal::Error::Length)
            );
        }
        assert_eq!(field(8, &2u32.to_le_bytes()), Err(journal::Error::Version));
        for at in [28, 31, 112, 300, 503] {
            assert_eq!(field(at, &[1]), Err(journal::Error::Fields), "{at}");
        }
        assert_eq!(field(16, &0u64.to_le_bytes()), Err(journal::Error::Fields));
        assert_eq!(field(40, &19u64.to_le_bytes()), Err(journal::Error::Fields));
        let long = [good, good].concat();
        assert_eq!(Record::decode(&long), Err(journal::Error::Length));
        // The marker is outside the digest: damage there is never Ok.
        for at in 504..512 {
            for bit in 0..8 {
                let mut b = good;
                b[at] ^= 1 << bit;
                assert_eq!(Record::decode(&b), Err(journal::Error::Uncommitted));
            }
        }
    }

    #[test]
    fn preaudit_replay_refuses_rollback_skip_and_newer_current_state() {
        let hash = [9; 32];
        let r = Record {
            generation: 3,
            inode: 2,
            old_block: 19,
            new_block: 30,
            data_hash: hash,
        };
        let at = |generation, block| Pointer { generation, block };
        assert_eq!(r.replay(at(2, 19), hash), Ok(Replay::Apply(at(3, 30))));
        assert_eq!(r.replay(at(3, 30), hash), Ok(Replay::AlreadyApplied));
        // Older-by-two, same or newer generation, or a foreign block: never
        // applied and never mistaken for an already-applied record.
        for current in [
            at(1, 19),
            at(3, 19),
            at(4, 19),
            at(4, 30),
            at(2, 30),
            at(u64::MAX, 30),
            at(2, 31),
        ] {
            assert_eq!(r.replay(current, hash), Err(journal::Error::Conflict));
            assert_eq!(r.replay(current, [0; 32]), Err(journal::Error::DataHash));
        }
    }

    #[test]
    fn preaudit_unowned_block_is_released_only_when_a_committed_record_names_it() {
        let mut d = fixture();
        d[26 * 512 + 40 / 8] |= 1 << (40 % 8);
        d[40 * 512..41 * 512].fill(0xee);
        assert_eq!(plan(&d).ownership.unknown, 1);
        // An already-applied record naming LBA40 as its old block explains it.
        let named = |old_block| Record {
            generation: 1,
            inode: 4,
            old_block,
            new_block: 22,
            data_hash: virtio_ring::sector_sha256(d[22 * 512..23 * 512].try_into().unwrap()),
        };
        let mut explained = d.clone();
        explained[29 * 512..30 * 512].copy_from_slice(&named(40).encode(true).unwrap());
        let o = plan(&explained).ownership;
        assert_eq!((plan(&explained).already, o.released, o.unknown), (1, 1, 0));
        let (r, outcome) = replay(&explained);
        assert_eq!(outcome.unwrap().writes, 1);
        assert!(!bit(&r.disk, 40) && r.disk[40 * 512..41 * 512].iter().all(|b| *b == 0xee));
        // A record naming an OWNED block (MULTI at LBA20) is a conflict.
        let mut owned = d.clone();
        owned[29 * 512..30 * 512].copy_from_slice(&named(20).encode(true).unwrap());
        let (m, outcome) = replay(&owned);
        assert_eq!(outcome, Err(Error::Conflict));
        assert!(m.ops.is_empty());
    }
}