sqlite-forensic 0.10.2

//! `sqlite-forensic` — Tier-2 anomaly auditor over [`sqlite_core`].
//!
//! WS-C skeleton. The reader (`sqlite-core`) answers "what does this database
//! header show?"; this crate grades the forensically-notable observations into
//! severity-ranked [`forensicnomicon::report::Finding`]s, so a `SQLite`
//! evidence database's anomalies aggregate uniformly with the partition /
//! container / filesystem layers.
//!
//! Each anomaly is an *observation* ("consistent with …"); the examiner draws
//! the conclusions.
//!
//! # Capabilities
//!
//! - [`carve_deleted_records`] — recover deleted rows from free (unallocated)
//!   pages, the headline capability rusqlite structurally cannot provide. Each
//!   recovered row is confidence-graded, flagged `allocated: false`, and carries
//!   page/offset/rowid provenance.
//! - [`audit`] grades header reserved-space, a non-empty freelist (prior
//!   deletions), an active WAL overlay (uncheckpointed state), and a header/file
//!   page-count mismatch into severity-ranked
//!   [`forensicnomicon::report::Finding`]s.
//! - [`audit_journal`] grades the rollback-`-journal` design-§6 observations
//!   (hot journal, recoverable PERSIST pre-images, checksum mismatch, journaled
//!   schema page, duplicate page record, db-size delta) — additive to [`audit`].
//!
//! Deferred: a full anomaly suite (overflow-chain integrity, schema-format /
//! text-encoding checks) and a fuzz harness.

#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]

pub mod blob;
pub mod carve;
pub mod case_uco;
pub mod interpret;

use forensicnomicon::report::{
    Confidence, Evidence, Finding, Location, Observation, Severity, Source,
};
use sqlite_core::{CommitId, Database, JournalHeader, RollbackJournal, Value, WalTimeline};

/// The classified `SQLite` forensic anomalies this auditor can grade.
///
/// `#[non_exhaustive]` so WS-E can add carving / WAL / freelist variants without
/// a breaking change; downstream `match` arms must carry a `_` arm.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum AnomalyKind {
    /// The header's reserved-space-per-page field is non-zero. Standard
    /// `SQLite` leaves this at 0; a non-zero value is used by page-level
    /// extensions (e.g. encryption such as SQLCipher/SEE, or checksum VFS) and
    /// is worth flagging on an evidence database.
    NonZeroReservedSpace {
        /// The reserved bytes per page reported by the header.
        reserved: u8,
    },
    /// A record-shaped cell was recovered from unallocated / free space —
    /// consistent with a deleted row that has not yet been overwritten.
    DeletedRecordRecovered {
        /// 1-based page the residue was carved from.
        page: u32,
        /// Byte offset of the cell within that page.
        offset: usize,
        /// Recovered rowid.
        rowid: i64,
    },
    /// A `sqlite_master` row was recovered from page-1 free space whose
    /// definition is absent from the live schema — consistent with a **dropped
    /// (or replaced)** table/index/view/trigger whose `CREATE` statement and
    /// existence survive the drop under `secure_delete=OFF`.
    DroppedSchemaRecovered {
        /// `sqlite_master.type` — `table`, `index`, `view`, or `trigger`.
        object_type: String,
        /// Name of the dropped object.
        name: String,
    },
    /// The freelist is non-empty: the database holds free (unallocated) pages.
    /// Consistent with prior deletions (`DELETE` without `VACUUM`); those pages
    /// may retain recoverable deleted records.
    NonEmptyFreelist {
        /// Number of free pages on the freelist.
        free_pages: u32,
    },
    /// A `-wal` sidecar carried committed-but-unflushed page versions that the
    /// main database file does not yet reflect. Consistent with an evidence
    /// database captured while a write transaction was checkpoint-pending; the
    /// main file alone would under-report the true state.
    WalUncheckpointedState {
        /// Number of pages the WAL overlay superseded in the main file.
        overlaid_pages: u32,
    },
    /// The in-header page count disagrees with the page count implied by the
    /// file length. Consistent with truncation, carving, or out-of-band
    /// modification of the database file.
    PageCountMismatch {
        /// Page count recorded in the file header (offset 28).
        header_pages: u32,
        /// Page count implied by `file_len / page_size`.
        file_pages: u32,
    },
    /// A `-journal` with an intact header (valid magic) sits beside the database
    /// — a *hot journal*. Consistent with an interrupted or in-progress write
    /// transaction (crash, power loss, process kill, or acquisition captured
    /// mid-write); `SQLite` would roll it back on next open, so the main db may
    /// require rollback. The journal holds the pre-interruption state. (Design §6
    /// item 1 — state the observation; never assert "anti-forensic".)
    HotJournal {
        /// Database page count recorded at transaction start (`dbOrigSize`).
        mx_page: u32,
        /// Number of pre-transaction page images the journal carries — the scope
        /// of the interrupted transaction.
        journaled_pages: u32,
    },
    /// A committed PERSIST `-journal` (header zeroed, bodies intact) carries
    /// recoverable pre-images. Consistent with a normally-committed transaction
    /// whose deleted/modified rows remain recoverable (design §6 item 2).
    JournalRecoverable {
        /// Number of pre-transaction page images the journal carries.
        images: usize,
    },
    /// One or more journal page records failed the `pager.c` checksum (Tier A
    /// only — Tier B has no nonce to verify against). Consistent with corruption,
    /// a torn page (power-loss mid-sector), or post-write modification of the
    /// journal (design §6 item 3).
    JournalChecksumMismatch {
        /// The page numbers whose stored checksum did not match, ascending.
        pgnos: Vec<u32>,
        /// Total page records the journal carries (the denominator).
        total: usize,
    },
    /// The journal's prior **page-1 image** carries a different schema cookie
    /// (file-header offset 40, 4-byte BE) than the live database. The cookie
    /// advances only on `CREATE`/`DROP`/`ALTER`, so a difference is consistent
    /// with a DDL change in the last transaction; the prior schema is recoverable
    /// (design §6 item 6). Page 1 alone is NOT sufficient — it is journaled on
    /// nearly every write (the change-counter / freelist-count / db-size header
    /// fields update routinely), so the cookie comparison, not page-1 presence,
    /// is the DDL signal.
    JournalSchemaChange {
        /// Schema cookie in the journal's prior page-1 image (offset 40, BE).
        journal_cookie: u32,
        /// Schema cookie in the live database's page 1 (offset 40, BE).
        db_cookie: u32,
    },
    /// A `pgno` appeared more than once across the journal's page records. The
    /// spec journals a page at most once, so a repeat is consistent with
    /// corruption, a savepoint/super-journal artifact, or tampering (design §6
    /// item 9).
    JournalDuplicatePage {
        /// The page numbers that repeated (each once, first-seen order) — the
        /// offending values, so the finding can name them.
        pgnos: Vec<u32>,
    },
    /// The database page count recorded at transaction start (`mxPage`, Tier A
    /// only) differs from the current page count: the last transaction changed
    /// the db size. `mxPage < current` ⇒ growth (INSERTs); `mxPage > current` ⇒
    /// shrink, consistent with auto-vacuum/incremental-vacuum or truncation — NOT
    /// an ordinary `DELETE` (design §6 item 5).
    JournalDbSizeDelta {
        /// Database page count at transaction start (`mxPage`, journal header).
        mx_page: u32,
        /// Current database page count.
        current_pages: u32,
    },
    /// The free-page count declared in the header (offset 36) disagrees with the
    /// count obtained by walking the freelist trunk chain — or the chain is
    /// unwalkable (out-of-range/cyclic trunk pointer). The header field is a
    /// *claim*; the walk *verifies* it. A mismatch is consistent with freelist
    /// tampering (a hand-edited count to hide freed pages) or corruption
    /// (design §2.1 — distrust the header).
    FreelistCountInconsistent {
        /// Free-page count declared in the header (offset 36).
        declared: u32,
        /// Free-page count obtained by walking the trunk chain, or `None` when
        /// the chain could not be walked (malformed/cyclic/out-of-range trunk).
        walked: Option<u32>,
    },
    /// Freed freelist *leaf* pages are entirely zero. A leaf page keeps its
    /// former content byte-for-byte, so an all-zero freed leaf means the deleted
    /// records it held were overwritten with zeros — residue deliberately
    /// destroyed. Consistent with `secure_delete=ON` or a manual wipe (design
    /// §2.1); it is a fingerprint of destruction, never proof of intent.
    ZeroedFreelistResidue {
        /// Number of freed leaf pages that are entirely zero.
        zeroed: u32,
        /// Total number of freed leaf pages (the denominator).
        free_leaves: u32,
        /// The lowest zeroed freed-leaf page number — a pointer to the evidence.
        first_zeroed: u32,
    },
}

impl AnomalyKind {
    /// Severity, derived from the kind.
    #[must_use]
    pub fn severity(&self) -> Severity {
        match self {
            AnomalyKind::NonZeroReservedSpace { .. } | AnomalyKind::NonEmptyFreelist { .. } => {
                Severity::Low
            }
            AnomalyKind::DeletedRecordRecovered { .. }
            | AnomalyKind::DroppedSchemaRecovered { .. }
            | AnomalyKind::WalUncheckpointedState { .. }
            | AnomalyKind::JournalRecoverable { .. }
            | AnomalyKind::JournalSchemaChange { .. }
            | AnomalyKind::JournalDuplicatePage { .. }
            | AnomalyKind::ZeroedFreelistResidue { .. } => Severity::Medium,
            AnomalyKind::PageCountMismatch { .. }
            | AnomalyKind::HotJournal { .. }
            | AnomalyKind::JournalChecksumMismatch { .. }
            | AnomalyKind::FreelistCountInconsistent { .. } => Severity::High,
            AnomalyKind::JournalDbSizeDelta { .. } => Severity::Low,
        }
    }

    /// Stable, scheme-prefixed machine code (a published contract).
    #[must_use]
    pub fn code(&self) -> &'static str {
        match self {
            AnomalyKind::NonZeroReservedSpace { .. } => "SQLITE-RESERVED-SPACE-NONZERO",
            AnomalyKind::DeletedRecordRecovered { .. } => "SQLITE-DELETED-RECORD-RECOVERED",
            AnomalyKind::DroppedSchemaRecovered { .. } => "SQLITE-DROPPED-SCHEMA-RECOVERED",
            AnomalyKind::NonEmptyFreelist { .. } => "SQLITE-FREELIST-NONEMPTY",
            AnomalyKind::WalUncheckpointedState { .. } => "SQLITE-WAL-UNCHECKPOINTED",
            AnomalyKind::PageCountMismatch { .. } => "SQLITE-PAGECOUNT-MISMATCH",
            AnomalyKind::HotJournal { .. } => "SQLITE-JOURNAL-HOT",
            AnomalyKind::JournalRecoverable { .. } => "SQLITE-JOURNAL-RECOVERABLE",
            AnomalyKind::JournalChecksumMismatch { .. } => "SQLITE-JOURNAL-CHECKSUM-MISMATCH",
            AnomalyKind::JournalSchemaChange { .. } => "SQLITE-JOURNAL-SCHEMA-CHANGE",
            AnomalyKind::JournalDuplicatePage { .. } => "SQLITE-JOURNAL-DUPLICATE-PAGE",
            AnomalyKind::JournalDbSizeDelta { .. } => "SQLITE-JOURNAL-DBSIZE-DELTA",
            AnomalyKind::FreelistCountInconsistent { .. } => "SQLITE-FREELIST-COUNT-INCONSISTENT",
            AnomalyKind::ZeroedFreelistResidue { .. } => "SQLITE-FREELIST-RESIDUE-ZEROED",
        }
    }

    /// Human-readable, "consistent with" note.
    #[must_use]
    pub fn note(&self) -> String {
        match self {
            AnomalyKind::NonZeroReservedSpace { reserved } => {
                // Name the scheme the reserved-byte count matches (values measured
                // against sqlcipher 4.16 and the checksum-VFS spec, cksumvfs.html);
                // an unmatched value is labelled unrecognized with the raw bytes
                // shown, never guessed into a named scheme.
                let scheme = match reserved {
                    80 => {
                        "consistent with SQLCipher 4 (a 16-byte per-page IV plus a \
                           64-byte HMAC-SHA512)"
                    }
                    48 => {
                        "consistent with SQLCipher 1-3 (a 16-byte per-page IV plus a \
                           32-byte HMAC-SHA1)"
                    }
                    16 => {
                        "consistent with a 16-byte per-page IV (an AES-CBC cipher with \
                           per-page HMAC disabled)"
                    }
                    8 => {
                        "consistent with the SQLite checksum VFS (an 8-byte per-page \
                          checksum)"
                    }
                    _ => {
                        "does not match a known page-level scheme's per-page reservation \
                          (an unrecognized extension)"
                    }
                };
                format!(
                    "file header reserves {reserved} byte(s) per page (offset 20) — \
                     non-standard; {scheme}. Recovering any deleted records would require \
                     the encryption key (or the corresponding VFS)"
                )
            }
            AnomalyKind::DeletedRecordRecovered {
                page,
                offset,
                rowid,
            } => format!(
                "recovered a record-shaped cell (rowid {rowid}) from unallocated \
                 space at page {page} offset {offset} — consistent with a deleted \
                 row not yet overwritten"
            ),
            AnomalyKind::DroppedSchemaRecovered { object_type, name } => format!(
                "recovered a deleted sqlite_master row for {object_type} \"{name}\" \
                 from page-1 free space — consistent with a dropped (or replaced) \
                 {object_type} whose definition survives the drop"
            ),
            AnomalyKind::NonEmptyFreelist { free_pages } => format!(
                "{free_pages} free page(s) on the freelist — consistent with prior \
                 deletions (DELETE without VACUUM); free pages may retain \
                 recoverable deleted records"
            ),
            AnomalyKind::WalUncheckpointedState { overlaid_pages } => format!(
                "the -wal sidecar carries {overlaid_pages} committed page version(s) \
                 the main file does not reflect — consistent with capture while a \
                 write transaction was checkpoint-pending; the main file alone \
                 under-reports the true state; acquire the live -wal before the \
                 application terminates, because a checkpoint (e.g. on its next clean \
                 close) folds the WAL into the main file and discards the \
                 uncheckpointed deleted/superseded residue, which the \
                 post-checkpoint main file alone would no longer contain"
            ),
            AnomalyKind::PageCountMismatch {
                header_pages,
                file_pages,
            } => format!(
                "in-header page count ({header_pages}) disagrees with the file \
                 length ({file_pages} pages) — consistent with truncation, \
                 carving, or out-of-band modification"
            ),
            AnomalyKind::HotJournal { .. } => {
                "a hot rollback journal (valid header magic) sits beside \
                 the database — consistent with an interrupted or in-progress write transaction \
                 (crash, power loss, process kill, or acquisition captured mid-write); SQLite \
                 would roll it back on next open, so the main database may require rollback"
                    .to_string()
            }
            AnomalyKind::JournalRecoverable { images } => format!(
                "the rollback journal carries {images} pre-transaction page \
                 image(s) (PERSIST post-commit) — consistent with a committed \
                 transaction whose pre-images (deleted/modified rows) remain \
                 recoverable"
            ),
            AnomalyKind::JournalChecksumMismatch { pgnos, total } => {
                let list = pgnos
                    .iter()
                    .map(u32::to_string)
                    .collect::<Vec<_>>()
                    .join(", ");
                format!(
                    "{} of {total} journal page record(s) failed the page \
                     checksum (page(s) {list}) — consistent with corruption, a \
                     torn page write (power-loss mid-sector), or post-write \
                     modification of the journal",
                    pgnos.len()
                )
            }
            AnomalyKind::JournalSchemaChange {
                journal_cookie,
                db_cookie,
            } => format!(
                "the journal's prior schema cookie ({journal_cookie}) differs from the \
                 database's ({db_cookie}) — consistent with a DDL change (CREATE/DROP/ALTER) \
                 in the last transaction; the prior schema is recoverable"
            ),
            AnomalyKind::JournalDuplicatePage { .. } => {
                "a page number appears more than once across the \
                 journal's page records — the spec journals a page at most once, so this is \
                 consistent with corruption, a savepoint/super-journal artifact, or tampering"
                    .to_string()
            }
            AnomalyKind::JournalDbSizeDelta {
                mx_page,
                current_pages,
            } => {
                let direction = if *mx_page < *current_pages {
                    "the database grew (consistent with INSERTs adding pages)"
                } else {
                    "the database shrank (consistent with auto-vacuum / \
                     incremental-vacuum or truncation, NOT an ordinary DELETE)"
                };
                format!(
                    "the journal records a transaction-start page count \
                     (mxPage = {mx_page}) that differs from the current page \
                     count ({current_pages}) — the last transaction changed the \
                     database size: {direction}"
                )
            }
            AnomalyKind::FreelistCountInconsistent { declared, walked } => match walked {
                Some(walked) => format!(
                    "the header declares {declared} free page(s) (offset 36) but walking the \
                     freelist trunk chain yields {walked} — consistent with freelist tampering \
                     (an edited count to hide freed pages) or corruption"
                ),
                None => format!(
                    "the header declares {declared} free page(s) (offset 36) but the freelist \
                     trunk chain is unwalkable (out-of-range or cyclic trunk pointer) — \
                     consistent with freelist tampering or corruption"
                ),
            },
            AnomalyKind::ZeroedFreelistResidue {
                zeroed,
                free_leaves,
                first_zeroed,
            } => format!(
                "{zeroed} of {free_leaves} freed leaf page(s) are entirely zero (first at page \
                 {first_zeroed}) — a freed leaf keeps its former content, so the deleted records \
                 they held were overwritten with zeros; consistent with secure_delete=ON or a \
                 manual wipe"
            ),
        }
    }
}

/// A `SQLite` forensic anomaly: an observation graded by severity, with a stable
/// code and note derived from its [`AnomalyKind`] so they cannot drift.
#[derive(Debug, Clone, PartialEq)]
pub struct Anomaly {
    /// Severity, derived from `kind`.
    pub severity: Severity,
    /// Stable machine-readable code, derived from `kind`.
    pub code: &'static str,
    /// The classified anomaly.
    pub kind: AnomalyKind,
    /// Human-readable note, derived from `kind`.
    pub note: String,
    /// Heuristic confidence, present for inferential findings (e.g. a carved
    /// deleted record); `None` for structurally-certain header observations.
    pub confidence: Option<f32>,
}

impl Anomaly {
    /// Build an [`Anomaly`], deriving severity/code/note from `kind`.
    #[must_use]
    pub fn new(kind: AnomalyKind) -> Self {
        Anomaly {
            severity: kind.severity(),
            code: kind.code(),
            note: kind.note(),
            kind,
            confidence: None,
        }
    }

    /// Attach a heuristic confidence (used for carved deleted records).
    #[must_use]
    pub fn with_confidence(mut self, confidence: f32) -> Self {
        self.confidence = Some(confidence);
        self
    }
}

impl Observation for Anomaly {
    fn severity(&self) -> Option<Severity> {
        Some(self.severity)
    }
    fn code(&self) -> &'static str {
        self.code
    }
    fn note(&self) -> String {
        self.note.clone()
    }

    fn confidence(&self) -> Option<Confidence> {
        self.confidence.and_then(Confidence::new)
    }

    fn category(&self) -> forensicnomicon::report::Category {
        use forensicnomicon::report::Category;
        match &self.kind {
            // Deleted-record residue and free (deallocated) pages are recoverability
            // findings; the code keywords don't trip Category::from_code's Residue
            // classifier, so classify them explicitly.
            // Deleted-record residue and free pages are recoverability findings;
            // so are a recoverable PERSIST journal and a journaled schema page
            // (the prior schema/rows are recoverable from the journal images).
            AnomalyKind::DeletedRecordRecovered { .. }
            | AnomalyKind::DroppedSchemaRecovered { .. }
            | AnomalyKind::NonEmptyFreelist { .. }
            | AnomalyKind::JournalRecoverable { .. }
            | AnomalyKind::JournalSchemaChange { .. }
            // Zeroed freed leaves are a residue (recoverability) finding: the
            // residue that WOULD be recoverable was destroyed.
            | AnomalyKind::ZeroedFreelistResidue { .. } => Category::Residue,
            // A WAL-only/uncheckpointed state, a header/file page-count mismatch,
            // and the journal integrity observations (hot journal, checksum /
            // duplicate corruption, db-size delta) are integrity-of-state.
            AnomalyKind::WalUncheckpointedState { .. }
            | AnomalyKind::PageCountMismatch { .. }
            | AnomalyKind::HotJournal { .. }
            | AnomalyKind::JournalChecksumMismatch { .. }
            | AnomalyKind::JournalDuplicatePage { .. }
            | AnomalyKind::JournalDbSizeDelta { .. }
            // A header-vs-chain freelist inconsistency is integrity-of-state.
            | AnomalyKind::FreelistCountInconsistent { .. } => Category::Integrity,
            other => Category::from_code(other.code()),
        }
    }

    fn evidence(&self) -> Vec<Evidence> {
        match &self.kind {
            AnomalyKind::DeletedRecordRecovered {
                page,
                offset,
                rowid,
            } => vec![
                Evidence {
                    field: "rowid".to_string(),
                    value: rowid.to_string(),
                    location: Some(Location::RecordId(u64::try_from(*rowid).unwrap_or(0))),
                },
                Evidence {
                    field: "source_page".to_string(),
                    value: page.to_string(),
                    location: Some(Location::Other {
                        space: "sqlite:page".to_string(),
                        value: u64::from(*page),
                    }),
                },
                Evidence {
                    field: "cell_offset".to_string(),
                    value: offset.to_string(),
                    location: Some(Location::ByteOffset(*offset as u64)),
                },
            ],
            AnomalyKind::DroppedSchemaRecovered { object_type, name } => vec![
                Evidence {
                    field: "object_type".to_string(),
                    value: object_type.clone(),
                    location: None,
                },
                Evidence {
                    field: "name".to_string(),
                    value: name.clone(),
                    location: None,
                },
            ],
            AnomalyKind::NonEmptyFreelist { free_pages } => vec![Evidence {
                field: "free_pages".to_string(),
                value: free_pages.to_string(),
                location: None,
            }],
            AnomalyKind::WalUncheckpointedState { overlaid_pages } => vec![Evidence {
                field: "overlaid_pages".to_string(),
                value: overlaid_pages.to_string(),
                location: None,
            }],
            AnomalyKind::PageCountMismatch {
                header_pages,
                file_pages,
            } => vec![
                Evidence {
                    field: "header_pages".to_string(),
                    value: header_pages.to_string(),
                    location: Some(Location::Field("in_header_db_size".to_string())),
                },
                Evidence {
                    field: "file_pages".to_string(),
                    value: file_pages.to_string(),
                    location: None,
                },
            ],
            AnomalyKind::JournalRecoverable { images } => vec![Evidence {
                field: "page_images".to_string(),
                value: images.to_string(),
                location: None,
            }],
            AnomalyKind::JournalChecksumMismatch { pgnos, total } => {
                let mut ev: Vec<Evidence> = pgnos
                    .iter()
                    .map(|pgno| Evidence {
                        field: "checksum_mismatch_pgno".to_string(),
                        value: pgno.to_string(),
                        location: Some(Location::Other {
                            space: "sqlite:page".to_string(),
                            value: u64::from(*pgno),
                        }),
                    })
                    .collect();
                ev.push(Evidence {
                    field: "page_records".to_string(),
                    value: total.to_string(),
                    location: None,
                });
                ev
            }
            AnomalyKind::JournalDbSizeDelta {
                mx_page,
                current_pages,
            } => vec![
                Evidence {
                    field: "mx_page".to_string(),
                    value: mx_page.to_string(),
                    location: None,
                },
                Evidence {
                    field: "current_pages".to_string(),
                    value: current_pages.to_string(),
                    location: None,
                },
            ],
            AnomalyKind::JournalSchemaChange {
                journal_cookie,
                db_cookie,
            } => vec![
                Evidence {
                    field: "journal_schema_cookie".to_string(),
                    value: journal_cookie.to_string(),
                    location: None,
                },
                Evidence {
                    field: "db_schema_cookie".to_string(),
                    value: db_cookie.to_string(),
                    location: None,
                },
            ],
            AnomalyKind::NonZeroReservedSpace { reserved } => vec![Evidence {
                field: "reserved_bytes_per_page".to_string(),
                value: reserved.to_string(),
                // The reserved-space-per-page byte lives at file-header offset 20.
                location: Some(Location::ByteOffset(20)),
            }],
            AnomalyKind::JournalDuplicatePage { pgnos } => pgnos
                .iter()
                .map(|pgno| Evidence {
                    field: "duplicate_pgno".to_string(),
                    value: pgno.to_string(),
                    location: Some(Location::Other {
                        space: "sqlite:page".to_string(),
                        value: u64::from(*pgno),
                    }),
                })
                .collect(),
            AnomalyKind::HotJournal {
                mx_page,
                journaled_pages,
            } => vec![
                Evidence {
                    field: "db_page_count_at_txn_start".to_string(),
                    value: mx_page.to_string(),
                    location: None,
                },
                Evidence {
                    field: "journaled_pages".to_string(),
                    value: journaled_pages.to_string(),
                    location: None,
                },
            ],
            AnomalyKind::FreelistCountInconsistent { declared, walked } => vec![
                Evidence {
                    field: "declared_free_pages".to_string(),
                    value: declared.to_string(),
                    // The free-page count lives at file-header offset 36.
                    location: Some(Location::ByteOffset(36)),
                },
                Evidence {
                    field: "walked_free_pages".to_string(),
                    value: walked.map_or_else(|| "malformed".to_string(), |w| w.to_string()),
                    location: None,
                },
            ],
            AnomalyKind::ZeroedFreelistResidue {
                zeroed,
                free_leaves,
                first_zeroed,
            } => vec![
                Evidence {
                    field: "zeroed_leaf_pages".to_string(),
                    value: zeroed.to_string(),
                    location: None,
                },
                Evidence {
                    field: "free_leaf_pages".to_string(),
                    value: free_leaves.to_string(),
                    location: None,
                },
                Evidence {
                    field: "first_zeroed_page".to_string(),
                    value: first_zeroed.to_string(),
                    location: Some(Location::Other {
                        space: "sqlite:page".to_string(),
                        value: u64::from(*first_zeroed),
                    }),
                },
            ],
        }
    }
}

/// Which class of free space a deleted record was carved from. Records the
/// recovery provenance so the examiner can weigh reliability by class.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum RecoverySource {
    /// A whole page that was freed onto the freelist (the strongest case: the
    /// page holds only deallocated content).
    FreelistPage,
    /// The in-page free space (unallocated gap / freeblock slack) of a page that
    /// is still allocated. The record is genuinely deleted but more likely to be
    /// partially overwritten, so it is graded lower.
    InPageFreeBlock,
    /// A page whose table was `DROP`ped — on the freelist with no `sqlite_master`
    /// schema, so the column count was inferred from the record itself.
    DroppedTable,
    /// A **prior version** of a still-live row: an `UPDATE` freed the old version
    /// of the row into slack while the new version kept the same rowid. The
    /// recovered values DIFFER from the current live row (e.g. an edited message
    /// or a changed amount), so it is genuine deleted content — the edit history.
    PriorVersion,
    /// A record rebuilt by **freeblock reconstruction**: the freed cell's first
    /// four bytes (payload-length + rowid varints, `header_len`, and the leading
    /// serial type) were overwritten by SQLite's freeblock header, so the record
    /// was rebuilt from its surviving serial-type tail plus a schema template. The
    /// rowid is destroyed (surfaced as `0`), so this is the weakest in-page class
    /// — a low-confidence "consistent with a deleted row" lead.
    FreeblockReconstructed,
    /// Residue carved from an **uncheckpointed WAL frame's page image** rather
    /// than the on-disk pages. A `-wal` frame holds a committed page version the
    /// main file does not yet reflect; deleted cells freed within that version
    /// survive in the frame's slack and exist NOWHERE on disk. The record carries
    /// the `(salt1, salt2, frame_index)` log-sequence provenance in
    /// [`CarvedRecord::wal`].
    WalFrame,
    /// Residue carved from a **materialized commit snapshot** — the database state
    /// replayed up to one COMMIT frame of the `-wal` (base image ∪ committed frames
    /// to that commit). A row that is a live cell at this commit but deleted by a
    /// later commit survives ONLY in this snapshot's page images. The record
    /// carries the commit's `(salt1, salt2, commit_frame_index)` LSN in
    /// [`CarvedRecord::wal`] — the per-commit temporal coordinate, distinct from a
    /// raw [`RecoverySource::WalFrame`] residue's frame index.
    CommitSnapshot,
    /// A **prior allocated row from the rollback journal** (design §5). The row
    /// was LIVE in the pre-transaction state the `-journal` preserves and the last
    /// transaction deleted or modified it — it is NOT a free-space residue, so
    /// downstream reports must not relabel it as unallocated. Carries the page /
    /// segment / header-state / checksum provenance in [`RollbackJournalSource`].
    RollbackJournal(RollbackJournalSource),
}

/// Whether a rollback-journal header was parsed intact (Tier A) or reconstructed
/// from a zeroed (PERSIST post-commit) header (Tier B). A reconstructed header
/// could not verify page checksums (the nonce was zeroed), so its recoveries
/// rest on cross-record structural consistency — a graded distinction the
/// examiner weighs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JournalHeaderState {
    /// Tier A: header magic present; the checksum nonce was known and verified.
    Valid,
    /// Tier B: header zeroed (PERSIST post-commit); parameters reconstructed,
    /// checksums unverifiable.
    ReconstructedZeroed,
}

/// Provenance for a record recovered from a rollback journal (design §5):
/// where in the journal its page image lived and how trustworthy the parse was.
/// `Copy` (all fields are scalar) so it threads through the existing
/// attribution → output pipeline exactly like [`WalProvenance`]; the journal's
/// path is supplied once at the API boundary, not duplicated per row.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RollbackJournalSource {
    /// 0-based journal segment the page image came from.
    pub segment: usize,
    /// 1-based database page number the prior image restored.
    pub pgno: u32,
    /// Whether the header was intact (Tier A) or reconstructed (Tier B).
    pub header_state: JournalHeaderState,
    /// `Some(true/false)` when the page checksum could be verified (Tier A);
    /// `None` in Tier B (nonce zeroed).
    pub checksum_valid: Option<bool>,
}

/// Provenance for a record carved from a `-wal` frame: the
/// `(salt1, salt2, frame_index)` log-sequence identity of the frame it came from
/// (the LSN task #55 will formalize).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WalProvenance {
    /// 0-based position of the source frame within the `-wal` file.
    pub frame_index: usize,
    /// WAL header salt-1 (checkpoint generation) of the source frame.
    pub salt1: u32,
    /// WAL header salt-2 (checkpoint generation) of the source frame.
    pub salt2: u32,
}

/// Provenance for a record reassembled across an **overflow-page chain** (task
/// #73): the pages whose bytes were concatenated to recover the row. An examiner
/// citing the row as evidence can name exactly where its bytes came from. Present
/// only on chain-reassembled rows; `None` for every contiguous recovery.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OverflowProvenance {
    /// First overflow page of the chain (the page the local-prefix pointer named).
    pub first_page: u32,
    /// Ordered overflow pages whose content was assembled into the payload.
    pub chain: Vec<u32>,
}

/// A deleted record recovered from unallocated space — the headline capability
/// rusqlite cannot provide. Carries the decoded row plus provenance so the
/// examiner can weigh it as a "consistent with a deleted row" observation.
#[derive(Debug, Clone, PartialEq)]
pub struct CarvedRecord {
    /// 1-based page the record was carved from.
    pub page: u32,
    /// Byte offset of the cell within that page.
    pub offset: usize,
    /// Recovered rowid.
    pub rowid: i64,
    /// Decoded column values, in column order.
    pub values: Vec<Value>,
    /// Heuristic confidence in `(0.0, 1.0]` that these bytes are a real record.
    pub confidence: f32,
    /// Always `false`: a carved record lives in unallocated space, never in the
    /// live b-tree. Present so callers cannot mistake it for an allocated row.
    pub allocated: bool,
    /// Which class of free space this record was recovered from.
    pub source: RecoverySource,
    /// WAL log-sequence provenance, present **only** for
    /// [`RecoverySource::WalFrame`] records (the frame the residue was carved
    /// from); `None` for every on-disk class.
    pub wal: Option<WalProvenance>,
    /// Overflow-chain provenance, present **only** for rows reassembled across a
    /// freed overflow-page chain (task #73); `None` for every contiguous recovery.
    pub overflow: Option<OverflowProvenance>,
}

/// A **Tier-2 partial recovery**: a freed cell whose full row could not be
/// reconstructed but at least one *distinctive* cell (TEXT ≥ 4 bytes of valid
/// UTF-8, or REAL) survived at a structural anchor.
///
/// Deliberately NOT a [`CarvedRecord`]: it has no rowid (clobbered) and an
/// incomplete column set, and it does **not** share the full-row
/// 0-false-positive guarantee — it is a lead-generation surface with an expected
/// non-zero false-positive rate. The type system keeps it out of the full-row
/// output so a fragment can never be silently rendered as a recovered row
/// (secure by design). Returned only by [`carve_with_fragments`] in the opt-in
/// [`CarveTiers::fragments`] bucket. A fragment is "consistent with a partial
/// deleted row" — the examiner draws the conclusion.
#[derive(Debug, Clone, PartialEq)]
pub struct CarvedFragment {
    /// 1-based page the fragment was salvaged from.
    pub page: u32,
    /// Byte offset of the failed cell's anchor within that page.
    pub offset: usize,
    /// `(column_index, value)` for each column that decoded cleanly, ascending by
    /// index — meaningful against the table's column order.
    pub surviving: Vec<(usize, Value)>,
    /// Number of the row's columns that did NOT decode.
    pub missing: usize,
    /// Always the flat Tier-2 fragment confidence (0.2) for now — strictly below
    /// every full-row class.
    pub confidence: f32,
    /// Which class of free space the fragment was salvaged from
    /// ([`RecoverySource::FreeblockReconstructed`] for chain-pass fragments,
    /// [`RecoverySource::InPageFreeBlock`] for gap-pass fragments).
    pub source: RecoverySource,
    /// WAL log-sequence provenance. `None` in v1 (no WAL fragment pass yet).
    pub wal: Option<WalProvenance>,
}

/// The two strictly-separated recovery tiers returned by [`carve_with_fragments`].
///
/// `full` is **byte-identical** to [`carve_all_deleted_records`] and keeps its
/// structural 0-false-positive guarantee — the zero-config, high-precision
/// output. `fragments` is the opt-in Tier-2 lead surface (see [`CarvedFragment`]).
#[derive(Debug, Clone, PartialEq)]
pub struct CarveTiers {
    /// Tier-1 full rows — identical to [`carve_all_deleted_records`].
    pub full: Vec<CarvedRecord>,
    /// Tier-2 partial recoveries (opt-in; expected non-zero false-positive rate).
    pub fragments: Vec<CarvedFragment>,
}

/// Recover deleted records by carving the database's free (unallocated) pages.
///
/// Each free page from [`Database::freelist_pages`] is scanned with
/// [`Database::carve_cells`] for record-shaped cells of `column_count` columns.
/// Free pages hold only deallocated content, so the recovered rows are deleted
/// ones — the carver never re-surfaces a live (allocated) row. Recovered rows
/// are **confidence-graded observations**, not certainties: a carved record is
/// "consistent with a deleted row", and the examiner draws the conclusion.
///
/// Read-only and panic-free: a malformed freelist simply yields fewer (or no)
/// carved records rather than an error.
#[must_use]
pub fn carve_deleted_records(db: &Database, column_count: usize) -> Vec<CarvedRecord> {
    let mut out = Vec::new();
    let Ok(free) = db.freelist_pages() else {
        return out;
    };
    for page in free {
        let Some(page_bytes) = db.raw_page(page) else {
            continue; // cov:unreachable: freelist_pages only yields in-range pages
        };
        for cell in db.carve_cells(&page_bytes, column_count) {
            out.push(CarvedRecord {
                page,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::FreelistPage,
                wal: None,
                overflow: None,
            });
        }
    }
    out
}

/// Every currently-live row's decoded values, across ALL tables, UNCOLLAPSED.
///
/// [`Database::live_rows`] keys live-row identity by a global rowid, so two tables
/// that share a rowid collapse to a single entry — a live row can then vanish from
/// the exclusion guard and be re-surfaced as deleted (roadmap §1.1). This walks
/// `live_table_rows()` (every row, per table) so the live-row guard is complete
/// even under cross-table rowid collisions. Best-effort and panic-free.
fn live_value_tuples(db: &Database) -> Vec<Vec<Value>> {
    db.live_table_rows()
        .into_iter()
        .flat_map(|t| t.rows.into_iter().map(|r| r.values))
        .collect()
}

/// Recover deleted records across **every** free-space class — the full-coverage
/// carver. Drives, in order:
///
/// 1. **Freelist pages** (whole pages freed onto the freelist), with the column
///    count inferred per record so it also recovers **dropped-table** pages whose
///    `sqlite_master` schema is gone (e.g. a `DROP TABLE` left the page on the
///    freelist with no recorded column count).
/// 2. **In-page free space** of still-allocated table-leaf pages (the unallocated
///    gap and inter-cell slack), via [`Database::carve_free_regions`], which
///    carves only the complement of the live cells — so a live (allocated) row is
///    **never** re-surfaced (the 0-false-positive guarantee, enforced
///    structurally).
///
/// Records are de-duplicated by `(rowid, values)` keeping the highest-confidence
/// copy, since a row can survive in more than one place. Every record is graded:
/// freelist-page recovery highest, dropped-table next, in-page residue lowest.
///
/// Read-only and panic-free; a malformed structure yields fewer records, never an
/// error or panic.
#[must_use]
pub fn carve_all_deleted_records(db: &Database) -> Vec<CarvedRecord> {
    let mut out: Vec<CarvedRecord> = Vec::new();

    // (1) Freelist pages, inferring the column count per record. Inference makes
    // this recover both normal freed pages AND schema-gone dropped-table pages
    // (a `DROP TABLE` leaves the page on the freelist with no recorded column
    // count). When the database has no live user table at all, the freed content
    // is necessarily from a dropped table, so mark those records accordingly.
    let dropped_table_db = !db.has_user_table();
    if let Ok(free) = db.freelist_pages() {
        for page in free {
            let Some(page_bytes) = db.raw_page(page) else {
                continue; // cov:unreachable: freelist_pages yields in-range pages
            };
            let source = if dropped_table_db {
                RecoverySource::DroppedTable
            } else {
                RecoverySource::FreelistPage
            };
            for cell in db.carve_cells_inferred(&page_bytes) {
                out.push(CarvedRecord {
                    page,
                    offset: cell.offset,
                    rowid: cell.rowid,
                    values: cell.values,
                    confidence: cell.confidence,
                    allocated: false,
                    source,
                    wal: None,
                    overflow: None,
                });
            }
        }
    }

    // (2) In-page free space of every still-allocated table-leaf page.
    let page_count = db.page_count();
    for page in 1..=page_count {
        let Some(page_bytes) = db.raw_page(page) else {
            continue; // cov:unreachable: 1..=page_count is in range
        };
        for cell in db.carve_free_regions(&page_bytes, 0) {
            out.push(CarvedRecord {
                page,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::InPageFreeBlock,
                wal: None,
                overflow: None,
            });
        }
        // (2b) Freeblock reconstruction: the freed cells whose first four bytes
        // were clobbered by freeblock conversion, rebuilt from their surviving
        // serial tail plus the page's schema template. These carry an unknown
        // (destroyed) rowid, so the value-collision pass below — not the
        // rowid-keyed filter — is what guarantees no live row is re-surfaced.
        for cell in db.reconstruct_freeblock_records(&page_bytes) {
            out.push(CarvedRecord {
                page,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::FreeblockReconstructed,
                wal: None,
                overflow: None,
            });
        }
        // (2c) Chain-aware overflow recovery (task #73): a freed cell whose
        // payload spilled onto an overflow-page chain, reassembled when every
        // chain page survives as a freelist leaf. Graded below the in-page
        // full-row tier (NOT part of the structural 0-FP guarantee — Codex
        // ruling #1); a broken chain (e.g. a trunk-clobbered page) is rejected
        // here and degrades to a Tier-2 fragment elsewhere.
        for (cell, chain) in db.carve_overflow_records(&page_bytes) {
            let first_page = chain.first().copied().unwrap_or(0);
            out.push(CarvedRecord {
                page,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::InPageFreeBlock,
                wal: None,
                overflow: Some(OverflowProvenance { first_page, chain }),
            });
        }
        // (2d) Freeblock-clobbered spilled cells (task #73, Codex ruling #5;
        // UNPROVEN-BY-CORPUS — synthetic validation only). A spilled cell whose
        // prefix was also freeblock-clobbered: P re-derived from the surviving
        // serial array, chain resolved through freelist leaves, rowid destroyed.
        for (cell, chain) in db.carve_overflow_template_records(&page_bytes) {
            let first_page = chain.first().copied().unwrap_or(0);
            out.push(CarvedRecord {
                page,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::FreeblockReconstructed,
                wal: None,
                overflow: Some(OverflowProvenance { first_page, chain }),
            });
        }
    }

    // (3) WAL-frame carving (additive — runs ONLY when a `-wal` overlay is in
    // effect). The on-disk path above reads only the main file's pages, so it
    // finds the SAME residue whether or not a WAL was supplied; the genuinely-
    // different deleted records live in the uncheckpointed WAL frames.
    //
    // A `-wal` frame is a FULL page snapshot at one point in the transaction
    // history. A row deleted late in that history is still a live cell in an
    // EARLIER frame's image; it exists nowhere on disk and in no later frame.
    // Three primitives over each committed frame's page image surface it:
    //
    //   * `carve_leaf_cells` — every cell the frame records as allocated. A cell
    //     that is allocated in a superseded frame but ABSENT from the final
    //     WAL-applied live view is exactly such a deleted row (recovered as a
    //     clean, intact record — the strongest WAL case).
    //   * `carve_free_regions` / `reconstruct_freeblock_records` — residue freed
    //     WITHIN a frame (e.g. the DELETE-commit frame's own freeblocks), matching
    //     the on-disk in-page classes.
    //
    // Every candidate is tagged `WalFrame` with the (salt1, salt2, frame_index)
    // LSN provenance, and the shared live-row precision filter below drops any
    // whose values match a currently-live row — so a surviving row is never
    // re-surfaced (the 0-false-positive guarantee, against the WAL-applied view).
    for frame in db.wal_frame_pages() {
        let prov = WalProvenance {
            frame_index: frame.frame_index,
            salt1: frame.salt1,
            salt2: frame.salt2,
        };
        let cells = db
            .carve_leaf_cells(&frame.page)
            .into_iter()
            .chain(db.carve_free_regions(&frame.page, 0))
            .chain(db.reconstruct_freeblock_records(&frame.page));
        for cell in cells {
            out.push(CarvedRecord {
                page: frame.page_no,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::WalFrame,
                wal: Some(prov),
                overflow: None,
            });
        }
    }

    // VALUE-AWARE classification for carved records whose rowid is currently
    // LIVE. Two very different cases share a live rowid:
    //
    //   * Stale rebalance copy — a b-tree rebalance moved a still-live row to
    //     another page, leaving a byte-identical copy in the old page's slack.
    //     SAME rowid, SAME values → NOT deleted → drop (the 0-false-positive
    //     precision win we must preserve).
    //   * Prior version — an UPDATE freed the OLD version of the row into slack
    //     while the new version kept the same rowid. SAME rowid, DIFFERENT values
    //     → genuinely-deleted content (the edited message / changed amount, often
    //     THE evidence) → recover, tagged `PriorVersion`.
    //
    // A rowid-only filter cannot tell these apart and drops both (a false
    // negative on prior versions). Comparing decoded values does.
    // Structural-noise rejection (precision, not the live-row guarantee): drop
    // any carved record that carries no content — a rowid echo followed by an
    // all-NULL tail, the inferred over-read of free-space zero bytes. Applied to
    // every tier uniformly (no per-source special-case).
    out.retain(|rec| !is_structural_noise(&rec.values));

    let live = db.live_rows();
    // COMPLETE value-level identity of every live row, across ALL tables and
    // UNCOLLAPSED. `live_rows()` keys by a global rowid, so two tables sharing a
    // rowid collapse to one entry and a live row can vanish from the guard; build
    // the guard from every live row (roadmap §1.1) plus the CURRENT `sqlite_master`
    // rows (a record carved from a materialized page 1 whose values equal a live
    // schema entry is that live row re-surfaced, not deleted residue — value-based,
    // so a genuinely-deleted PRIOR schema version is still recovered).
    let live_values = live_value_tuples(db);
    let live_value_keys: std::collections::HashSet<String> = live_values
        .iter()
        .chain(db.live_schema_rows().iter())
        .map(|v| format!("{v:?}"))
        .collect();
    out.retain_mut(|rec| {
        // Exclusion invariant (EVERY source): a record whose decoded values equal
        // ANY currently-live row is that live row re-surfaced — never report it as
        // deleted. This guards the cross-table rowid-collision case the rowid-keyed
        // branch below cannot, since the global live map collapses shared rowids.
        if live_value_keys.contains(&format!("{:?}", rec.values)) {
            return false;
        }
        // Freeblock reconstructions (destroyed rowid) and WAL-frame residue carry no
        // rowid usable by the branch below; the value guard above already protected
        // them, so keep them with their source tag intact.
        if rec.source == RecoverySource::FreeblockReconstructed
            || rec.source == RecoverySource::WalFrame
        {
            return true;
        }
        // A live rowid with DIFFERING values (exact matches were dropped above) is a
        // deleted prior version of that row; an absent rowid is an ordinary deletion.
        // Either way the record is kept — only its classification changes.
        if live.contains_key(&rec.rowid) {
            rec.source = RecoverySource::PriorVersion;
        }
        true
    });

    dedup_keep_best(out, &db.page_to_table_map())
}

/// A carved full record is **structural noise** — not an information-bearing
/// deleted row — when every column past the first is NULL. The inferred carver,
/// lacking a column hint, can read a run of free-space zero bytes as a long
/// serial-type-0 (NULL) array; that surfaces as a rowid echo (the INTEGER-PK
/// column) followed by an all-NULL tail, carrying no recoverable content. A
/// genuine row — live, dropped-table, or overflow — has a non-NULL value in some
/// non-leading column.
///
/// The column count is deliberately NOT bounded against the live tables: a
/// dropped table can legitimately be wider than every surviving one, so a
/// width cap would discard genuine unattributed records. Requiring length >= 2
/// keeps a real single-column row from being mistaken for noise.
fn is_structural_noise(values: &[Value]) -> bool {
    values.len() >= 2 && values.iter().skip(1).all(|v| matches!(v, Value::Null))
}

/// A schema-table (`sqlite_master`) row recovered from free space — a **dropped
/// or replaced** table/index/view/trigger whose definition no longer appears in
/// the live schema. Recovering it tells the examiner an object *existed* and its
/// structure (the `CREATE` statement), which in turn explains residual data on
/// the freelist. A finding, not a certainty: it is "consistent with a dropped
/// `<name>`" — the examiner draws the conclusion.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecoveredSchema {
    /// `sqlite_master.type` — `table`, `index`, `view`, or `trigger`.
    pub object_type: String,
    /// `sqlite_master.name` — the object's name.
    pub name: String,
    /// `sqlite_master.tbl_name` — the table the object belongs to (== `name` for
    /// a table).
    pub tbl_name: String,
    /// `sqlite_master.rootpage` — the (now-freed) b-tree root, `None` when the
    /// recovered cell stored it as NULL (views/triggers).
    pub rootpage: Option<i64>,
    /// `sqlite_master.sql` — the recovered `CREATE` statement.
    pub sql: String,
}

/// Recover **dropped or replaced** schema definitions from free space: carved
/// `sqlite_master`-shaped rows whose `(name, sql)` no longer matches a live
/// schema entry. A `DROP TABLE`/`DROP INDEX` deletes the object's `sqlite_master`
/// row (its bytes survive in page-1 free space under `secure_delete=OFF`); this
/// surfaces that residue as a structured [`RecoveredSchema`] rather than a raw
/// 5-column carved record.
#[must_use]
pub fn recover_dropped_schemas(db: &Database) -> Vec<RecoveredSchema> {
    use std::collections::HashSet;
    // Live `(name, sql)` pairs — an exactly-live definition carved from page-1
    // residue is a re-surfaced live row, never a drop.
    let live: HashSet<(String, String)> = db.schema_sql().into_iter().collect();
    let mut seen: HashSet<(String, String)> = HashSet::new();
    let mut out = Vec::new();
    for rec in carve_all_deleted_records(db) {
        // A `sqlite_master` row lives only on page 1 (plus its overflow); restrict
        // there so a 5-column data record elsewhere is never mistaken for schema.
        if rec.page != 1 {
            continue;
        }
        let Some(schema) = as_schema_row(&rec.values) else {
            continue;
        };
        let key = (schema.name.clone(), schema.sql.clone());
        if live.contains(&key) {
            continue;
        }
        if seen.insert(key) {
            out.push(schema);
        }
    }
    out
}

/// Interpret a carved record as a `sqlite_master` row
/// `(type, name, tbl_name, rootpage, sql)`, or `None` if it does not match that
/// shape. `type` must be one of the four schema object kinds, the `name` /
/// `tbl_name` / `sql` columns TEXT, and `rootpage` an INTEGER (or NULL for
/// views/triggers).
fn as_schema_row(values: &[Value]) -> Option<RecoveredSchema> {
    let [type_v, name_v, tbl_v, root_v, sql_v] = values else {
        return None;
    };
    let object_type = match type_v {
        Value::Text(t) if matches!(t.as_str(), "table" | "index" | "view" | "trigger") => t.clone(),
        _ => return None,
    };
    let (Value::Text(name), Value::Text(tbl_name), Value::Text(sql)) = (name_v, tbl_v, sql_v)
    else {
        return None;
    };
    let rootpage = match root_v {
        Value::Integer(n) => Some(*n),
        Value::Null => None,
        _ => return None,
    };
    Some(RecoveredSchema {
        object_type,
        name: name.clone(),
        tbl_name: tbl_name.clone(),
        rootpage,
        sql: sql.clone(),
    })
}

/// Two-tier deleted-record recovery: Tier-1 full rows **plus** Tier-2 partial
/// fragments, in one pass.
///
/// [`CarveTiers::full`] is byte-identical to [`carve_all_deleted_records`] (the
/// high-precision, structurally-0-false-positive output). [`CarveTiers::fragments`]
/// is the opt-in lead surface: at every freeblock/gap anchor where full
/// reconstruction failed but ≥ 1 distinctive cell (TEXT ≥ 4 bytes of valid UTF-8,
/// or REAL) survived, the maximal decodable column prefix is salvaged as a
/// [`CarvedFragment`]. Three suppression layers keep the fragment bucket honest:
///
/// 1. **By construction** — a fragment is emitted only where full reconstruction
///    failed, so an anchor yields a full cell or a fragment, never both.
/// 2. **Full-row value suppression** — a fragment whose every surviving
///    `(column, value)` matches the corresponding column of a Tier-1 record in
///    `full` is dropped (the row was already recovered another way).
/// 3. **Live-row suppression** — a fragment whose every surviving `(column, value)`
///    matches the corresponding columns of a currently-live row is dropped (never
///    re-surface a live row, the fragment analog of the rebalance-copy drop).
///
/// Read-only and panic-free, identically to [`carve_all_deleted_records`].
#[must_use]
pub fn carve_with_fragments(db: &Database) -> CarveTiers {
    let full = carve_all_deleted_records(db);

    // Salvage Tier-2 fragments from every on-disk table-leaf page. The core
    // walker emits a fragment only where full reconstruction failed (layer 1).
    let mut fragments: Vec<CarvedFragment> = Vec::new();
    let page_count = db.page_count();
    for page in 1..=page_count {
        let Some(page_bytes) = db.raw_page(page) else {
            continue; // cov:unreachable: 1..=page_count is in range
        };
        for frag in db.reconstruct_freeblock_fragments(&page_bytes) {
            fragments.push(CarvedFragment {
                page,
                offset: frag.offset,
                surviving: frag.surviving,
                missing: frag.missing,
                confidence: frag.confidence,
                source: RecoverySource::FreeblockReconstructed,
                wal: None,
            });
        }
        // Broken-chain overflow fragments (task #73, Codex ruling #4): a spilled
        // cell whose overflow chain was destroyed (e.g. a trunk-clobbered chain
        // page) still has an intact local prefix; salvage its locally-decodable
        // columns. The chain-resident columns are lost (untrusted), so this is a
        // Tier-2 lead, never a full row.
        for frag in db.carve_overflow_fragments(&page_bytes) {
            fragments.push(CarvedFragment {
                page,
                offset: frag.offset,
                surviving: frag.surviving,
                missing: frag.missing,
                confidence: frag.confidence,
                source: RecoverySource::InPageFreeBlock,
                wal: None,
            });
        }
    }

    // Layer 3: live-row suppression. A fragment whose surviving set matches the
    // corresponding columns of a live row is a stale partial copy of a live row.
    // Complete, uncollapsed live set (roadmap §1.1): see `live_value_tuples`.
    let live_values = live_value_tuples(db);
    fragments.retain(|frag| {
        !live_values
            .iter()
            .any(|lv| fragment_matches_columns(frag, lv))
    });

    // Layer 2: full-row value suppression. A fragment already covered by a
    // recovered full row in this carve is a duplicate — drop it.
    fragments.retain(|frag| {
        !full
            .iter()
            .any(|rec| fragment_matches_columns(frag, &rec.values))
    });

    // Fragment dedup: one fragment per (page, offset) anchor by construction,
    // then collapse value-level duplicates keeping the copy with more surviving
    // columns (mirrors dedup_keep_best for the full tier).
    fragments = dedup_fragments(fragments);

    CarveTiers { full, fragments }
}

/// Whether a fragment's every surviving `(column_index, value)` pair equals the
/// corresponding column of `row` — the projection test shared by the full-row
/// (layer 2) and live-row (layer 3) suppression passes. Equality of the **whole**
/// surviving set is required: a fragment that coincides with a row on only some
/// cells is kept (it may be genuine residue), mirroring the full-row rule's
/// whole-values comparison.
fn fragment_matches_columns(frag: &CarvedFragment, row: &[Value]) -> bool {
    !frag.surviving.is_empty()
        && frag
            .surviving
            .iter()
            .all(|(idx, v)| row.get(*idx) == Some(v))
}

/// De-duplicate fragments: first by `(page, offset)` anchor (one fragment per
/// anchor by construction), then collapse value-level duplicates keeping the copy
/// with the most surviving columns. Mirrors [`dedup_keep_best`] for the full tier.
fn dedup_fragments(mut frags: Vec<CarvedFragment>) -> Vec<CarvedFragment> {
    use std::collections::HashSet;
    // More surviving columns first, so the kept copy of each identity is richest.
    frags.sort_by_key(|f| std::cmp::Reverse(f.surviving.len()));
    let mut seen_anchor: HashSet<(u32, usize)> = HashSet::new();
    let mut seen_values: HashSet<String> = HashSet::new();
    let mut kept = Vec::new();
    for frag in frags {
        if !seen_anchor.insert((frag.page, frag.offset)) {
            continue;
        }
        let vkey = format!("{:?}", frag.surviving);
        if !seen_values.insert(vkey) {
            continue;
        }
        kept.push(frag);
    }
    kept
}

/// Carve the deleted residue of **one materialized commit snapshot** of the
/// `-wal`, the per-commit temporal building block of the N-snapshot carve.
///
/// `id` addresses a [`CommitId`] in `timeline`; the snapshot it resolves to is the
/// database state replayed up to that COMMIT (base image ∪ every committed frame to
/// that commit, capped to `db_size_after_commit`). This runs the same three
/// carving primitives the on-disk path uses — [`Database::carve_leaf_cells`]
/// (intact cells the snapshot records as allocated, the strongest case),
/// [`Database::carve_free_regions`], and [`Database::reconstruct_freeblock_records`]
/// — over each of the snapshot's materialized page images, then applies the SAME
/// live-row precision filter: a record whose decoded values match a currently-live
/// row (the WAL-applied view) is dropped, so a row that survived to the final state
/// is **never** re-surfaced as deleted. Surviving records are tagged
/// [`RecoverySource::CommitSnapshot`] and carry the commit's
/// `(salt1, salt2, commit_frame_index)` LSN in [`CarvedRecord::wal`].
///
/// An unknown [`CommitId`] (absent from `timeline`) yields an empty vector, never a
/// panic. Read-only throughout.
#[must_use]
pub fn carve_at_commit(db: &Database, timeline: &WalTimeline, id: CommitId) -> Vec<CarvedRecord> {
    let Some(snapshot) = timeline.snapshot_at(id) else {
        return Vec::new();
    };
    let lsn = snapshot.lsn();
    let prov = WalProvenance {
        frame_index: lsn.frame_index,
        salt1: lsn.salt1,
        salt2: lsn.salt2,
    };

    let mut out: Vec<CarvedRecord> = Vec::new();
    for page_no in snapshot.page_numbers() {
        let Some(image) = snapshot.page_version(page_no) else {
            continue; // cov:unreachable: page_numbers only yields materialized pages
        };
        let cells = db
            .carve_leaf_cells(&image.bytes)
            .into_iter()
            .chain(db.carve_free_regions(&image.bytes, 0))
            .chain(db.reconstruct_freeblock_records(&image.bytes));
        for cell in cells {
            out.push(CarvedRecord {
                page: page_no,
                offset: cell.offset,
                rowid: cell.rowid,
                values: cell.values,
                confidence: cell.confidence,
                allocated: false,
                source: RecoverySource::CommitSnapshot,
                wal: Some(prov),
                overflow: None,
            });
        }
    }

    // Live-row precision filter (the WAL-applied view): drop any record whose
    // decoded values match a currently-live row, so a row that survives to the
    // final state is never re-surfaced as "deleted at an earlier commit". The
    // live set includes the CURRENT `sqlite_master` rows, because a snapshot's
    // materialized page 1 (the schema table) still holds the live schema cell —
    // value-based, so a deleted PRIOR schema version is still recovered.
    // Complete, uncollapsed live set (roadmap §1.1): see `live_value_tuples`.
    let live_value_keys: std::collections::HashSet<String> = live_value_tuples(db)
        .iter()
        .chain(db.live_schema_rows().iter())
        .map(|v| format!("{v:?}"))
        .collect();
    out.retain(|rec| !live_value_keys.contains(&format!("{:?}", rec.values)));

    dedup_keep_best(out, &db.page_to_table_map())
}

/// De-duplicate carved records by content identity, keeping the
/// highest-confidence copy of each (a row can survive in several free regions).
///
/// `Value` carries an `f64` (`Real`) so it is not `Hash`/`Eq`; the identity key
/// is the record's `rowid` plus a stable `Debug` rendering of its values, which
/// is sufficient to collapse byte-identical recoveries.
fn dedup_keep_best(
    mut records: Vec<CarvedRecord>,
    page_table: &std::collections::BTreeMap<u32, String>,
) -> Vec<CarvedRecord> {
    use std::collections::HashSet;
    let content = |r: &CarvedRecord| format!("{}:{:?}", r.rowid, r.values);

    // Identities (rowid, values) that have at least one record attributed to a
    // LIVE table. A copy of such an identity carved from a freed/unattributed page
    // is the SAME deleted row surviving in two places; it is folded into the
    // attributed copy rather than double-listed (roadmap §1.2).
    let attributed: HashSet<String> = records
        .iter()
        .filter(|r| page_table.contains_key(&r.page))
        .map(&content)
        .collect();

    let mut seen: HashSet<String> = HashSet::new();
    let mut kept: Vec<CarvedRecord> = Vec::new();
    // Highest confidence first, so the kept copy of each identity is the best one.
    records.sort_by(|a, b| {
        b.confidence
            .partial_cmp(&a.confidence)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    for rec in records.drain(..) {
        let c = content(&rec);
        // Table-aware identity (roadmap §1.2): a record from a page a live table
        // owns is keyed by (table, rowid, values) — the same row recovered from
        // several free regions of ONE table collapses, but two DIFFERENT live
        // tables sharing (rowid, values) stay distinct. An unattributed copy of an
        // identity that IS attributed elsewhere is the same deleted row → fold it
        // in. Otherwise (no live-table copy at all) keep the plain content identity.
        let key = match page_table.get(&rec.page) {
            Some(table) => format!("T:{table}:{c}"),
            None if attributed.contains(&c) => continue,
            None => format!("U:{c}"),
        };
        if seen.insert(key) {
            kept.push(rec);
        }
    }
    kept
}

/// A row recovered from the prior (pre-transaction) snapshot of a rollback
/// journal (design §5). In the prior state it was a LIVE allocated row that the
/// last transaction DELETED — so it is recovered at full fidelity and is NOT a
/// free-space carve (`allocated` was true in the prior state). Carries full
/// journal provenance in [`RollbackJournalSource`].
#[derive(Debug, Clone, PartialEq)]
pub struct PriorRow {
    /// The table the row belonged to (prior schema).
    pub table: String,
    /// The row's rowid (its INTEGER PRIMARY KEY for a rowid table).
    pub rowid: i64,
    /// The decoded prior column values, in column order.
    pub values: Vec<Value>,
    /// Journal provenance: page / segment / header-state / checksum status.
    pub source: RecoverySource,
}

/// A row present in BOTH the prior snapshot and the live db with DIFFERENT values
/// — the last transaction modified it (design §4). Carries the PRIOR (pre-edit)
/// values and the CURRENT values. `replaced_rowid` is set because a delete +
/// insert reusing the same rowid is indistinguishable from an in-place UPDATE
/// here, so identity may differ — never asserted as "UPDATE".
#[derive(Debug, Clone, PartialEq)]
pub struct PriorVersionRecord {
    /// The table the row belonged to (prior schema).
    pub table: String,
    /// The rowid present in both states.
    pub rowid: i64,
    /// The decoded PRIOR (pre-transaction) column values.
    pub prior_values: Vec<Value>,
    /// The decoded CURRENT (live) column values.
    pub current_values: Vec<Value>,
    /// Always `true`: the rowid is shared but identity may differ (delete+insert
    /// reusing a rowid is indistinguishable from an UPDATE), so the prior value is
    /// edit history, not an asserted in-place update.
    pub replaced_rowid: bool,
    /// Journal provenance for the prior image.
    pub source: RecoverySource,
}

/// Structured counts of what the rollback-journal recovery found — the NIST
/// SFT-03 "number of deleted/modified records" report (design §5/§6.11).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct JournalCounts {
    /// Rows deleted by the last transaction (prior \ current by rowid).
    pub deleted: usize,
    /// Rows modified by the last transaction (present in both, values differ).
    pub modified: usize,
}

/// The result of recovering deletions + modifications from a rollback journal
/// (design §5). Deleted rows are full-fidelity prior allocated rows (NOT
/// free-space carves); modified rows carry the prior and current values.
#[derive(Debug, Clone, PartialEq)]
pub struct JournalRecovery {
    /// Rows present in the prior snapshot but absent now — deleted by the last
    /// transaction, recovered at full fidelity.
    pub deleted: Vec<PriorRow>,
    /// Rows present in both, values differ — prior (pre-edit) + current values.
    pub modified: Vec<PriorVersionRecord>,
    /// Structured counts (the NIST source-file report).
    pub counts: JournalCounts,
}

/// Recover the last transaction's **deletions and modifications** from a rollback
/// `-journal` by diffing the pre-transaction snapshot against the live database
/// (design §4) — the temporal inverse of WAL recovery.
///
/// For each prior rowid table, the prior rows (read through `PriorSnapshot`)
/// are diffed against the current rows by rowid:
/// - **deleted** = rowid in prior but not current → a full-fidelity [`PriorRow`];
/// - **modified** = rowid in both with different values → a [`PriorVersionRecord`]
///   flagged `replaced_rowid` (identity may differ; never asserted as UPDATE);
/// - **inserted** (rowid only in current) is a timeline fact, not a recovery target.
///
/// `WITHOUT ROWID` tables are excluded (no integer rowid key; design §4 limit).
/// Read-only and panic-free: a malformed/truncated journal, or a WAL-applied db
/// (the modes are exclusive), yields an empty recovery rather than an error or
/// panic — the journal source is bound to THIS database via
/// [`Database::rollback_prior`].
#[must_use]
pub fn carve_rollback_journal(db: &Database, journal: &[u8]) -> JournalRecovery {
    let mut deleted = Vec::new();
    let mut modified = Vec::new();

    let Ok(prior) = db.rollback_prior(journal) else {
        return JournalRecovery {
            deleted,
            modified,
            counts: JournalCounts::default(),
        };
    };
    // Header state + checksum status are uniform across a single journal; carry
    // them on every recovered row's provenance.
    let Some(header_state) = prior_header_state(journal, db) else {
        // cov:unreachable: rollback_prior above already parsed the same journal,
        // so a re-parse here cannot fail; the arm degrades to an empty recovery.
        return JournalRecovery {
            deleted,
            modified,
            counts: JournalCounts::default(),
        };
    };
    // Per-row checksum status is not threaded through PriorSnapshot in v1, so the
    // verifiable fact recorded here is the journal-level one: Tier A could verify
    // checksums (nonce known), Tier B could not (nonce zeroed). A finer per-image
    // status is a later enhancement; `header_state` carries the Tier either way.
    let checksum_valid = match header_state {
        JournalHeaderState::Valid => Some(true),
        JournalHeaderState::ReconstructedZeroed => None,
    };

    // Index the current (live) rows per table by rowid, so the diff is O(rows).
    let live = db.live_table_rows();

    for ptable in prior.tables() {
        if ptable.without_rowid {
            continue; // design §4 limit: WITHOUT ROWID excluded from the rowid diff.
        }
        let col_count = ptable.columns.len();
        let Ok(prior_rows) = prior.read_table_with_pages(ptable.rootpage, col_count) else {
            continue;
        };
        // Current rows of the same-named table (live schema), by rowid.
        let current: std::collections::BTreeMap<i64, Vec<Value>> = live
            .iter()
            .find(|d| d.name == ptable.name)
            .map(|d| d.rows.iter().map(|r| (r.rowid, r.values.clone())).collect())
            .unwrap_or_default();

        for (rowid, values, leaf_page) in prior_rows {
            let source = RecoverySource::RollbackJournal(RollbackJournalSource {
                segment: 0,
                pgno: leaf_page,
                header_state,
                checksum_valid,
            });
            match current.get(&rowid) {
                None => deleted.push(PriorRow {
                    table: ptable.name.clone(),
                    rowid,
                    values,
                    source,
                }),
                Some(cur) if *cur != values => modified.push(PriorVersionRecord {
                    table: ptable.name.clone(),
                    rowid,
                    prior_values: values,
                    current_values: cur.clone(),
                    replaced_rowid: true,
                    source,
                }),
                Some(_) => {} // unchanged: the page was journaled for another row.
            }
        }
    }

    let counts = JournalCounts {
        deleted: deleted.len(),
        modified: modified.len(),
    };
    JournalRecovery {
        deleted,
        modified,
        counts,
    }
}

/// Determine whether the journal header was intact (Tier A) or zeroed (Tier B),
/// by re-parsing it against the db's authoritative page size.
fn prior_header_state(journal: &[u8], db: &Database) -> Option<JournalHeaderState> {
    let parsed = RollbackJournal::parse(journal, db.header().page_size).ok()?;
    Some(match parsed.header() {
        JournalHeader::Valid { .. } => JournalHeaderState::Valid,
        JournalHeader::ReconstructedZeroed { .. } => JournalHeaderState::ReconstructedZeroed,
    })
}

/// Audit an opened [`Database`] for forensically-notable anomalies.
///
/// Covers the header reserved-space field, a non-empty freelist (prior
/// deletions), an active WAL overlay (uncheckpointed state), and a header/file
/// page-count mismatch. Deleted-record recovery is offered separately via
/// [`carve_deleted_records`] / [`audit_carved_findings`] because it requires the
/// table's column count.
#[must_use]
pub fn audit(db: &Database) -> Vec<Anomaly> {
    let mut out = Vec::new();

    let reserved = db.header().reserved;
    if reserved != 0 {
        out.push(Anomaly::new(AnomalyKind::NonZeroReservedSpace { reserved }));
    }

    let free_pages = db.freelist_count();
    if free_pages != 0 {
        out.push(Anomaly::new(AnomalyKind::NonEmptyFreelist { free_pages }));
    }

    // Freelist structural consistency + residue-destruction fingerprint (§2.1).
    // The header count at offset 36 is a *claim*; walking the trunk chain verifies
    // it. Freed *leaf* pages are content-preserving, so an all-zero freed leaf is
    // the fingerprint of destroyed residue (secure_delete / a wipe).
    let declared = free_pages;
    match db.freelist_pages_split() {
        Ok((leaves, trunks)) => {
            let walked = u32::try_from(leaves.len() + trunks.len()).unwrap_or(u32::MAX);
            if declared != walked {
                out.push(Anomaly::new(AnomalyKind::FreelistCountInconsistent {
                    declared,
                    walked: Some(walked),
                }));
            }
            let zeroed: Vec<u32> = leaves
                .iter()
                .copied()
                .filter(|&p| db.raw_page(p).is_some_and(|b| b.iter().all(|&x| x == 0)))
                .collect();
            if let Some(&first_zeroed) = zeroed.first() {
                out.push(Anomaly::new(AnomalyKind::ZeroedFreelistResidue {
                    zeroed: u32::try_from(zeroed.len()).unwrap_or(u32::MAX),
                    free_leaves: u32::try_from(leaves.len()).unwrap_or(u32::MAX),
                    first_zeroed,
                }));
            }
        }
        // An unwalkable chain (out-of-range / cyclic trunk pointer) is itself the
        // inconsistency: surface it loud rather than silently ignore the freelist.
        Err(_) => out.push(Anomaly::new(AnomalyKind::FreelistCountInconsistent {
            declared,
            walked: None,
        })),
    }

    if db.wal_applied() {
        // The overlay supersedes at least one page; report it as uncheckpointed
        // state. (The exact page count is not separately exposed; report ≥1.)
        out.push(Anomaly::new(AnomalyKind::WalUncheckpointedState {
            overlaid_pages: 1,
        }));
    }

    let header_pages = db.header_page_count();
    let file_pages = db.file_page_count();
    if header_pages != 0 && header_pages != file_pages {
        out.push(Anomaly::new(AnomalyKind::PageCountMismatch {
            header_pages,
            file_pages,
        }));
    }

    // Dropped/replaced schema objects whose sqlite_master row survives in page-1
    // free space (a DROP under secure_delete=OFF). Surfaced as findings so a
    // dropped table's existence + name is reported, not just buried in the carve.
    for schema in recover_dropped_schemas(db) {
        out.push(Anomaly::new(AnomalyKind::DroppedSchemaRecovered {
            object_type: schema.object_type,
            name: schema.name,
        }));
    }

    out
}

/// Audit an opened [`Database`] and convert each anomaly to the canonical
/// [`Finding`] under the supplied [`Source`], ready to merge into a `Report`.
#[must_use]
pub fn audit_findings(db: &Database, source: &Source) -> Vec<Finding> {
    audit(db)
        .into_iter()
        .map(|a| a.to_finding(source.clone()))
        .collect()
}

/// The database schema cookie: file-header bytes `40..44` as a big-endian `u32`
/// (file-format §1.3). Returns `None` when the page is too short to hold the
/// field — bounded and panic-free, so a truncated page-1 image degrades to "no
/// cookie" rather than indexing out of range.
fn schema_cookie(page: &[u8]) -> Option<u32> {
    let bytes = page.get(40..44)?;
    Some(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
}

/// Audit a rollback `-journal` (raw bytes) bound to `db` for the design-§6
/// observations, additive to [`audit`] (which covers main-db anomalies only).
///
/// The journal is parsed with the database's authoritative page size and graded
/// into observations, each derivable from the parser API and each "consistent
/// with …" (the examiner draws the conclusion):
/// - a **hot journal** (valid header magic) — an interrupted/in-progress write;
/// - a **recoverable** PERSIST journal (zeroed header, page images intact);
/// - **checksum mismatch(es)** (Tier A) — corruption / torn page / modification;
/// - a **schema-cookie advance** (journal page-1 image vs live db, offset 40) — a
///   DDL change, prior schema recoverable;
/// - a **duplicate page record** — corruption / savepoint / tampering;
/// - a **db-size delta** (Tier A `mxPage` vs current) — growth or shrink.
///
/// Read-only and panic-free: an unparsable/garbage journal simply yields no
/// observations rather than an error or panic.
#[must_use]
pub fn audit_journal(db: &Database, journal: &[u8]) -> Vec<Anomaly> {
    let mut out = Vec::new();

    let Ok(parsed) = RollbackJournal::parse(journal, db.header().page_size) else {
        // A page size outside [512, 65536] / non-power-of-two is the only parse
        // error; a database that opened cannot have one, so this degrades to no
        // observations rather than masking a real failure.
        return out;
    };

    let images = parsed.page_images();

    match parsed.header() {
        JournalHeader::Valid { mx_page, .. } => {
            // A valid-magic header is a hot journal (interrupted/in-progress).
            out.push(Anomaly::new(AnomalyKind::HotJournal {
                mx_page: *mx_page,
                journaled_pages: u32::try_from(images.len()).unwrap_or(u32::MAX),
            }));

            // mxPage (db size at txn start) vs the current page count. 0 is the
            // "size unknown" sentinel, so only a non-zero, differing value is a
            // reportable size delta.
            let current_pages = db.page_count();
            if *mx_page != 0 && *mx_page != current_pages {
                out.push(Anomaly::new(AnomalyKind::JournalDbSizeDelta {
                    mx_page: *mx_page,
                    current_pages,
                }));
            }
        }
        JournalHeader::ReconstructedZeroed { .. } => {
            // A zeroed (PERSIST post-commit) header with >=1 page image carries
            // recoverable pre-images.
            if !images.is_empty() {
                out.push(Anomaly::new(AnomalyKind::JournalRecoverable {
                    images: images.len(),
                }));
            }
        }
    }

    // Checksum mismatches (Tier A only — Tier B's checksum_valid is None).
    let bad: Vec<u32> = images
        .iter()
        .filter(|i| i.checksum_valid == Some(false))
        .map(|i| i.pgno)
        .collect();
    if !bad.is_empty() {
        out.push(Anomaly::new(AnomalyKind::JournalChecksumMismatch {
            pgnos: bad,
            total: images.len(),
        }));
    }

    // A DDL change is signalled by the schema cookie (file-header offset 40, BE)
    // advancing — NOT merely by page 1 being journaled. Page 1 carries the
    // change-counter / freelist-count / db-size header fields that update on
    // nearly every write, so it is present in almost every rollback journal; only
    // a cookie that differs between the journal's prior page-1 image and the live
    // database indicates CREATE/DROP/ALTER. Read both cookies defensively: if the
    // page-1 image is absent or the live page 1 is unreadable, do not fire.
    //
    // Caveat: for an UNCOMMITTED hot journal the live main-db file may not yet
    // carry the new cookie, so a DDL in a hot journal can be undetectable here.
    // That is acceptable — the detector never false-positives; it just cannot
    // always observe the committed-cookie case mid-flight.
    if let (Some(journal_cookie), Some(db_cookie)) = (
        images
            .iter()
            .find(|i| i.pgno == 1)
            .and_then(|i| schema_cookie(&i.bytes)),
        db.raw_page(1).as_deref().and_then(schema_cookie),
    ) {
        if journal_cookie != db_cookie {
            out.push(Anomaly::new(AnomalyKind::JournalSchemaChange {
                journal_cookie,
                db_cookie,
            }));
        }
    }

    // A page journaled more than once (the spec forbids it).
    if parsed.has_duplicate_pgno() {
        out.push(Anomaly::new(AnomalyKind::JournalDuplicatePage {
            pgnos: parsed.duplicate_pgnos().to_vec(),
        }));
    }

    out
}

/// Audit a rollback `-journal` and convert each design-§6 observation to the
/// canonical [`Finding`] under `source`, ready to merge into a `Report`. The
/// additive counterpart to [`audit_findings`] for the journal sidecar.
#[must_use]
pub fn audit_journal_findings(db: &Database, journal: &[u8], source: &Source) -> Vec<Finding> {
    audit_journal(db, journal)
        .into_iter()
        .map(|a| a.to_finding(source.clone()))
        .collect()
}

/// Carve deleted records and convert each to a canonical [`Finding`] under
/// `source`. The per-record confidence is threaded into the finding's context so
/// downstream consumers can filter low-confidence recoveries.
#[must_use]
pub fn audit_carved_findings(db: &Database, column_count: usize, source: &Source) -> Vec<Finding> {
    carve_deleted_records(db, column_count)
        .into_iter()
        .map(|rec| {
            Anomaly::new(AnomalyKind::DeletedRecordRecovered {
                page: rec.page,
                offset: rec.offset,
                rowid: rec.rowid,
            })
            .with_confidence(rec.confidence)
            .to_finding(source.clone())
        })
        .collect()
}

// ---------------------------------------------------------------------------
// Table attribution: reattach a carved deleted row to its source table in three
// honest tiers (CERTAIN / INFERRED / UNATTRIBUTED).
// ---------------------------------------------------------------------------

/// How confidently a carved deleted row has been reattached to a live table.
///
/// The three tiers encode distinct epistemic strengths (observed fact / forensic
/// inference / unattributed), and the renderer keeps them honest: a Tier-2 guess
/// is "consistent with" a table, never asserted as the table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Attribution {
    /// **Tier 1 — CERTAIN.** The record was carved from a page still part of a
    /// live table's b-tree, so the owning table is known for sure. Carries that
    /// table's name.
    Known(String),
    /// **Tier 2 — INFERRED.** The whole page was freed (linkage cut); the
    /// record's *shape* (column count + per-column affinity) matched one or more
    /// surviving tables. A forensic inference, never asserted as fact.
    Inferred {
        /// The candidate table whose shape the record matches.
        guess: String,
        /// `true` when more than one surviving table matched the shape, so the
        /// guess is one of several equally-consistent candidates.
        ambiguous: bool,
    },
    /// **Tier 3 — UNATTRIBUTED.** Dropped-table residue, or a shape matching no
    /// surviving table.
    Unattributed,
}

/// Whether a carved value is **storage-compatible** with a column of the given
/// declared affinity — the per-cell test the shape matcher applies.
///
/// Rules (permissive enough to survive `SQLite`'s numeric coercions, strict
/// enough to discriminate tables of equal arity but different type layout):
/// a NULL is compatible with anything; INTEGER/REAL are mutually compatible and
/// also satisfy NUMERIC; TEXT requires a text-leaning affinity (TEXT) or the
/// catch-all BLOB/NUMERIC; a BLOB requires BLOB. The BLOB affinity (or a column
/// with no declared type) accepts any value, matching SQLite's "no datatype"
/// semantics.
#[must_use]
pub fn value_fits_affinity(value: &Value, affinity: sqlite_core::attribution::Affinity) -> bool {
    use sqlite_core::attribution::Affinity;
    // BLOB affinity (incl. no-declared-type columns) is SQLite's catch-all: a
    // value of any storage class can live there.
    if affinity == Affinity::Blob {
        return true;
    }
    match value {
        // A NULL fits any column.
        Value::Null => true,
        Value::Integer(_) | Value::Real(_) => matches!(
            affinity,
            Affinity::Integer | Affinity::Real | Affinity::Numeric
        ),
        // NUMERIC accepts text too (SQLite stores a non-numeric string as text in
        // a NUMERIC column), so a TEXT value is consistent with TEXT or NUMERIC.
        Value::Text(_) => matches!(affinity, Affinity::Text | Affinity::Numeric),
        // A BLOB only sits in a BLOB column (handled above) — never a typed one.
        Value::Blob(_) => false,
    }
}

/// Whether a carved record's values match a live table's shape: equal column
/// count AND every value storage-compatible with the corresponding column
/// affinity ([`value_fits_affinity`]).
#[must_use]
pub fn shape_matches(values: &[Value], table: &sqlite_core::attribution::LiveTable) -> bool {
    if values.len() != table.affinities.len() {
        return false;
    }
    values
        .iter()
        .zip(&table.affinities)
        .all(|(v, &aff)| value_fits_affinity(v, aff))
}

/// The names of every live table whose shape a record's values match, in the
/// order the tables were listed. Zero matches → unattributable shape; one →
/// a clean guess; more than one → ambiguous.
#[must_use]
pub fn matching_tables(
    values: &[Value],
    tables: &[sqlite_core::attribution::LiveTable],
) -> Vec<String> {
    tables
        .iter()
        .filter(|t| shape_matches(values, t))
        .map(|t| t.name.clone())
        .collect()
}

/// Attribute one carved record to a tier given the live tables and the
/// page→table map. Pure (no DB access) so it is directly unit-testable.
///
/// - Tier 1 (CERTAIN): `source` is an in-page class ([`RecoverySource::InPageFreeBlock`]
///   or [`RecoverySource::FreeblockReconstructed`]) AND the record's page is in
///   `page_table_map` → [`Attribution::Known`].
/// - Tier 2 (INFERRED): `source` is [`RecoverySource::FreelistPage`] → match the
///   record's shape against `tables`; exactly one match → a guess, more than one
///   → ambiguous, none → [`Attribution::Unattributed`].
/// - Tier 3 (UNATTRIBUTED): [`RecoverySource::DroppedTable`], any other source,
///   or a shape matching no surviving table.
#[must_use]
pub fn attribute_record(
    rec: &CarvedRecord,
    tables: &[sqlite_core::attribution::LiveTable],
    page_table_map: &std::collections::BTreeMap<u32, String>,
) -> Attribution {
    // Tier 1 — CERTAIN: an in-page record on a page still owned by a live
    // table's b-tree. The page→table map is the hard linkage.
    if matches!(
        rec.source,
        RecoverySource::InPageFreeBlock | RecoverySource::FreeblockReconstructed
    ) {
        if let Some(name) = page_table_map.get(&rec.page) {
            return Attribution::Known(name.clone());
        }
        // In-page residue on a page that is NOT a live-table page (e.g. the
        // owning table itself was freed): no certain linkage, and in-page is not
        // the freelist class, so it is unattributed rather than inferred.
        return Attribution::Unattributed;
    }

    // Tier 2 — INFERRED: a whole-page freelist record; the linkage is cut, so
    // attribute by shape against the surviving tables.
    if rec.source == RecoverySource::FreelistPage {
        let mut matches = matching_tables(&rec.values, tables);
        return match matches.len() {
            0 => Attribution::Unattributed,
            1 => Attribution::Inferred {
                guess: matches.remove(0),
                ambiguous: false,
            },
            _ => Attribution::Inferred {
                guess: matches.remove(0),
                ambiguous: true,
            },
        };
    }

    // Tier 3 — UNATTRIBUTED: dropped-table residue or any other source.
    Attribution::Unattributed
}

/// Attribute every carved record, reading the live tables and page→table map
/// from `db` once. The returned vector is parallel to `records`.
#[must_use]
pub fn attribute_records(db: &Database, records: &[CarvedRecord]) -> Vec<Attribution> {
    let tables = db.live_tables();
    let page_table_map = db.page_to_table_map();
    records
        .iter()
        .map(|rec| attribute_record(rec, &tables, &page_table_map))
        .collect()
}

/// A **non-overclaiming diagnostic HINT** about a recovered record's relationship
/// to the *instance* of the table it was attributed to — ORTHOGONAL to
/// [`Attribution`] (it never changes the tier, the routing, or the table claim).
///
/// Detector A (`docs/design/drop-recreate-attribution.md`): when a record
/// attributed `Known(T)` has a rowid exceeding `T`'s `AUTOINCREMENT`
/// `sqlite_sequence` high-water mark, that is *consistent with* residue from a
/// prior incarnation of `T` (a drop-recreate reset `sqlite_sequence`) — but it is
/// **not proof**: the Codex review confirmed `rowid > seq` is equally reachable by
/// an `UPDATE` of the rowid, a manual `sqlite_sequence` edit, or a current-instance
/// deletion. So the flag is surfaced as a hint that names its evidence, and the
/// examiner draws the conclusion. The `note` never asserts "predecessor".
///
/// `#[non_exhaustive]` so the documented follow-up `SidecarSchemaChanged` variant
/// (Detector B — a sidecar `-wal`/`-journal` DDL boundary) can be added without a
/// breaking change.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum TableInstanceRisk {
    /// No instance-provenance hint applies to this record.
    None,
    /// The record was attributed `Known(table)`, `table` is `AUTOINCREMENT`, has a
    /// `sqlite_sequence` entry `seq`, and the record's `rowid` exceeds it. The
    /// current instance never assigned a rowid above `seq` via `INSERT`, so this is
    /// *consistent with* prior-incarnation residue — and equally with an `UPDATE`
    /// to the rowid, a `sqlite_sequence` edit, or a current-instance deletion.
    RowidExceedsAutoincHighwater {
        /// The live table the record was attributed to.
        table: String,
        /// The record's recovered rowid (the value exceeding the high-water mark).
        rowid: i64,
        /// The table's `AUTOINCREMENT` high-water mark (`sqlite_sequence.seq`).
        seq: i64,
    },
    /// Detector B (`docs/design/drop-recreate-attribution.md`): the record was
    /// attributed `Known(table)`, a `-wal`/`-journal` sidecar is in play, and the
    /// sidecar's PRIOR `sqlite_master` for `table` is **absent** OR carries a
    /// **different CREATE SQL text** than the current schema — an UNAMBIGUOUS
    /// table-level schema change within the captured window. It is *consistent
    /// with* a `CREATE`/`ALTER` or a drop+recreate of `table`, so residue
    /// attributed to `table` may predate the current schema. It is **table-level**,
    /// NOT row-level provenance, and deliberately does NOT fire on a rootpage move
    /// with identical CREATE SQL (a `VACUUM`), a schema-cookie advance alone, or a
    /// same-schema drop+recreate (indistinguishable from a benign page move).
    SidecarSchemaChanged {
        /// The live table whose sidecar prior schema differs from the current.
        table: String,
    },
}

impl TableInstanceRisk {
    /// A human, evidence-bearing note for the examiner — or `""` for
    /// [`TableInstanceRisk::None`]. The wording deliberately states what the
    /// observation is *consistent with* and lists the equally-consistent benign
    /// explanations; it never asserts the row is a predecessor.
    #[must_use]
    pub fn note(&self) -> String {
        match self {
            Self::None => String::new(),
            Self::RowidExceedsAutoincHighwater { table, rowid, seq } => format!(
                "rowid {rowid} exceeds table {table}'s AUTOINCREMENT high-water mark \
                 (sqlite_sequence={seq}) — consistent with residue from a prior incarnation of \
                 this table, but also explainable by an UPDATE to the rowid, a sqlite_sequence \
                 edit, or a current-instance deletion; the examiner should cross-check the RowID \
                 and any WAL/journal schema history."
            ),
            Self::SidecarSchemaChanged { table } => format!(
                "the schema for table {table} differs between the sidecar's (-wal/-journal) prior \
                 state and the current database (present-with-different-SQL or absent in the prior) \
                 — consistent with a CREATE/ALTER or drop+recreate within the captured window; \
                 residue attributed to {table} may predate the current schema, so reconcile against \
                 the prior schema. (A same-schema drop+recreate is indistinguishable from a benign \
                 page move and does NOT raise this.)"
            ),
        }
    }

    /// A short, stable provenance token for column/field output (`null`-like for
    /// [`TableInstanceRisk::None`] → empty string), carrying the evidence inline:
    /// `rowid_exceeds_autoinc_highwater(r=…,seq=…)`.
    #[must_use]
    pub fn token(&self) -> String {
        match self {
            Self::None => String::new(),
            Self::RowidExceedsAutoincHighwater { rowid, seq, .. } => {
                format!("rowid_exceeds_autoinc_highwater(r={rowid},seq={seq})")
            }
            Self::SidecarSchemaChanged { table } => {
                format!("sidecar_schema_changed({table})")
            }
        }
    }
}

/// Per-record [`TableInstanceRisk`], parallel to `records` and their
/// `attributions` — the Detector-A diagnostic pass.
///
/// A record trips [`TableInstanceRisk::RowidExceedsAutoincHighwater`] **only** when
/// ALL hold: it is `Attribution::Known(table)`; `table` is an `AUTOINCREMENT`
/// rowid table ([`sqlite_core::is_autoincrement`] on its `CREATE TABLE`); `table`
/// has a `sqlite_sequence` entry `seq` ([`Database::sqlite_sequence`]); and
/// `rec.rowid > seq`. Every other record is [`TableInstanceRisk::None`]. This is
/// orthogonal to attribution — it reads the same inputs but never alters the
/// attribution, tier, or routing.
#[must_use]
pub fn table_instance_risks(
    db: &Database,
    records: &[CarvedRecord],
    attributions: &[Attribution],
) -> Vec<TableInstanceRisk> {
    // The set of AUTOINCREMENT table names, from the live schema's CREATE sql.
    let autoinc: std::collections::BTreeSet<String> = db
        .live_tables()
        .into_iter()
        .filter(|t| sqlite_core::is_autoincrement(&t.create_sql))
        .map(|t| t.name)
        .collect();
    let sequences = db.sqlite_sequence();

    records
        .iter()
        .zip(attributions)
        .map(|(rec, attr)| {
            let Attribution::Known(table) = attr else {
                return TableInstanceRisk::None;
            };
            if !autoinc.contains(table) {
                return TableInstanceRisk::None;
            }
            match sequences.get(table) {
                Some(&seq) if rec.rowid > seq => TableInstanceRisk::RowidExceedsAutoincHighwater {
                    table: table.clone(),
                    rowid: rec.rowid,
                    seq,
                },
                _ => TableInstanceRisk::None,
            }
        })
        .collect()
}

/// The set of live tables whose CURRENT `CREATE TABLE` SQL differs from a sidecar
/// PRIOR `sqlite_master` — Detector B's UNAMBIGUOUS schema-change predicate
/// (`docs/design/drop-recreate-attribution.md`).
///
/// A table `T` is schema-changed when it is present in `current` and the prior
/// snapshot either (a) **lacks** `T` (absent in the prior `sqlite_master`) or
/// (b) carries `T` with a **different CREATE SQL text**. A table whose prior SQL
/// is byte-identical (the DML-only case, and the same-schema drop+recreate the
/// design refuses to claim) is NOT included — that is the anti-false-positive
/// boundary. An EMPTY `prior_schema` (no sidecar in play) yields the empty set, so
/// Detector B is silent unless a sidecar genuinely captured a different schema.
fn sidecar_schema_changed_tables(
    current: &std::collections::BTreeMap<String, String>,
    prior_schema: &std::collections::BTreeMap<String, String>,
) -> std::collections::BTreeSet<String> {
    // No sidecar (empty prior) ⟹ no Detector-B signal at all. Guarding here keeps
    // the "when in doubt, do NOT fire" rule explicit rather than implicit.
    if prior_schema.is_empty() {
        return std::collections::BTreeSet::new();
    }
    current
        .iter()
        .filter(|(name, current_sql)| match prior_schema.get(*name) {
            // Present in the prior with a DIFFERENT CREATE SQL ⟹ schema changed.
            Some(prior_sql) => prior_sql != *current_sql,
            // Absent in the prior (the table did not exist) ⟹ created since.
            None => true,
        })
        .map(|(name, _)| name.clone())
        .collect()
}

/// Per-record [`TableInstanceRisk`] over BOTH Detector A (bare AUTOINCREMENT
/// high-water) AND Detector B (sidecar schema-change), the sidecar-aware pass the
/// CLI composes when a `-wal`/`-journal` is in play
/// (`docs/design/drop-recreate-attribution.md`).
///
/// `prior_schema` is the sidecar's PRIOR `sqlite_master` as `name -> CREATE SQL`
/// ([`sqlite_core::PriorSnapshot::schema_sql`] for a `-journal`;
/// [`Database::schema_sql`] of the pre-WAL base bytes for a `-wal`). Pass an EMPTY
/// map to run Detector A only (the no-sidecar path), so this is a strict superset
/// of [`table_instance_risks`].
///
/// Detector B trips [`TableInstanceRisk::SidecarSchemaChanged`] for a record that
/// is `Attribution::Known(table)` whose `table` has a sidecar schema change —
/// prior absent, or prior CREATE SQL differs from current.
/// **Precedence:** Detector A wins where it fires — its rowid+seq evidence is more
/// specific than B's table-level boundary — so a record qualifying for both is
/// surfaced as `RowidExceedsAutoincHighwater`; Detector B fills only records A left
/// as [`TableInstanceRisk::None`]. Neither alters attribution, tier, or routing.
#[must_use]
pub fn table_instance_risks_with_sidecar(
    db: &Database,
    records: &[CarvedRecord],
    attributions: &[Attribution],
    prior_schema: &std::collections::BTreeMap<String, String>,
) -> Vec<TableInstanceRisk> {
    let mut risks = table_instance_risks(db, records, attributions);
    let changed = sidecar_schema_changed_tables(&db.schema_sql(), prior_schema);
    if changed.is_empty() {
        return risks; // no sidecar schema-change signal ⟹ Detector A result unchanged.
    }
    for (risk, attr) in risks.iter_mut().zip(attributions) {
        // Detector A takes precedence: only fill records A left as None.
        if *risk != TableInstanceRisk::None {
            continue;
        }
        let Attribution::Known(table) = attr else {
            continue;
        };
        if changed.contains(table) {
            *risk = TableInstanceRisk::SidecarSchemaChanged {
                table: table.clone(),
            };
        }
    }
    risks
}

/// The core per-rowid VERSION HISTORY ([`Database::row_histories`]) augmented with
/// free-space CARVED RESIDUE — the forensic-layer completion of Phase 1.
///
/// Carved residue is ORDER-UNKNOWN: a freeblock persists across commits, so
/// [`carve_at_commit`] / [`carve_with_fragments`] tag where residue was OBSERVED,
/// not where the row was deleted. Each carved record is therefore emitted as a
/// [`CarvedResidue`](sqlite_core::row_history::VersionOrigin::CarvedResidue) /
/// [`CarvedResidue`](sqlite_core::row_history::ViewState::CarvedResidue) version with
/// `commit_seq: None` (never a fabricated commit position), `is_deleted: true`,
/// and `is_guessed` set when its [`Attribution`] is `Inferred`. Residue is
/// attributed to a table via [`attribute_records`] and DEDUPED against any WAL
/// `AbsentInFinalView` version of the same rowid + values (the WAL holds it at
/// higher fidelity), so a row is never double-listed.
#[must_use]
pub fn row_histories_with_residue(db: &Database) -> Vec<sqlite_core::row_history::TableHistory> {
    use sqlite_core::row_history::{RowVersion, VersionOrigin, ViewState};

    let mut histories = db.row_histories();

    // Gather every carved residue record: the on-disk free space (Tier-1 full
    // rows) plus each materialized commit snapshot of the WAL. carve_at_commit /
    // carve_with_fragments already de-duplicate within their own pass; we collect
    // across passes and de-duplicate by content identity below.
    let mut records: Vec<CarvedRecord> = carve_with_fragments(db).full;
    if let Some(timeline) = db.wal_timeline() {
        for snapshot in timeline.commit_snapshots() {
            records.extend(carve_at_commit(db, &timeline, snapshot.id()));
        }
    }
    // Collapse byte-identical carves (a row can recur across commits / regions).
    let records = dedup_keep_best(records, &db.page_to_table_map());

    // Attribute each carved record to a table. The strongest signal is page
    // linkage: a record carved from a page still owned by a live table's b-tree
    // (e.g. a PRIOR-version remnant in that table's slack) belongs to THAT table
    // for certain — stronger than, and independent of, the source-class tiering in
    // `attribute_record`. We consult the page→table map first, then fall back to
    // the shape-based `attribute_records` for off-table (freelist-page) residue.
    let page_table_map = db.page_to_table_map();
    let attributions = attribute_records(db, &records);

    // A residue carve is deduped against a WAL `AbsentInFinalView` version of the
    // SAME rowid + values already in that table's history (the WAL holds it at
    // higher fidelity). Build the set of those (table, rowid, values) keys.
    let absent_keys: std::collections::HashSet<String> = histories
        .iter()
        .flat_map(|h| {
            h.versions
                .iter()
                .filter(|v| v.view_state == ViewState::AbsentInFinalView)
                .map(move |v| residue_key(&h.table, v.rowid, &v.values))
        })
        .collect();

    // Group new residue versions by target table name.
    let mut to_add: std::collections::BTreeMap<String, Vec<RowVersion>> =
        std::collections::BTreeMap::new();
    for (rec, attr) in records.iter().zip(attributions.iter()) {
        // Page linkage (CERTAIN) wins; else the shape-based tier.
        let (table, is_guessed) = if let Some(name) = page_table_map.get(&rec.page) {
            (name.clone(), false)
        } else {
            match attr {
                Attribution::Known(name) => (name.clone(), false),
                Attribution::Inferred { guess, .. } => (guess.clone(), true),
                // No surviving table to attach the residue to — skip rather than
                // invent a home for it (no fabricated attribution).
                Attribution::Unattributed => continue,
            }
        };
        // A clobbered rowid (carving surfaces it as 0) is genuinely unknown.
        let rowid = if rec.rowid == 0 {
            None
        } else {
            Some(rec.rowid)
        };
        let key = residue_key(&table, rowid, &rec.values);
        if absent_keys.contains(&key) {
            continue; // already listed as a higher-fidelity WAL AbsentInFinalView
        }
        to_add.entry(table).or_default().push(RowVersion {
            rowid,
            values: rec.values.clone(),
            origin: VersionOrigin::CarvedResidue,
            // Order-unknown: a freeblock persists across commits, so a carve has
            // no trustworthy commit position. NEVER fabricate one.
            commit_seq: None,
            view_state: ViewState::CarvedResidue,
            is_deleted: true,
            is_guessed,
            rowid_reused: false,
            // An inferred (shape-matched) attribution is uncertain by nature.
            attribution_uncertain: is_guessed,
            // Carved residue is order-unknown; a delete+reinsert gap is a WAL-view
            // signal only, which a carve cannot witness.
            reinserted_after_gap: false,
        });
    }

    // Merge the residue versions into their tables, de-duplicating residue that is
    // byte-identical to a version already present, then re-sort.
    for h in &mut histories {
        if let Some(extra) = to_add.remove(&h.table) {
            let existing: std::collections::HashSet<String> = h
                .versions
                .iter()
                .map(|v| residue_key(&h.table, v.rowid, &v.values))
                .collect();
            for v in extra {
                if existing.contains(&residue_key(&h.table, v.rowid, &v.values)) {
                    continue;
                }
                h.versions.push(v);
            }
            sqlite_core::row_history::sort_table_versions(&mut h.versions);
        }
    }
    histories
}

/// A content-identity key for residue de-duplication: table + rowid + a stable
/// `Debug` rendering of the values (`Value` carries an `f64`, so it is not
/// `Hash`/`Eq` directly).
fn residue_key(table: &str, rowid: Option<i64>, values: &[Value]) -> String {
    format!("{table}:{rowid:?}:{values:?}")
}

#[cfg(test)]
mod structural_noise_tests {
    use super::is_structural_noise;
    use sqlite_core::Value;

    #[test]
    fn rejects_rowid_echo_with_all_null_tail() {
        // The inferred over-read: a rowid echoed as the INTEGER-PK first column
        // followed by a long serial-type-0 (NULL) tail — no recoverable content.
        let mut overwide = vec![Value::Null; 102];
        overwide[0] = Value::Integer(14);
        assert!(is_structural_noise(&overwide));
    }

    #[test]
    fn rejects_all_null_record() {
        assert!(is_structural_noise(&[
            Value::Null,
            Value::Null,
            Value::Null
        ]));
    }

    #[test]
    fn keeps_ordinary_recovered_row() {
        let good = vec![
            Value::Integer(3),
            Value::Text("Bowl".into()),
            Value::Real(11.23),
            Value::Null,
        ];
        assert!(!is_structural_noise(&good));
    }

    #[test]
    fn keeps_dropped_table_row_wider_than_any_live_table() {
        // A dropped table can be wider than every surviving live table; such an
        // unattributed record carries real values and MUST be kept (a width cap
        // would discard it — the regression this guards).
        let wide_real = vec![
            Value::Integer(7),
            Value::Text("secret".into()),
            Value::Integer(42),
            Value::Text("more".into()),
        ];
        assert!(!is_structural_noise(&wide_real));
    }

    #[test]
    fn keeps_single_column_row() {
        // A genuine one-column row is not noise (length < 2 short-circuits).
        assert!(!is_structural_noise(&[Value::Text("solo".into())]));
    }
}

#[cfg(test)]
mod attribution_tests {
    use super::{
        attribute_record, attribute_records, matching_tables, shape_matches, value_fits_affinity,
        Attribution, CarvedRecord, RecoverySource,
    };
    use sqlite_core::attribution::{Affinity, LiveTable};
    use sqlite_core::Value;

    fn table(name: &str, root: u32, affinities: Vec<Affinity>) -> LiveTable {
        LiveTable {
            name: name.to_string(),
            rootpage: root,
            column_names: None,
            affinities,
            create_sql: String::new(),
        }
    }

    fn rec(page: u32, source: RecoverySource, values: Vec<Value>) -> CarvedRecord {
        CarvedRecord {
            page,
            offset: 0,
            rowid: 1,
            values,
            confidence: 0.9,
            allocated: false,
            source,
            wal: None,
            overflow: None,
        }
    }

    #[test]
    fn value_affinity_compatibility() {
        assert!(value_fits_affinity(&Value::Null, Affinity::Integer));
        assert!(value_fits_affinity(&Value::Integer(1), Affinity::Integer));
        assert!(value_fits_affinity(&Value::Integer(1), Affinity::Real));
        assert!(value_fits_affinity(&Value::Real(1.0), Affinity::Numeric));
        assert!(value_fits_affinity(
            &Value::Text("x".into()),
            Affinity::Text
        ));
        assert!(value_fits_affinity(&Value::Blob(vec![1]), Affinity::Blob));
        // A text value does NOT fit an integer column.
        assert!(!value_fits_affinity(
            &Value::Text("x".into()),
            Affinity::Integer
        ));
        // A blob does not fit a text column.
        assert!(!value_fits_affinity(&Value::Blob(vec![1]), Affinity::Text));
        // BLOB (or no-type) affinity is the catch-all: any value fits.
        assert!(value_fits_affinity(
            &Value::Text("x".into()),
            Affinity::Blob
        ));
        assert!(value_fits_affinity(&Value::Integer(1), Affinity::Blob));
    }

    #[test]
    fn shape_match_requires_equal_arity_and_compatible_cells() {
        let t = table("people", 2, vec![Affinity::Integer, Affinity::Text]);
        assert!(shape_matches(
            &[Value::Integer(1), Value::Text("a".into())],
            &t
        ));
        // Wrong arity.
        assert!(!shape_matches(&[Value::Integer(1)], &t));
        // Incompatible cell (text in an integer column).
        assert!(!shape_matches(
            &[Value::Text("a".into()), Value::Text("b".into())],
            &t
        ));
    }

    #[test]
    fn clean_single_match() {
        let tables = vec![
            table("people", 2, vec![Affinity::Integer, Affinity::Text]),
            table("amounts", 3, vec![Affinity::Real, Affinity::Real]),
        ];
        let m = matching_tables(&[Value::Integer(1), Value::Text("a".into())], &tables);
        assert_eq!(m, vec!["people"]);
    }

    #[test]
    fn ambiguous_two_tables_same_shape() {
        let tables = vec![
            table("a", 2, vec![Affinity::Integer, Affinity::Text]),
            table("b", 3, vec![Affinity::Integer, Affinity::Text]),
        ];
        let m = matching_tables(&[Value::Integer(1), Value::Text("x".into())], &tables);
        assert_eq!(m, vec!["a", "b"]);
    }

    #[test]
    fn no_match_is_unattributed_shape() {
        let tables = vec![table("a", 2, vec![Affinity::Integer, Affinity::Text])];
        let m = matching_tables(&[Value::Blob(vec![1]), Value::Blob(vec![2])], &tables);
        assert!(m.is_empty());
    }

    #[test]
    fn tier1_inpage_page_in_map_is_known() {
        let tables = vec![table("people", 2, vec![Affinity::Integer, Affinity::Text])];
        let mut map = std::collections::BTreeMap::new();
        map.insert(5u32, "people".to_string());
        let r = rec(
            5,
            RecoverySource::InPageFreeBlock,
            vec![Value::Integer(1), Value::Text("a".into())],
        );
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Known("people".to_string())
        );
    }

    #[test]
    fn tier1_freeblock_reconstructed_in_map_is_known() {
        let tables = vec![table("people", 2, vec![Affinity::Integer, Affinity::Text])];
        let mut map = std::collections::BTreeMap::new();
        map.insert(7u32, "people".to_string());
        let r = rec(7, RecoverySource::FreeblockReconstructed, vec![Value::Null]);
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Known("people".to_string())
        );
    }

    #[test]
    fn tier2_freelist_single_match_is_inferred_unambiguous() {
        let tables = vec![
            table("people", 2, vec![Affinity::Integer, Affinity::Text]),
            table("amounts", 3, vec![Affinity::Real, Affinity::Real]),
        ];
        let map = std::collections::BTreeMap::new();
        let r = rec(
            9,
            RecoverySource::FreelistPage,
            vec![Value::Integer(1), Value::Text("a".into())],
        );
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Inferred {
                guess: "people".to_string(),
                ambiguous: false
            }
        );
    }

    #[test]
    fn tier2_freelist_multi_match_is_ambiguous() {
        let tables = vec![
            table("a", 2, vec![Affinity::Integer, Affinity::Text]),
            table("b", 3, vec![Affinity::Integer, Affinity::Text]),
        ];
        let map = std::collections::BTreeMap::new();
        let r = rec(
            9,
            RecoverySource::FreelistPage,
            vec![Value::Integer(1), Value::Text("x".into())],
        );
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Inferred {
                guess: "a".to_string(),
                ambiguous: true
            }
        );
    }

    #[test]
    fn tier2_freelist_no_match_is_unattributed() {
        let tables = vec![table("a", 2, vec![Affinity::Integer, Affinity::Text])];
        let map = std::collections::BTreeMap::new();
        let r = rec(
            9,
            RecoverySource::FreelistPage,
            vec![Value::Blob(vec![1]), Value::Blob(vec![2])],
        );
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Unattributed
        );
    }

    #[test]
    fn tier3_dropped_table_is_unattributed() {
        let tables = vec![table("a", 2, vec![Affinity::Integer, Affinity::Text])];
        let map = std::collections::BTreeMap::new();
        let r = rec(
            9,
            RecoverySource::DroppedTable,
            vec![Value::Integer(1), Value::Text("x".into())],
        );
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Unattributed
        );
    }

    #[test]
    fn tier1_inpage_page_not_in_map_falls_through_to_unattributed() {
        // An in-page record on a page that is NOT a live-table page (e.g. its
        // table was the one freed): no Tier-1 certainty, and in-page is not the
        // freelist class, so it is unattributed rather than inferred.
        let tables = vec![table("people", 2, vec![Affinity::Integer, Affinity::Text])];
        let map = std::collections::BTreeMap::new();
        let r = rec(
            42,
            RecoverySource::InPageFreeBlock,
            vec![Value::Integer(1), Value::Text("a".into())],
        );
        assert_eq!(
            attribute_record(&r, &tables, &map),
            Attribution::Unattributed
        );
    }

    #[test]
    fn attribute_records_reads_schema_from_a_real_database() {
        // Mint a tiny valid db via the writer (no sqlite3 needed), open it, and
        // drive the db-backed attribute_records wrapper end to end.
        use sqlite_core::rebuild::{build_recovered_db_tables, RecoveredTable};
        use sqlite_core::Database;
        let seed = vec![RecoveredTable {
            name: "people".to_string(),
            columns: vec!["id".to_string(), "name".to_string()],
            rows: vec![vec![Value::Integer(1), Value::Text("a".into())]],
        }];
        let db = Database::open(build_recovered_db_tables(&seed)).expect("minted db opens");
        let records = vec![super::CarvedRecord {
            page: 99,
            offset: 0,
            rowid: 0,
            values: vec![Value::Integer(2), Value::Text("b".into())],
            confidence: 0.9,
            allocated: false,
            source: RecoverySource::FreelistPage,
            wal: None,
            overflow: None,
        }];
        let attrs = attribute_records(&db, &records);
        assert_eq!(attrs.len(), 1);
        // (INTEGER, TEXT) freelist row matches the people table's shape.
        assert!(matches!(attrs[0], Attribution::Inferred { .. }));
    }
}

#[cfg(test)]
mod tests {
    use super::{dedup_fragments, CarvedFragment, RecoverySource};
    use sqlite_core::Value;

    fn frag(page: u32, offset: usize, surviving: Vec<(usize, Value)>) -> CarvedFragment {
        let missing = 3usize.saturating_sub(surviving.len());
        CarvedFragment {
            page,
            offset,
            surviving,
            missing,
            confidence: 0.2,
            source: RecoverySource::InPageFreeBlock,
            wal: None,
        }
    }

    /// `dedup_fragments` orders by survivor count (richest kept), drops a repeated
    /// `(page, offset)` anchor, and collapses a value-level duplicate at a fresh
    /// anchor — exercising both dedup arms and the survivor-count sort over a
    /// multi-element input (a single-element vector never invokes the comparator).
    #[test]
    fn dedup_keeps_richest_and_drops_duplicates() {
        let rich = frag(
            1,
            10,
            vec![(0, Value::Integer(1)), (1, Value::Text("a".into()))],
        );
        let poor_same_anchor = frag(1, 10, vec![(0, Value::Integer(1))]);
        let value_dup_new_anchor = frag(
            2,
            20,
            vec![(0, Value::Integer(1)), (1, Value::Text("a".into()))],
        );

        let out = dedup_fragments(vec![poor_same_anchor, rich, value_dup_new_anchor]);

        // The richest copy of the shared identity survives; the poorer same-anchor
        // copy and the value-identical fresh-anchor copy are both dropped.
        assert_eq!(
            out.len(),
            1,
            "duplicates collapse to the richest single copy"
        );
        assert_eq!(
            out[0].surviving.len(),
            2,
            "the 2-column copy is the one kept"
        );
    }
}