//! Per-build records for `pkh deb`.
//!
//! A record is one build's metadata: identity, lineage, outcome, the
//! SHA-256 of the chroot tarball its environment was extracted from, a
//! tree phase journal, the measured build durations, the full build
//! log, and — for `--incremental` runs — the staged tree's overlay
//! upperdir. Records live under
//! `/var/tmp/pkh/sessions/<slug>/<record-id>/`.
//!
//! The chroot itself is per-run scratch: every build (plain and
//! `--incremental` alike) extracts a fresh one from the sha-verified
//! mmdebstrap tarball cache into the record's `chroot/` directory and
//! removes it at the end of the run — success, failure and interrupt
//! all included. Only the cheap, content-addressed state (upperdir,
//! log, metadata) is persisted; space is a hard constraint.
//!
//! The layout of one record root:
//!
//! ```text
//! /var/tmp/pkh/sessions/<slug>/
//!   <record-id>/        # one record per build
//!     manifest.json     # identity, lineage, outcome, tarball sha,
//!                       # journal, build durations
//!     log/build.log     # full tee of the build output (always)
//!     tree-upper/       # the staged tree's overlay upperdir
//!                       # (--incremental runs only, post-run)
//!     host-files.list   # host-tree snapshot, while tree-upper is kept
//!     chroot/           # DURING the run only; removed at end of run
//!   <record-id>.lock    # advisory lock (flock) marking a live build,
//!                       # outside the record root it guards
//! ```
//!
//! Metadata and the log are kept for EVERY build (plain runs become
//! metadata-only records); keeping the upperdir is what
//! `--incremental` buys. A later `--incremental` run forks a base
//! record: its upperdir is carried when the base's recorded tarball
//! sha still matches the cache (the toolchain did not move) and the
//! changelog version is unchanged, and never mutated in place. This
//! module holds the manifest model, discovery, locking, forking, the
//! keep-last-5 upper cap, the host-tree snapshot used to propagate
//! host-side deletions into a carried upperdir, the `pkh deb list`
//! rendering and the `pkh deb log` streaming. The base-selection
//! policy lives in [`crate::deb`].

use std::collections::BTreeMap;
use std::fs;
use std::io::{self, Seek, Write};
use std::os::unix::fs::{FileTypeExt, MetadataExt};
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::context::{LineSink, Stream};

/// Manifest schema version: a record recorded by another schema is
/// ignored (and eventually pruned) rather than forked.
pub(crate) const SCHEMA_VERSION: u32 = 1;

/// How long a record must be untouched before the default `pkh prune`
/// removes it.
pub const SESSION_RETENTION: Duration = Duration::from_secs(7 * 24 * 60 * 60);

/// How many records of one identity may retain their tree-upper at the
/// same time: when a newer upper-retaining record appears, the oldest
/// one beyond this count degrades to a metadata-only record (its
/// upperdir and host-tree snapshot are deleted; manifest and log stay).
pub(crate) const KEEP_UPPERS: usize = 5;

/// The build is running (`last_outcome`).
pub const OUTCOME_RUNNING: &str = "running";
/// The build recorded in this record succeeded (`last_outcome`).
pub const OUTCOME_SUCCESS: &str = "success";
/// The build recorded in this record failed (`last_outcome`).
pub const OUTCOME_FAILED: &str = "failed";

/// Name of the manifest file inside a record root.
const MANIFEST_NAME: &str = "manifest.json";
/// Name of the host-tree snapshot inside a record root.
const SNAPSHOT_NAME: &str = "host-files.list";
/// Directory holding the build log inside a record root.
const LOG_DIR: &str = "log";
/// Name of the build log inside [`LOG_DIR`].
const LOG_NAME: &str = "build.log";
/// The per-run chroot directory inside a record root (removed at end of
/// run).
const CHROOT_DIR: &str = "chroot";
/// The persisted overlay upperdir inside a record root.
const UPPER_DIR: &str = "tree-upper";
/// The per-run overlay workdir inside a record root (removed at end of
/// run; overlayfs state is meaningless between runs).
const WORK_DIR: &str = "tree-upper-work";
/// Advisory lock file suffix next to a record root (`<record-id>.lock`).
const LOCK_NAME: &str = "lock";
/// Stable build root inside the chroot for record builds: the overlay
/// upperdir mirrors paths relative to the merge point, so the staged
/// tree must live at the same in-chroot path on every attempt for the
/// object files to be reusable.
pub(crate) const BUILD_ROOT: &str = "/tmp/pkh-build";

/// What a record was built for: the identity keys the environment and
/// the fork candidates.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionIdentity {
    /// Source package name.
    pub package: String,
    /// Distribution series the chroot was bootstrapped for.
    pub series: String,
    /// Target architecture.
    pub arch: String,
    /// Whether the build cross-compiles.
    pub cross: bool,
}

impl std::fmt::Display for SessionIdentity {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{}/{} {}{}",
            self.package,
            self.series,
            self.arch,
            if self.cross { " (cross)" } else { "" }
        )
    }
}

/// One journal entry: when a phase last completed, and its input stamp.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PhaseStamp {
    /// RFC 3339 completion timestamp.
    pub at: String,
    /// Hash of the phase inputs (the patch tree), when the phase has
    /// one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stamp: Option<String>,
}

/// Wall-clock durations of one build's work, as recorded in a record's
/// manifest: the dependency installation (every apt phase before the
/// build — package list update, essential and injected packages,
/// build-dependencies) and the package build itself (`debian/rules
/// build` plus `binary`). The full build time is the sum of both.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct BuildTimings {
    /// Seconds spent installing dependencies.
    pub deps_secs: u64,
    /// Seconds spent building the package.
    pub build_secs: u64,
}

impl BuildTimings {
    /// Accumulate a measured duration into the dependency-installation
    /// total.
    pub fn add_deps(&mut self, elapsed: Duration) {
        self.deps_secs += elapsed.as_secs();
    }

    /// Accumulate a measured duration into the package-build total.
    pub fn add_build(&mut self, elapsed: Duration) {
        self.build_secs += elapsed.as_secs();
    }

    /// Whether nothing was measured: the build ended before any timed
    /// command ran, and the record must not claim durations of zero.
    pub fn is_empty(&self) -> bool {
        self.deps_secs == 0 && self.build_secs == 0
    }
}

/// One executed phase of a build profile: the stable machine name of a
/// [`crate::deb::Phase`] ([`crate::deb::Phase::key`]) and the
/// wall-clock milliseconds it ran.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PhaseTiming {
    /// Stable machine name of the phase.
    pub phase: String,
    /// Wall-clock milliseconds the phase ran.
    pub millis: u64,
}

/// Per-phase wall-clock profile of one build, recorded in the record's
/// manifest and shown by `pkh deb profile`: one entry per executed
/// phase, in execution order. A phase the flow skipped (the chroot
/// tarball was already cached, the device nodes were ready) has no
/// entry; a failed build's profile ends at the phase the error
/// propagated out of.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct BuildProfile {
    /// Executed phases with their durations, in execution order.
    pub phases: Vec<PhaseTiming>,
}

impl BuildProfile {
    /// Whether nothing was recorded: the build ended before any phase
    /// produced a measurement, and the record must not claim an empty
    /// profile.
    pub fn is_empty(&self) -> bool {
        self.phases.is_empty()
    }

    /// Total milliseconds across the recorded phases.
    pub fn total_millis(&self) -> u64 {
        self.phases.iter().map(|p| p.millis).sum()
    }
}

/// The record manifest: identity, lineage, outcome and tree state.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionManifest {
    /// Manifest schema version ([`SCHEMA_VERSION`]).
    pub schema: u32,
    /// pkh version that last wrote the manifest.
    pub pkh: String,
    /// Record id: the build-start timestamp of this record
    /// (`YYYYMMDDTHHMMSS`, UTC).
    pub id: String,
    /// Canonical host tree the record was recorded from; `pkh deb list`
    /// and `--incremental <id>` scope records to this.
    pub host_tree: String,
    /// What the environment was built for.
    pub identity: SessionIdentity,
    /// RFC 3339 creation timestamp.
    pub created: String,
    /// RFC 3339 timestamp of this record's build.
    pub last_used: String,
    /// Outcome of this record's build ([`OUTCOME_RUNNING`]/[`OUTCOME_SUCCESS`]/[`OUTCOME_FAILED`]).
    pub last_outcome: String,
    /// Id of the record this one was forked from, when any
    /// (`--incremental` lineage; `None` when built from scratch).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub base: Option<String>,
    /// SHA-256 of the chroot tarball this record's environment was
    /// extracted from: a fork carries the base's upperdir only while
    /// this still equals the cached tarball's digest (the toolchain
    /// moved otherwise, and carried objects would be stale). Empty
    /// until the bootstrap completes.
    #[serde(default)]
    pub tarball_sha256: String,
    /// Phase journal: completed tree phases with their input stamps.
    /// Environment phases (apt, build-deps) rerun in every fresh chroot
    /// and are not recorded.
    pub phases: BTreeMap<String, PhaseStamp>,
    /// Changelog version of this record's build: a version change
    /// between a base and its fork drops the carried upperdir (the
    /// objects are of the old tree).
    pub tree_version: String,
    /// Whether this record retains its tree-upper after the run
    /// (`--incremental` only).
    #[serde(default)]
    pub kept_upper: bool,
    /// Whether this record's build staged the tree over a carried
    /// overlay upperdir (build-level resume was active).
    pub build_resume: bool,
    /// Wall-clock durations of this record's build (`None`: not
    /// recorded — the build ended before any timed command ran, or the
    /// record predates the field).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub timings: Option<BuildTimings>,
    /// Per-phase wall-clock profile of this record's build (`None`: not
    /// recorded — the record predates the field, or the build ended
    /// before any phase produced a measurement).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub profile: Option<BuildProfile>,
}

/// State shared through the [`Session`] handle: the manifest plus the
/// advisory lock file held open for the lifetime of the handle.
#[derive(Debug)]
struct SessionState {
    manifest: SessionManifest,
    lock: Option<fs::File>,
}

/// A handle to one record root: manifest access, journal recording and
/// the paths of everything the build flow touches. Clones share the state.
#[derive(Clone, Debug)]
pub struct Session {
    root: PathBuf,
    state: Arc<Mutex<SessionState>>,
}

/// What the end-of-run retention of a record's staged tree does,
/// decided from the request and the disk: a keep requested for a run
/// that ended before staging created a tree-upper has nothing to keep,
/// and the record must not claim otherwise.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum UpperRetention {
    /// Keep the staged tree: the record is a fork candidate.
    Keep,
    /// Drop the staged tree: a plain run.
    Drop,
    /// An `--incremental` run whose staging never happened: metadata
    /// and log only, `kept_upper` stays false.
    KeepNothing,
}

/// Decide the retention of a record's staged tree: an
/// `--incremental` run keeps it — unless the run ended before staging
/// created one, in which case there is nothing to keep — and a plain
/// run drops it.
pub(crate) fn upper_retention(requested: bool, upper_exists: bool) -> UpperRetention {
    match (requested, upper_exists) {
        (true, true) => UpperRetention::Keep,
        (true, false) => UpperRetention::KeepNothing,
        (false, _) => UpperRetention::Drop,
    }
}

impl Session {
    /// The record root directory.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// A snapshot of the manifest.
    pub fn manifest(&self) -> SessionManifest {
        self.state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .manifest
            .clone()
    }

    /// The per-run chroot directory inside the record root: extracted
    /// fresh from the tarball cache at the start of the run, removed at
    /// its end.
    pub fn chroot_path(&self) -> PathBuf {
        self.root.join(CHROOT_DIR)
    }

    /// The persisted overlay upperdir (host-side path): the build's
    /// writes land here and are carried by a fork when the record
    /// retains it.
    pub fn upper_dir(&self) -> PathBuf {
        self.root.join(UPPER_DIR)
    }

    /// The per-run overlay workdir (host-side path); scratch, removed at
    /// the end of the run.
    pub fn work_dir(&self) -> PathBuf {
        self.root.join(WORK_DIR)
    }

    /// The host-tree snapshot file inside the record root.
    pub(crate) fn snapshot_file(&self) -> PathBuf {
        self.root.join(SNAPSHOT_NAME)
    }

    /// The build log file inside the record root.
    pub fn log_file(&self) -> PathBuf {
        self.root.join(LOG_DIR).join(LOG_NAME)
    }

    /// Whether the record retains a tree-upper on disk (a fork
    /// candidate).
    pub fn has_upper(&self) -> bool {
        self.upper_dir().is_dir()
    }

    /// The chroot-side path of the staged build root (stable across
    /// records, see [`BUILD_ROOT`]).
    pub fn build_root(&self) -> String {
        BUILD_ROOT.to_string()
    }

    /// Persist the manifest atomically (write to a temp file, rename).
    pub fn save(&self) -> io::Result<()> {
        let manifest = self.manifest();
        save_manifest(&self.root, &manifest)
    }

    /// Record a phase completion in the journal and persist it.
    pub fn record_phase(&self, phase: &str, stamp: Option<String>) {
        let entry = PhaseStamp {
            at: now_rfc3339(),
            stamp,
        };
        self.with_manifest(|manifest| {
            manifest.phases.insert(phase.to_string(), entry);
            manifest.last_used = now_rfc3339();
        });
    }

    /// Record the outcome of this record's build and persist it.
    pub fn set_outcome(&self, outcome: &str) {
        self.with_manifest(|manifest| {
            manifest.last_outcome = outcome.to_string();
            manifest.last_used = now_rfc3339();
        });
    }

    /// Record the SHA-256 of the chroot tarball this run's environment
    /// was extracted from, and persist it.
    pub fn set_tarball_sha256(&self, sha256: &str) {
        self.with_manifest(|manifest| {
            manifest.tarball_sha256 = sha256.to_string();
        });
    }

    /// Record whether this record's build reused the overlay upperdir.
    pub fn set_build_resume(&self, value: bool) {
        self.with_manifest(|manifest| {
            manifest.build_resume = value;
        });
    }

    /// Record the measured build durations (dependency installation,
    /// package build) and persist them.
    pub fn set_timings(&self, timings: BuildTimings) {
        self.with_manifest(|manifest| {
            manifest.timings = Some(timings);
        });
    }

    /// Record the per-phase wall-clock profile of this record's build
    /// and persist it.
    pub fn set_profile(&self, profile: BuildProfile) {
        self.with_manifest(|manifest| {
            manifest.profile = Some(profile);
        });
    }

    /// Run `f` over the manifest and persist the result; failures are
    /// logged and never abort the build (the journal is a cache).
    fn with_manifest(&self, f: impl FnOnce(&mut SessionManifest)) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        f(&mut state.manifest);
        if let Err(e) = save_manifest(&self.root, &state.manifest) {
            log::warn!(
                "Failed to save the build record manifest '{}': {}",
                self.root.join(MANIFEST_NAME).display(),
                e
            );
        }
    }

    /// Retain the record at the end of its build: the outcome is
    /// recorded on success (failures already carry theirs), and the staged
    /// tree follows [`upper_retention`] — kept exactly for `--incremental`
    /// runs that actually staged one; a plain run's tree is dropped and a
    /// run that ended before staging keeps nothing, both leaving a
    /// metadata-only record (manifest + log). The chroot was already
    /// removed by the caller's teardown. The interrupt path is handled by
    /// the SIGINT cleanup hook, which follows the same policy.
    pub(crate) fn retain_upper(&self, keep_upper: bool, succeeded: bool) {
        if succeeded {
            self.set_outcome(OUTCOME_SUCCESS);
        }
        match upper_retention(keep_upper, self.has_upper()) {
            UpperRetention::Keep => {
                self.with_manifest(|manifest| {
                    manifest.kept_upper = true;
                });
                log::info!(
                    "Build record {} kept its staged tree (the next \
                 --incremental run forks it)",
                    self.manifest().id
                );
                self.enforce_upper_cap();
            }
            UpperRetention::KeepNothing => {
                log::info!(
                    "Build record {} kept (no staged tree was created — the \
                 run ended before staging)",
                    self.manifest().id
                );
            }
            UpperRetention::Drop => {
                log::debug!(
                    "Dropping the staged tree of build record {} (plain run)",
                    self.manifest().id
                );
                self.wipe_upper();
            }
        }
    }

    /// Drop the record's persisted upperdir state: the tree-upper, its
    /// host-tree snapshot and the scratch workdir.
    fn wipe_upper(&self) {
        for path in [self.upper_dir(), self.work_dir(), self.snapshot_file()] {
            if let Err(e) = crate::deb::ephemeral::privileged_remove(&path)
                && e.kind() != io::ErrorKind::NotFound
            {
                log::warn!("Failed to remove '{}': {}", path.display(), e);
            }
        }
    }

    /// Enforce [`KEEP_UPPERS`] for this record's identity: degrade the
    /// oldest upper-retaining records beyond the cap to metadata-only
    /// (upperdir and snapshot deleted, manifest and log kept). Records
    /// locked by a live build are never degraded.
    fn enforce_upper_cap(&self) {
        // The record root is <sessions root>/<slug>/<id>: the sessions
        // root is two levels up, keeping the cap on the same root the
        // record lives in (tests use fixture roots).
        let Some(root) = self.root.parent().and_then(Path::parent) else {
            return;
        };
        let identity = self.manifest().identity;
        let mut uppers: Vec<Session> = identity_records_in(root, &identity)
            .into_iter()
            .filter(|record| record.has_upper())
            .collect();
        uppers.sort_by(|a, b| record_ordering(a, b).reverse());
        for record in uppers.into_iter().skip(KEEP_UPPERS) {
            if lock_is_held(&record.lock_path()).unwrap_or(true) {
                log::debug!(
                    "Record {} is locked by a live build; skipping the \
                     upper cap",
                    record.manifest().id
                );
                continue;
            }
            log::debug!(
                "Record {} degrades to metadata-only (beyond the newest \
                 {KEEP_UPPERS} staged trees of {identity})",
                record.manifest().id
            );
            record.wipe_upper();
            record.with_manifest(|manifest| {
                manifest.kept_upper = false;
            });
        }
    }

    /// Remove the per-run state of this record: the chroot directory and
    /// the overlay scratch workdir, unmounting anything still mounted
    /// under them first. Called at the end of every run (the chroot is
    /// never persisted) and by the crashed-run sweep.
    pub(crate) fn remove_run_state(&self) {
        for path in [self.chroot_path(), self.work_dir()] {
            unmount_under(&path);
            if let Err(e) = crate::deb::ephemeral::privileged_remove(&path)
                && e.kind() != io::ErrorKind::NotFound
            {
                log::warn!(
                    "Failed to remove the per-run state {} of build \
                     record {}: {}",
                    path.display(),
                    self.manifest().id,
                    e
                );
            }
        }
    }

    /// Record an interrupted build: the outcome is failed, and
    /// `kept_upper` follows the disk — an `--incremental` run keeps
    /// whatever staging created, and a run interrupted before staging
    /// keeps no claim of a staged tree. The per-run state itself is the
    /// SIGINT hook's to remove (unmounting and rm are not safe here, on
    /// the interrupted thread).
    pub(crate) fn record_interrupted(&self, keep_upper: bool) {
        self.set_outcome(OUTCOME_FAILED);
        if upper_retention(keep_upper, self.has_upper()) == UpperRetention::Keep {
            self.with_manifest(|manifest| {
                manifest.kept_upper = true;
            });
        }
    }

    /// Take the advisory lock marking a live build on this record.
    ///
    /// Fails with [`SessionBusy`] when another build holds it. The lock is
    /// released when the handle (and every clone) drops.
    ///
    /// The lock file lives NEXT TO the record root (not inside it): a
    /// teardown removes the root while still holding the lock, and a lock
    /// file inside it would be deleted under the holder — a concurrent
    /// opener would then create a fresh inode and lock nothing.
    pub fn acquire(&self) -> Result<(), SessionBusy> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if state.lock.is_some() {
            return Ok(());
        }
        let file = fs::OpenOptions::new()
            .create(true)
            .write(true)
            .truncate(false)
            .open(self.lock_path())
            .map_err(|e| SessionBusy {
                root: self.root.clone(),
                source: e.to_string(),
            })?;
        let taken = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
        if taken != 0 {
            return Err(SessionBusy {
                root: self.root.clone(),
                source: io::Error::last_os_error().to_string(),
            });
        }
        state.lock = Some(file);
        Ok(())
    }

    /// The advisory lock file path: `<slug dir>/<record id>.lock`, outside
    /// the record root it guards (see [`Session::acquire`]).
    pub(crate) fn lock_path(&self) -> PathBuf {
        let name = self
            .root
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .unwrap_or_default();
        self.root.with_file_name(format!("{name}.{LOCK_NAME}"))
    }

    /// Wipe the persisted overlay upperdir and workdir (used when no
    /// host-tree snapshot exists: object files of an unknown tree must
    /// not leak into the build).
    pub fn wipe_build_artifacts(&self) {
        for dir in [self.upper_dir(), self.work_dir()] {
            if let Err(e) = fs::remove_dir_all(&dir)
                && e.kind() != io::ErrorKind::NotFound
            {
                log::warn!("Failed to remove '{}': {}", dir.display(), e);
            }
        }
    }

    /// Wipe the staged tree inside the chroot (the stable build root).
    ///
    /// Used when overlay staging is unavailable on a forked record: the
    /// fallback re-stages a fresh copy, and copying over the base's tree
    /// would merge stale build artifacts and resurrect host-deleted files
    /// into the build.
    pub fn wipe_staged_tree(&self) -> io::Result<()> {
        crate::deb::ephemeral::clear_dir(&self.chroot_path().join("tmp/pkh-build"))
    }

    /// Remove the whole record tree (unmounting anything mounted under it
    /// first), then unlink the lock file. The caller holds the flock; an
    /// opener that raced ahead of the unlink failed its lock attempt
    /// while we held it, and the next one creates a fresh record.
    pub fn remove(&self) {
        if let Err(e) = remove_session_dir(&self.root) {
            log::warn!(
                "Failed to remove build record {}: {}",
                self.root.display(),
                e
            );
        }
        let _ = fs::remove_file(self.lock_path());
    }
}

/// Error returned when a record is locked by another live build.
#[derive(Debug)]
pub struct SessionBusy {
    /// The locked record root.
    pub root: PathBuf,
    /// Underlying error detail.
    pub source: String,
}

impl std::fmt::Display for SessionBusy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "build record {} is locked by another build",
            self.root.display()
        )
    }
}

impl std::error::Error for SessionBusy {}

/// The sessions directory: `$PKH_SESSIONS_DIR`, defaulting to
/// `/var/tmp/pkh/sessions`.
pub fn sessions_root() -> PathBuf {
    if let Ok(dir) = std::env::var("PKH_SESSIONS_DIR")
        && !dir.is_empty()
    {
        return PathBuf::from(dir);
    }
    PathBuf::from("/var/tmp/pkh/sessions")
}

/// RFC 3339 UTC timestamp for the manifest fields.
fn now_rfc3339() -> String {
    chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}

/// Build-start timestamp used as record id (`YYYYMMDDTHHMMSS`, UTC).
fn now_session_id() -> String {
    chrono::Utc::now().format("%Y%m%dT%H%M%S").to_string()
}

/// Sanitized, hash-suffixed directory name for an identity: human-readable
/// and unique without trusting timestamps.
fn slug_for(identity: &SessionIdentity) -> String {
    let raw = format!(
        "{}_{}_{}_{}",
        identity.package, identity.series, identity.arch, identity.cross
    );
    let sanitized: String = raw
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.') {
                c
            } else {
                '_'
            }
        })
        .collect();
    let mut hasher = Sha256::new();
    hasher.update(raw.as_bytes());
    let hash = hex(&hasher.finalize())[..8].to_string();
    format!("{sanitized}-{hash}")
}

/// Lowercase hex encoding of a digest.
fn hex(digest: &[u8]) -> String {
    digest.iter().map(|b| format!("{b:02x}")).collect()
}

/// SHA-256 of a file, streamed.
pub(crate) fn sha256_file(path: &Path) -> io::Result<String> {
    let mut file = fs::File::open(path)?;
    let mut hasher = Sha256::new();
    io::copy(&mut file, &mut hasher)?;
    Ok(hex(&hasher.finalize()))
}

/// SHA-256 stamp of a directory's content: one digest over every file's
/// relative path and bytes, walked in sorted order. `None` when the
/// directory does not exist.
pub(crate) fn dir_content_stamp(dir: &Path) -> Option<String> {
    let mut hasher = Sha256::new();
    let mut stack = vec![dir.to_path_buf()];
    let mut paths: Vec<PathBuf> = Vec::new();
    while let Some(current) = stack.pop() {
        let Ok(entries) = fs::read_dir(&current) else {
            continue;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if path.is_dir() {
                stack.push(path);
            } else {
                paths.push(path);
            }
        }
    }
    if paths.is_empty() {
        return None;
    }
    paths.sort();
    for path in paths {
        let rel = path.strip_prefix(dir).ok()?;
        hasher.update(rel.to_string_lossy().as_bytes());
        hasher.update([0]);
        if let Ok(content) = fs::read(&path) {
            hasher.update(&content);
        }
        hasher.update([0]);
    }
    Some(hex(&hasher.finalize()))
}

/// Input stamp of the quilt patch phase: a content digest of the host
/// tree's `debian/patches` directory. `None` when the tree carries none.
pub fn patches_stamp(host_tree: &Path) -> Option<String> {
    dir_content_stamp(&host_tree.join("debian/patches"))
}

/// Input stamp of the build-dependency phase: the `debian/control` content
/// plus every selector that changes dependency resolution (arch, cross,
/// PPA list, injected packages, series, pocket).
pub fn control_stamp(host_tree: &Path, selectors: &str) -> Option<String> {
    let control = fs::read(host_tree.join("debian/control")).ok()?;
    let mut hasher = Sha256::new();
    hasher.update(&control);
    hasher.update([0]);
    hasher.update(selectors.as_bytes());
    Some(hex(&hasher.finalize()))
}

/// Persist a manifest atomically.
///
/// A handle may outlive its record (a plain run discards it at the end);
/// such a save must not resurrect the directory, so a vanished root is
/// skipped.
fn save_manifest(root: &Path, manifest: &SessionManifest) -> io::Result<()> {
    if !root.try_exists()? {
        log::debug!(
            "Record root '{}' is gone, skipping the manifest save",
            root.display()
        );
        return Ok(());
    }
    let tmp = root.join(format!("{MANIFEST_NAME}.tmp"));
    let content = serde_json::to_string_pretty(manifest)
        .map_err(|e| io::Error::other(format!("cannot serialize the record manifest: {e}")))?;
    fs::write(&tmp, content)?;
    fs::rename(&tmp, root.join(MANIFEST_NAME))
}

/// Load one record from its root; `None` when there is no readable
/// current-schema manifest (the directory is then inert for discovery
/// and eventually collected by `pkh prune`).
fn open_from_dir(root: PathBuf) -> Option<Session> {
    let content = fs::read_to_string(root.join(MANIFEST_NAME)).ok()?;
    let manifest: SessionManifest = serde_json::from_str(&content).ok()?;
    if manifest.schema != SCHEMA_VERSION {
        return None;
    }
    Some(Session {
        root,
        state: Arc::new(Mutex::new(SessionState {
            manifest,
            lock: None,
        })),
    })
}

/// Error of the record setup paths.
#[derive(Debug)]
pub enum SessionOpenError {
    /// The record is locked by a concurrent build.
    Busy(SessionBusy),
    /// The record could not be forked or created (copy failures,
    /// crashed-build leftovers that even the privileged removal could
    /// not lift).
    Unusable(String),
    /// Anything else (I/O, manifest errors).
    Other(String),
}

impl std::fmt::Display for SessionOpenError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SessionOpenError::Busy(busy) => write!(f, "{busy}"),
            SessionOpenError::Unusable(message) => write!(f, "{message}"),
            SessionOpenError::Other(message) => write!(f, "{message}"),
        }
    }
}

impl std::error::Error for SessionOpenError {}

/// The directory holding every record of `identity` under `root`:
/// `<root>/<slug(identity)>`, with one `<record id>` subdirectory (and
/// its `<record id>.lock`) per build.
pub(crate) fn identity_dir(root: &Path, identity: &SessionIdentity) -> PathBuf {
    root.join(slug_for(identity))
}

/// Root of the shared ccache directories (`/var/tmp/pkh/ccache`), one
/// per identity slug.
pub(crate) const CCACHE_ROOT: &str = "/var/tmp/pkh/ccache";

/// The ccache directory of an identity: `/var/tmp/pkh/ccache/<slug>/`,
/// bind-mounted at each chroot's `/var/cache/pkh-ccache` so compiled
/// objects are shared across runs (content-keyed, see the ccache setup
/// in `crate::deb::local`).
pub(crate) fn ccache_dir(identity: &SessionIdentity) -> PathBuf {
    PathBuf::from(CCACHE_ROOT).join(slug_for(identity))
}

/// A record id not taken yet under `identity_dir`: the plain timestamp,
/// suffixed `-02`, `-03`, ... when earlier builds took the same second
/// (zero-padded so the ids keep sorting in creation order).
///
/// Ids are allocated strictly beyond the highest one of the second —
/// freed slots are never reused: a re-created `stamp` while `stamp-02`,
/// ... of the same second still exist would sort before them and lie
/// about its creation order (the keep-last-5 pruning and the base
/// selection both order by id within one second).
fn unique_record_id(identity_dir: &Path) -> String {
    let stamp = now_session_id();
    let suffix_prefix = format!("{stamp}-");
    // The plain stamp occupies slot 1 of the second.
    let mut highest = if identity_dir.join(&stamp).exists() {
        1
    } else {
        0
    };
    if let Ok(entries) = fs::read_dir(identity_dir) {
        for entry in entries.flatten() {
            let Some(suffix) = entry
                .file_name()
                .to_string_lossy()
                .strip_prefix(&suffix_prefix)
                .and_then(|suffix| suffix.parse::<usize>().ok())
            else {
                continue;
            };
            highest = highest.max(suffix);
        }
    }
    if highest == 0 {
        stamp
    } else {
        format!("{stamp}-{:02}", highest + 1)
    }
}

/// Reserve a new record directory of `identity` under `root` and lock it:
/// the directory is created, its flock taken (so concurrent builds and
/// prune sweeps see it as live), the manifest — built by `manifest` from
/// the reserved id — persisted, and the build log started. The loop
/// absorbs the same-second collisions with concurrent record creations.
fn reserve_record(
    root: &Path,
    identity: &SessionIdentity,
    manifest: impl Fn(String) -> SessionManifest,
) -> Result<Session, SessionOpenError> {
    let dir = identity_dir(root, identity);
    fs::create_dir_all(&dir)
        .map_err(|e| SessionOpenError::Other(format!("cannot create {}: {e}", dir.display())))?;
    for _ in 0..16 {
        let id = unique_record_id(&dir);
        let record_dir = dir.join(&id);
        if fs::create_dir(&record_dir).is_err() {
            continue; // raced with a concurrent creation; take another id
        }
        let session = Session {
            root: record_dir,
            state: Arc::new(Mutex::new(SessionState {
                manifest: manifest(id),
                lock: None,
            })),
        };
        match session.acquire() {
            Ok(()) => {
                session.save().map_err(|e| {
                    SessionOpenError::Other(format!(
                        "cannot write the manifest of the new build record: {e}"
                    ))
                })?;
                // The log exists from creation on, so a build that
                // fails before any captured line still leaves one.
                fs::create_dir_all(session.root().join(LOG_DIR))
                    .and_then(|()| {
                        fs::OpenOptions::new()
                            .create(true)
                            .append(true)
                            .open(session.log_file())
                    })
                    .map_err(|e| {
                        SessionOpenError::Other(format!(
                            "cannot start the build log of the new record: {e}"
                        ))
                    })?;
                return Ok(session);
            }
            // Someone stole this directory between its creation and the
            // lock: impossible in practice (they cannot know the id
            // before it exists), but move on rather than fail.
            Err(_) => continue,
        }
    }
    Err(SessionOpenError::Other(
        "cannot reserve a fresh build record id: too many concurrent builds".to_string(),
    ))
}

/// A manifest for a record built from scratch: no base, no journal —
/// the run fills the tarball sha (at bootstrap) and the outcome (at its
/// end).
fn fresh_manifest(
    id: String,
    host_tree: &Path,
    identity: &SessionIdentity,
    tree_version: &str,
) -> SessionManifest {
    SessionManifest {
        schema: SCHEMA_VERSION,
        pkh: env!("CARGO_PKG_VERSION").to_string(),
        id,
        host_tree: host_tree.to_string_lossy().into_owned(),
        identity: identity.clone(),
        created: now_rfc3339(),
        last_used: now_rfc3339(),
        last_outcome: OUTCOME_RUNNING.to_string(),
        base: None,
        tarball_sha256: String::new(),
        phases: BTreeMap::new(),
        tree_version: tree_version.to_string(),
        kept_upper: false,
        build_resume: false,
        timings: None,
        profile: None,
    }
}

/// Create the record a `pkh deb` build runs in: a brand-new, empty
/// record of `identity`, locked for this build (the caller decides
/// whether its upperdir is retained at the end). Creating one also
/// sweeps the identity's crashed-run leftovers (chroot directories no
/// live build holds).
pub fn new_record(
    host_tree: &Path,
    identity: &SessionIdentity,
    tree_version: &str,
) -> Result<Session, SessionOpenError> {
    new_record_in(&sessions_root(), host_tree, identity, tree_version)
}

/// The testable core of [`new_record`] over an explicit sessions root.
pub(crate) fn new_record_in(
    root: &Path,
    host_tree: &Path,
    identity: &SessionIdentity,
    tree_version: &str,
) -> Result<Session, SessionOpenError> {
    sweep_identity_run_state(root, identity);
    reserve_record(root, identity, |id| {
        fresh_manifest(id, host_tree, identity, tree_version)
    })
}

/// `cp -a --reflink=auto` of `src`'s contents into `dst`: a full,
/// metadata-preserving copy (reflinks keep it cheap on CoW filesystems).
/// The overlay upperdir can hold kernel-created root-owned whiteouts
/// (char devices), so a plain-copy failure is retried privileged.
fn copy_dir(src: &Path, dst: &Path) -> io::Result<()> {
    let is_root = crate::utils::root::is_root().unwrap_or(false);
    // `src/.` into `dst`: an existing `dst` would otherwise receive a
    // nested `dst/<src name>` (cp's copy-into semantics for existing
    // destinations), and the fork's record root already exists.
    fs::create_dir_all(dst)?;
    let from = format!("{}/.", src.display());
    let run = |privileged: bool| {
        crate::deb::ephemeral::privileged_command("cp", privileged || is_root)
            .arg("-a")
            .arg("--reflink=auto")
            .arg(&from)
            .arg(dst)
            .status()
    };
    let status = match run(false) {
        Ok(status) if status.success() => return Ok(()),
        outcome => {
            log::debug!(
                "Unprivileged copy of {} failed ({outcome:?}); retrying privileged",
                src.display()
            );
            run(true)
        }
    };
    if matches!(&status, Ok(s) if s.success()) {
        Ok(())
    } else {
        Err(io::Error::other(format!(
            "cp -a {} {} exited with {status:?}",
            src.display(),
            dst.display()
        )))
    }
}

/// Fork `base` into a new record of `identity` under the sessions root.
/// See [`fork_record_in`].
pub fn fork_record(
    base: &Session,
    host_tree: &Path,
    identity: &SessionIdentity,
    version: &str,
    cache_tarball_sha256: &str,
) -> Result<Session, SessionOpenError> {
    fork_record_in(
        &sessions_root(),
        base,
        host_tree,
        identity,
        version,
        cache_tarball_sha256,
    )
}

/// The testable core of [`fork_record`] over an explicit sessions root.
///
/// A fork carries the base's tree-upper (and its host-tree snapshot and
/// tree journal) only through the fork gate: the base's recorded
/// tarball sha must still equal `cache_tarball_sha256` — the digest of
/// the chroot tarball this build would extract, empty when the cache
/// does not hold it yet — and the changelog version must be unchanged.
/// The toolchain moving or the tree changing invalidates the objects;
/// the fork then starts from a fresh upperdir. The base itself is never
/// modified, and the copy happens while holding the base's flock (a
/// concurrent build may still be writing into its upperdir).
pub(crate) fn fork_record_in(
    root: &Path,
    base: &Session,
    host_tree: &Path,
    identity: &SessionIdentity,
    version: &str,
    cache_tarball_sha256: &str,
) -> Result<Session, SessionOpenError> {
    let base_manifest = base.manifest();
    if base_manifest.identity != *identity {
        return Err(SessionOpenError::Other(format!(
            "cannot fork record {}: it was recorded for {}, but this build targets {}",
            base_manifest.id, base_manifest.identity, identity
        )));
    }
    base.acquire().map_err(SessionOpenError::Busy)?;

    let carry_upper = base.has_upper()
        && !base_manifest.tarball_sha256.is_empty()
        && base_manifest.tarball_sha256 == cache_tarball_sha256
        && base_manifest.tree_version == version;
    if base.has_upper() && !carry_upper {
        let why = if base_manifest.tree_version != version {
            format!(
                "the changelog version changed ({} -> {})",
                base_manifest.tree_version, version
            )
        } else if base_manifest.tarball_sha256 != cache_tarball_sha256 {
            "the chroot tarball changed since the base was built".to_string()
        } else {
            "the base carries no staged tree".to_string()
        };
        log::info!(
            "Fork of record {} starts from a fresh staged tree: {why}",
            base_manifest.id
        );
    }

    sweep_identity_run_state(root, identity);
    let record = reserve_record(root, identity, |id| SessionManifest {
        schema: SCHEMA_VERSION,
        pkh: env!("CARGO_PKG_VERSION").to_string(),
        id,
        host_tree: host_tree.to_string_lossy().into_owned(),
        identity: identity.clone(),
        created: now_rfc3339(),
        last_used: now_rfc3339(),
        last_outcome: OUTCOME_RUNNING.to_string(),
        base: Some(base_manifest.id.clone()),
        tarball_sha256: String::new(),
        // The journal only makes sense with the tree state it
        // describes: a fresh upperdir starts cold.
        phases: if carry_upper {
            base_manifest.phases.clone()
        } else {
            BTreeMap::new()
        },
        tree_version: version.to_string(),
        kept_upper: false,
        build_resume: false,
        timings: None,
        profile: None,
    })?;

    if carry_upper {
        let copied = copy_dir(&base.upper_dir(), &record.upper_dir()).and_then(|_bytes| {
            if base.snapshot_file().is_file() {
                fs::copy(base.snapshot_file(), record.snapshot_file())
            } else {
                Ok(0)
            }
        });
        if let Err(e) = copied {
            // The partial fork is inert (its manifest is written but
            // its staged tree is incomplete): remove it rather than
            // leave a trap for the next selection.
            let _ = remove_session_dir(record.root());
            return Err(SessionOpenError::Unusable(format!(
                "cannot copy the staged tree of record {} into {}: {e}. \
                 Free the space or remove old records with `pkh prune`, \
                 then rebuild",
                base.root().display(),
                record.root().display()
            )));
        }
    }
    log::info!(
        "Forked build record {} from record {}: {}",
        record.manifest().id,
        base_manifest.id,
        if carry_upper {
            "the staged tree (objects included) is carried"
        } else {
            "a fresh staged tree; the environment comes from the apt cache"
        }
    );
    Ok(record)
}

/// Sweep the per-run leftovers of a hard-killed build (OOM, SIGKILL —
/// anything that skipped the teardown): the chroot directory and the
/// overlay scratch workdir of every record of `identity` whose lock is
/// not held by a live build. Live builds keep theirs; nothing else is
/// touched.
pub(crate) fn sweep_identity_run_state(root: &Path, identity: &SessionIdentity) {
    let Ok(entries) = fs::read_dir(identity_dir(root, identity)) else {
        return;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() || (!path.join(CHROOT_DIR).is_dir() && !path.join(WORK_DIR).is_dir()) {
            continue;
        }
        let Some(record) = open_from_dir(path) else {
            continue;
        };
        if lock_is_held(&record.lock_path()).unwrap_or(true) {
            continue;
        }
        log::debug!(
            "Sweeping the crashed-run leftovers of build record {}",
            record.manifest().id
        );
        record.remove_run_state();
    }
}

/// Unmount every mount at or under `path` (deepest first), best-effort:
/// used before removing per-run state and record roots.
pub(crate) fn unmount_under(path: &Path) {
    let is_root = crate::utils::root::is_root().unwrap_or(false);
    for mount in crate::deb::ephemeral::host_mounts_under(path)
        .into_iter()
        .rev()
    {
        if !crate::deb::ephemeral::unmount_path(&mount, is_root) {
            log::warn!(
                "Failed to unmount {} under {}",
                mount.display(),
                path.display()
            );
        }
    }
}

/// Newness ordering of two records: by `created`, then id (both are
/// build-start timestamps of the same format).
fn record_ordering(a: &Session, b: &Session) -> std::cmp::Ordering {
    let a = a.manifest();
    let b = b.manifest();
    a.created.cmp(&b.created).then(a.id.cmp(&b.id))
}

/// Whether the lock file at `path` is currently held by a live build
/// (flock taken). `Ok(false)` also covers a missing file.
pub(crate) fn lock_is_held(path: &Path) -> io::Result<bool> {
    use std::os::unix::io::AsRawFd;
    let file = match fs::OpenOptions::new().read(true).open(path) {
        Ok(file) => file,
        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(false),
        Err(e) => return Err(e),
    };
    let taken = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
    if taken == 0 {
        // Nobody holds it: release the probe's own lock and report free.
        unsafe {
            libc::flock(file.as_raw_fd(), libc::LOCK_UN);
        }
        return Ok(false);
    }
    let err = io::Error::last_os_error();
    if err.kind() == io::ErrorKind::WouldBlock {
        return Ok(true);
    }
    Err(err)
}

/// The digest of the chroot tarball a build of `series`/`arch` would
/// extract, empty when the cache does not hold one yet (the bootstrap
/// downloads it during the run). This is the fork gate's reference
/// against the sha recorded in a base's manifest. `mirror` keys the
/// cache (see [`crate::deb::ephemeral::chroot_tarball_path`]), so records
/// never fork across a mirror change.
pub(crate) fn cache_tarball_sha256(
    series: &str,
    arch: Option<&str>,
    mirror: Option<&str>,
) -> String {
    crate::deb::ephemeral::chroot_tarball_path(series, arch, mirror)
        .ok()
        .and_then(|path| sha256_file(&path).ok())
        .unwrap_or_default()
}

/// Load every discoverable record under `root` (any identity), newest
/// first. `root` is the sessions root: `<root>/<slug>/<record>/`.
pub(crate) fn load_all_in(root: &Path) -> Vec<Session> {
    let mut records = Vec::new();
    let Ok(identities) = fs::read_dir(root) else {
        return records;
    };
    for identity_dir in identities.flatten().map(|entry| entry.path()) {
        if !identity_dir.is_dir() {
            continue;
        }
        let Ok(record_dirs) = fs::read_dir(&identity_dir) else {
            continue;
        };
        for record_dir in record_dirs.flatten().map(|entry| entry.path()) {
            if record_dir.is_dir()
                && let Some(record) = open_from_dir(record_dir)
            {
                records.push(record);
            }
        }
    }
    records.sort_by(|a, b| record_ordering(a, b).reverse());
    records
}

/// The records of `identity` under `root`, newest first.
pub(crate) fn identity_records_in(root: &Path, identity: &SessionIdentity) -> Vec<Session> {
    let dir = identity_dir(root, identity);
    let Ok(entries) = fs::read_dir(&dir) else {
        return Vec::new();
    };
    let mut records: Vec<Session> = entries
        .flatten()
        .map(|entry| entry.path())
        .filter(|path| path.is_dir())
        .filter_map(open_from_dir)
        .filter(|record| record.manifest().identity == *identity)
        .collect();
    records.sort_by(|a, b| record_ordering(a, b).reverse());
    records
}

/// The fork candidates of `identity` under `root`: its newest records
/// that still carry a tree-upper. A metadata-only record (plain run, or
/// degraded by the upper cap) holds nothing a fork could carry.
pub(crate) fn fork_candidates_in(root: &Path, identity: &SessionIdentity) -> Vec<Session> {
    identity_records_in(root, identity)
        .into_iter()
        .filter(|record| record.has_upper())
        .collect()
}

/// Records recorded from `tree` (any identity), newest first.
pub fn for_tree(tree: &Path) -> Vec<Session> {
    let tree = tree.to_string_lossy().into_owned();
    load_all_in(&sessions_root())
        .into_iter()
        .filter(|s| s.manifest().host_tree == tree)
        .collect()
}

/// Records whose id starts with `prefix` (the `pkh deb --incremental <id>`
/// lookup). Ambiguous prefixes yield several results; the caller rejects
/// them.
pub(crate) fn by_id_prefix_in(root: &Path, prefix: &str) -> Vec<Session> {
    load_all_in(root)
        .into_iter()
        .filter(|s| s.manifest().id.starts_with(prefix))
        .collect()
}

/// Order the `--incremental` base candidates, best first: the usable
/// records are given newest-first, and the successful ones come before
/// the rest (each group keeps its newest-first order). The caller tries
/// them in order until one accepts the fork, so a busy or unworkable
/// favorite falls through to the next-best base.
pub(crate) fn rank_base_candidates(candidates: Vec<Session>) -> Vec<Session> {
    let mut ranked: Vec<Session> = candidates
        .iter()
        .filter(|s| s.manifest().last_outcome == OUTCOME_SUCCESS)
        .cloned()
        .collect();
    ranked.extend(
        candidates
            .into_iter()
            .filter(|s| s.manifest().last_outcome != OUTCOME_SUCCESS),
    );
    ranked
}

/// One row of the `pkh deb list` table.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionListRow {
    /// Record id (build-start timestamp).
    pub id: String,
    /// Source package name.
    pub package: String,
    /// Changelog version of this record's build.
    pub version: String,
    /// `series/arch` target, with the cross marker folded into the arch.
    pub target: String,
    /// Id of the record this one was forked from, when any.
    pub base: Option<String>,
    /// Whether the record retains its staged tree (an `--incremental`
    /// fork candidate).
    pub upper: bool,
    /// Outcome of this record's build.
    pub outcome: String,
    /// Duration of the record's dependency installation, formatted
    /// ("-" when the record carries none).
    pub deps: String,
    /// Duration of the record's package build, formatted ("-" when the
    /// record carries none).
    pub build: String,
    /// Human-readable age of this record's build.
    pub age: String,
}

/// The rows `pkh deb list` shows for a tree, newest first.
pub fn list_for_tree(tree: &Path) -> Vec<SessionListRow> {
    for_tree(tree)
        .into_iter()
        .map(|session| {
            let manifest = session.manifest();
            let cross = if manifest.identity.cross { "*" } else { "" };
            let (deps, build) = match manifest.timings {
                Some(timings) => (
                    format_duration(timings.deps_secs),
                    format_duration(timings.build_secs),
                ),
                None => ("-".to_string(), "-".to_string()),
            };
            SessionListRow {
                id: manifest.id,
                package: manifest.identity.package,
                version: manifest.tree_version,
                target: format!(
                    "{}/{}{}",
                    manifest.identity.series, manifest.identity.arch, cross
                ),
                base: manifest.base,
                upper: session.has_upper(),
                outcome: manifest.last_outcome,
                deps,
                build,
                age: age_of(&manifest.last_used),
            }
        })
        .collect()
}

/// Human-readable age of an RFC 3339 timestamp ("2h", "3d"); "unknown"
/// when it cannot be parsed.
fn age_of(rfc3339: &str) -> String {
    let Ok(then) =
        chrono::DateTime::parse_from_rfc3339(rfc3339).map(|then| then.with_timezone(&chrono::Utc))
    else {
        return "unknown".to_string();
    };
    let Ok(elapsed) = (chrono::Utc::now() - then).to_std() else {
        return "now".to_string();
    };
    let secs = elapsed.as_secs();
    match secs {
        0 => "now".to_string(),
        1..=59 => format!("{secs}s"),
        60..=3599 => format!("{}m", secs / 60),
        3600..=86399 => format!("{}h", secs / 3600),
        _ => format!("{}d", secs / 86400),
    }
}

/// Human-readable duration of a build phase ("45s", "2m41s", "1h02m",
/// "3d04h"): unlike [`age_of`] the minutes matter at every scale.
fn format_duration(secs: u64) -> String {
    match secs {
        0..=59 => format!("{secs}s"),
        60..=3599 => format!("{}m{:02}s", secs / 60, secs % 60),
        3600..=86399 => format!("{}h{:02}m", secs / 3600, (secs % 3600) / 60),
        _ => format!("{}d{:02}h", secs / 86400, (secs % 86400) / 3600),
    }
}

/// Human-readable duration of a profiled phase: the phases can be
/// shorter than a second (artifact retrieval), so sub-second entries
/// render as milliseconds and the rest through [`format_duration`].
fn format_phase_millis(millis: u64) -> String {
    if millis < 1000 {
        format!("{millis}ms")
    } else {
        format_duration(millis / 1000)
    }
}

/// Render `pkh deb profile` for `record`: its identity in the header,
/// one row per recorded phase with its duration and its share of the
/// total, and — for a failed build — the phase the error surfaced
/// from. A record without a profile (predating the field, or a build
/// interrupted before any phase completed) says so.
pub fn render_record_profile(record: &Session) -> String {
    let manifest = record.manifest();
    let cross = if manifest.identity.cross { "*" } else { "" };
    let mut out = format!(
        "Profile of build {} — {} {}, {}/{}{}, {}",
        manifest.id,
        manifest.identity.package,
        manifest.tree_version,
        manifest.identity.series,
        manifest.identity.arch,
        cross,
        manifest.last_outcome
    );
    if let Some(base) = &manifest.base {
        out.push_str(&format!(" (fork of {base})"));
    }
    out.push_str(":\n");

    let Some(profile) = &manifest.profile else {
        out.push_str("No phase profile recorded for this build.\n");
        return out;
    };
    if profile.is_empty() {
        out.push_str("No phase profile recorded for this build.\n");
        return out;
    }

    let total = profile.total_millis();
    let rows: Vec<Vec<String>> = profile
        .phases
        .iter()
        .map(|entry| {
            let label = crate::deb::Phase::from_key(&entry.phase)
                .map(|phase| phase.label().to_string())
                .unwrap_or_else(|| entry.phase.clone());
            let share = entry
                .millis
                .checked_mul(100)
                .and_then(|scaled| scaled.checked_div(total.max(1)))
                .map(|share| format!("{share}%"))
                .unwrap_or_else(|| "-".to_string());
            vec![label, format_phase_millis(entry.millis), share]
        })
        .collect();
    let headers = ["PHASE", "TIME", "SHARE"];
    let widths: Vec<usize> = headers
        .iter()
        .enumerate()
        .map(|(i, header)| {
            rows.iter()
                .map(|row| row[i].chars().count())
                .max()
                .unwrap_or(0)
                .max(header.len())
        })
        .collect();
    let line = |cells: &[String]| {
        cells
            .iter()
            .enumerate()
            .map(|(i, cell)| {
                if i == 0 {
                    format!("{cell:<width$}", width = widths[i])
                } else {
                    format!("{cell:>width$}", width = widths[i])
                }
            })
            .collect::<Vec<_>>()
            .join("  ")
    };
    out.push_str(&line(
        &headers.iter().map(|h| h.to_string()).collect::<Vec<_>>(),
    ));
    out.push('\n');
    for row in &rows {
        out.push_str(&line(row));
        out.push('\n');
    }
    out.push_str(&line(&[
        "Total".to_string(),
        format_phase_millis(total),
        String::new(),
    ]));
    out.push('\n');
    if manifest.last_outcome == OUTCOME_FAILED
        && let Some(last) = profile.phases.last()
        && let Some(phase) = crate::deb::Phase::from_key(&last.phase)
    {
        out.push_str(&format!("The build failed during \"{}\".\n", phase.label()));
    }
    out
}

/// Render the `pkh deb list` table.
pub fn render_session_list(tree: &Path, rows: &[SessionListRow]) -> String {
    let mut out = String::new();
    if rows.is_empty() {
        out.push_str(&format!(
            "No build records recorded from {}.\n",
            tree.display()
        ));
        return out;
    }
    out.push_str(&format!("Records for {}:\n", tree.display()));
    let headers = [
        "ID", "PACKAGE", "VERSION", "TARGET", "BASE", "UPPER", "OUTCOME", "DEPS", "BUILD", "AGE",
    ];
    let table: Vec<Vec<String>> = rows
        .iter()
        .map(|row| {
            vec![
                row.id.clone(),
                row.package.clone(),
                row.version.clone(),
                row.target.clone(),
                row.base.clone().unwrap_or_else(|| "-".to_string()),
                if row.upper { "yes" } else { "-" }.to_string(),
                row.outcome.clone(),
                row.deps.clone(),
                row.build.clone(),
                row.age.clone(),
            ]
        })
        .collect();
    let widths: Vec<usize> = headers
        .iter()
        .enumerate()
        .map(|(i, header)| {
            table
                .iter()
                .map(|row| row[i].chars().count())
                .max()
                .unwrap_or(0)
                .max(header.len())
        })
        .collect();
    let line = |cells: &[String]| {
        cells
            .iter()
            .enumerate()
            .map(|(i, cell)| format!("{cell:<width$}", width = widths[i]))
            .collect::<Vec<_>>()
            .join("  ")
            .trim_end()
            .to_string()
    };
    out.push_str(&line(
        &headers.iter().map(|h| h.to_string()).collect::<Vec<_>>(),
    ));
    out.push('\n');
    for row in &table {
        out.push_str(&line(row));
        out.push('\n');
    }
    out.push_str(
        "\nFork with: pkh deb --incremental [<id>]  (* = cross build; \
         UPPER = the record keeps its staged tree)\n",
    );
    out
}

// ---------------------------------------------------------------------------
// Build log (`pkh deb log`)
// ---------------------------------------------------------------------------

/// Resolve the record a read-only deb query targets (`pkh deb log`,
/// `pkh deb profile`): with no id, the newest record of `tree`; with
/// an id (or unique prefix), that exact record of `tree`. `root` is
/// the sessions root (`sessions_root()` in production, a fixture root
/// in tests).
pub fn find_record(
    root: &Path,
    tree: &Path,
    id: Option<&str>,
) -> Result<Session, Box<dyn std::error::Error>> {
    let tree_str = tree.to_string_lossy().into_owned();
    match id {
        None => load_all_in(root)
            .into_iter()
            .find(|s| s.manifest().host_tree == tree_str)
            .ok_or_else(|| format!("no build record found for {}", tree.display()).into()),
        Some(id) => {
            let mut matches: Vec<Session> = by_id_prefix_in(root, id)
                .into_iter()
                .filter(|s| s.manifest().host_tree == tree_str)
                .collect();
            match matches.len() {
                1 => Ok(matches.remove(0)),
                0 => Err(format!(
                    "no build record matches id '{id}' for this tree \
                     (see `pkh deb list`)"
                )
                .into()),
                _ => {
                    let found = matches
                        .iter()
                        .map(|s| s.manifest().id.clone())
                        .collect::<Vec<_>>()
                        .join(", ");
                    Err(format!("ambiguous record id '{id}': matches {found}").into())
                }
            }
        }
    }
}

/// Stream a record's build log to `out`: everything written so far and,
/// with `follow`, whatever the build appends until its record reaches a
/// terminal state (outcome recorded, or the record's lock freed with the
/// outcome still `running` — a crashed build). A record whose log does
/// not exist yet streams nothing.
pub fn print_record_log(record: &Session, follow: bool, out: &mut dyn io::Write) -> io::Result<()> {
    let path = record.log_file();
    let mut offset: usize = 0;
    loop {
        offset = stream_appended(&path, offset, out)?;
        let done = open_from_dir(record.root().to_path_buf())
            .map(|live| {
                live.manifest().last_outcome != OUTCOME_RUNNING
                    || !lock_is_held(&live.lock_path()).unwrap_or(true)
            })
            .unwrap_or(true);
        if !follow || done {
            return out.flush();
        }
        std::thread::sleep(Duration::from_millis(200));
    }
}

/// Write the bytes of `path` beyond `offset` to `out`, returning the
/// new offset; a missing or shrunk file restarts from zero.
fn stream_appended(path: &Path, offset: usize, out: &mut dyn io::Write) -> io::Result<usize> {
    let Ok(meta) = fs::metadata(path) else {
        return Ok(offset);
    };
    let len = meta.len() as usize;
    if len < offset {
        return Ok(len); // truncated/rotated: restart from the beginning
    }
    if len == offset {
        return Ok(offset);
    }
    let mut file = fs::File::open(path)?;
    file.seek(io::SeekFrom::Start(offset as u64))?;
    io::copy(&mut file, out)?;
    Ok(len)
}

/// A sink appending every captured build line to a record's build log
/// (opened once, shared by every command of the run) and forwarding it
/// to the wrapped sink, if any: the log is a full tee of the build
/// output whatever view renders it.
pub(crate) struct RecordLogSink {
    log: Mutex<fs::File>,
    next: Option<Arc<dyn LineSink>>,
}

impl RecordLogSink {
    /// Open the build log of `record` for appending, forwarding to `next`
    /// as well (the view's own sink, when it has one).
    pub(crate) fn new(record: &Session, next: Option<Arc<dyn LineSink>>) -> io::Result<Self> {
        Ok(Self {
            log: Mutex::new(
                fs::OpenOptions::new()
                    .append(true)
                    .open(record.log_file())?,
            ),
            next,
        })
    }
}

impl LineSink for RecordLogSink {
    fn line(&self, stream: Stream, line: &str) {
        if let Ok(mut log) = self.log.lock() {
            let _ = writeln!(log, "{line}");
        }
        if let Some(next) = &self.next {
            next.line(stream, line);
        }
    }
}

/// A sink writing captured lines to two raw writers, one per stream:
/// stdout lines to the first, stderr lines to the second.
///
/// This is the record-log view's stand-in for the subprocess-inherit
/// fallback: with `--verbose` (or a piped stdout, where the live view
/// disables itself) the caller's view is [`crate::report::Quiet`], whose
/// sink is `None` — the runner would let the subprocess inherit the
/// terminal — but the record-log wrapper always provides a sink, so
/// the fallback is never reached anymore. Wrapping this sink in its
/// place keeps the raw lines on the terminal while the record's
/// build.log keeps the full tee.
pub(crate) struct RawWriterSink {
    stdout: Mutex<Box<dyn Write + Send>>,
    stderr: Mutex<Box<dyn Write + Send>>,
}

impl RawWriterSink {
    /// A sink over explicit writer pairs (tests capture into memory).
    pub(crate) fn new(stdout: Box<dyn Write + Send>, stderr: Box<dyn Write + Send>) -> Self {
        Self {
            stdout: Mutex::new(stdout),
            stderr: Mutex::new(stderr),
        }
    }

    /// A sink over the real terminal: the process's stdout and stderr.
    pub(crate) fn terminal() -> Self {
        Self::new(Box::new(io::stdout()), Box::new(io::stderr()))
    }
}

impl LineSink for RawWriterSink {
    fn line(&self, stream: Stream, line: &str) {
        let writer = match stream {
            Stream::Stdout => &self.stdout,
            Stream::Stderr => &self.stderr,
        };
        if let Ok(mut out) = writer.lock() {
            let _ = writeln!(out, "{line}");
        }
    }
}

// ---------------------------------------------------------------------------
// Host-tree snapshot and overlay upperdir sync
//
// On resume the staged tree is re-mounted as an overlay over the host tree
// with the SAME upperdir as the previous attempt, so build artifacts written
// inside the chroot (object files) survive while host-side edits show
// through the live lowerdir. Two things do not propagate through a plain
// re-mount, and the snapshot fixes both:
//
// - host-side deletions: a file the user deleted on the host still appears
//   in the merged view as long as the upperdir holds a copy of it (e.g.
//   written by quilt);
// - host-side modifications of files the upperdir shadows: the stale upper
//   copy would keep winning over the edited lower one.
//
// The snapshot records the host tree (metadata only) at staging time; the
// sync below drops upper entries for paths the host has since deleted or
// modified, keeps everything else (the build artifacts), and refreshes the
// snapshot for the next attempt.
// ---------------------------------------------------------------------------

/// Metadata of one host-tree entry, for change detection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct EntryMeta {
    /// Modification time in nanoseconds since the Unix epoch.
    pub mtime_ns: i128,
    /// Size in bytes (symlinks: target length).
    pub size: u64,
}

impl EntryMeta {
    fn from_metadata(metadata: &fs::Metadata) -> Self {
        let mtime_ns = metadata
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_nanos() as i128)
            .unwrap_or(0);
        EntryMeta {
            mtime_ns,
            size: metadata.len(),
        }
    }
}

/// Walk a host tree, mapping slash-separated relative paths to metadata.
/// Symlinks are recorded as themselves (not followed).
///
/// Best-effort below the root: a directory that cannot be read or an
/// entry that cannot be stat'd (a root-owned sibling tree in the staged
/// parent, a path deleted by a race mid-walk) is warned about and
/// skipped — the walk runs over the staged PARENT, which contains trees
/// unrelated to this build, and one unreadable path must not abort the
/// build. Entries absent from the map are simply untracked for the
/// whiteout sync, the already-handled case. Only the ROOT itself being
/// unreadable propagates: the package tree being unreadable is a real
/// error.
pub(crate) fn walk_tree_meta(root: &Path) -> io::Result<BTreeMap<String, EntryMeta>> {
    let mut map = BTreeMap::new();
    let mut stack = vec![root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let entries = match fs::read_dir(&dir) {
            Ok(entries) => entries,
            Err(e) => {
                if dir == root {
                    return Err(e);
                }
                log::warn!("skipping unreadable path {}: {e}", dir.display());
                continue;
            }
        };
        for entry in entries.flatten() {
            let path = entry.path();
            let metadata = match fs::symlink_metadata(&path) {
                Ok(metadata) => metadata,
                // A race deleted the entry between read_dir and stat.
                Err(e) => {
                    log::warn!("skipping unreadable path {}: {e}", path.display());
                    continue;
                }
            };
            if metadata.is_dir() {
                stack.push(path);
            } else {
                let rel = path
                    .strip_prefix(root)
                    .map_err(|e| io::Error::other(format!("cannot relativize '{path:?}': {e}")))?
                    .to_string_lossy()
                    .into_owned();
                map.insert(rel, EntryMeta::from_metadata(&metadata));
            }
        }
    }
    Ok(map)
}

/// Write the snapshot file for `meta`.
pub(crate) fn write_snapshot(meta: &BTreeMap<String, EntryMeta>, path: &Path) -> io::Result<()> {
    let mut content = String::new();
    for (rel, entry) in meta {
        content.push_str(&format!("f\t{}\t{}\t{rel}\n", entry.mtime_ns, entry.size));
    }
    fs::write(path, content)
}

/// Load the snapshot file; `None` when it is missing or unreadable.
pub(crate) fn load_snapshot(path: &Path) -> Option<BTreeMap<String, EntryMeta>> {
    let content = fs::read_to_string(path).ok()?;
    let mut map = BTreeMap::new();
    for line in content.lines() {
        let mut fields = line.split('\t');
        let _kind = fields.next()?;
        let mtime_ns = fields.next()?.parse::<i128>().ok()?;
        let size = fields.next()?.parse::<u64>().ok()?;
        let rel = fields.next()?.to_string();
        map.insert(rel, EntryMeta { mtime_ns, size });
    }
    Some(map)
}

/// Record the host tree snapshot for a fresh staging.
pub(crate) fn take_snapshot(host_tree: &Path, session: &Session) -> io::Result<()> {
    let meta = walk_tree_meta(host_tree)?;
    write_snapshot(&meta, &session.snapshot_file())
}

/// One kind of entry found in a persisted overlay upperdir.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum UpperKind {
    /// Overlayfs whiteout (char device 0:0) hiding a lower entry.
    Whiteout,
    /// Directory.
    Dir,
    /// Regular file, symlink or any other non-directory entry.
    File,
}

/// One entry of the upperdir walk, relative to the merge root.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct UpperEntry {
    /// Slash-separated path relative to the merge root.
    pub rel: String,
    /// Entry kind.
    pub kind: UpperKind,
}

/// One change to apply to the upperdir before re-mounting it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum UpperOp {
    /// Remove `rel` from the upperdir; `dir` removes the subtree.
    Remove { rel: String, dir: bool },
}

/// Walk the upperdir (pre-order: parents before children), classifying
/// whiteouts.
pub(crate) fn walk_upper(upper_root: &Path) -> io::Result<Vec<UpperEntry>> {
    let mut entries = Vec::new();
    let mut stack = vec![upper_root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        for entry in fs::read_dir(&dir)?.flatten() {
            let path = entry.path();
            let metadata = fs::symlink_metadata(&path)?;
            let rel = path
                .strip_prefix(upper_root)
                .map_err(|e| io::Error::other(format!("cannot relativize '{path:?}': {e}")))?
                .to_string_lossy()
                .into_owned();
            if metadata.file_type().is_char_device() && metadata.rdev() == 0 {
                entries.push(UpperEntry {
                    rel,
                    kind: UpperKind::Whiteout,
                });
            } else if metadata.is_dir() {
                entries.push(UpperEntry {
                    rel: rel.clone(),
                    kind: UpperKind::Dir,
                });
                stack.push(path);
            } else {
                entries.push(UpperEntry {
                    rel,
                    kind: UpperKind::File,
                });
            }
        }
    }
    Ok(entries)
}

/// Pure core of the resume sync: given the host tree at staging time
/// (`snapshot`), the host tree now (`current`) and the upperdir contents,
/// decide which upper entries must go. The rules:
///
/// - a whiteout is kept only when it still hides the file it hid at
///   staging time (present and unchanged on the host); a host modification
///   or re-creation must win;
/// - a directory the host deleted takes its whole upper subtree with it;
/// - an upper file whose host original was deleted or modified is dropped
///   (the build artifact cases — no host original — are kept).
pub(crate) fn plan_upper_sync(
    snapshot: &BTreeMap<String, EntryMeta>,
    current: &BTreeMap<String, EntryMeta>,
    upper: &[UpperEntry],
) -> Vec<UpperOp> {
    let mut ops = Vec::new();
    for entry in upper {
        match entry.kind {
            UpperKind::Whiteout => {
                // Keep only the still-accurate hiding of an unchanged host
                // file; anything else (host edited, re-created or removed
                // the path) makes the whiteout stale or wrong.
                let still_accurate = snapshot
                    .get(&entry.rel)
                    .is_some_and(|staged| current.get(&entry.rel) == Some(staged));
                if !still_accurate {
                    ops.push(UpperOp::Remove {
                        rel: entry.rel.clone(),
                        dir: false,
                    });
                }
            }
            UpperKind::Dir => {
                // A host-deleted directory removes its whole upper subtree
                // (including build artifacts inside it).
                if snapshot.contains_key(&entry.rel) && !current.contains_key(&entry.rel) {
                    ops.push(UpperOp::Remove {
                        rel: entry.rel.clone(),
                        dir: true,
                    });
                }
            }
            UpperKind::File => {
                let Some(staged) = snapshot.get(&entry.rel) else {
                    continue; // build artifact: keep
                };
                match current.get(&entry.rel) {
                    // Host deleted the file: the upper copy must not keep
                    // appearing in the merge.
                    None => ops.push(UpperOp::Remove {
                        rel: entry.rel.clone(),
                        dir: false,
                    }),
                    // Host modified the file: the upper copy would shadow
                    // the edit.
                    Some(now) if now != staged => ops.push(UpperOp::Remove {
                        rel: entry.rel.clone(),
                        dir: false,
                    }),
                    // Host unchanged: the upper copy carries the build's
                    // writes (e.g. quilt patches); keep it.
                    Some(_) => {}
                }
            }
        }
    }
    ops
}

/// Apply sync ops to the upperdir, skipping paths already covered by a
/// removed ancestor. Returns how many entries were actually removed.
pub(crate) fn apply_upper_sync(upper_root: &Path, ops: &[UpperOp]) -> io::Result<usize> {
    let mut removed_prefixes: Vec<String> = Vec::new();
    let mut removed = 0;
    for op in ops {
        let UpperOp::Remove { rel, dir } = op;
        if removed_prefixes
            .iter()
            .any(|prefix| rel.starts_with(prefix.as_str()))
        {
            continue;
        }
        let path = upper_root.join(rel.replace('/', std::path::MAIN_SEPARATOR_STR));
        let result = if *dir {
            fs::remove_dir_all(&path)
        } else {
            fs::remove_file(&path)
        };
        match result {
            Ok(()) => {
                removed += 1;
                if *dir {
                    let mut prefix = rel.clone();
                    if !prefix.ends_with('/') {
                        prefix.push('/');
                    }
                    removed_prefixes.push(prefix);
                }
            }
            Err(e) if e.kind() != io::ErrorKind::NotFound => {
                return Err(e);
            }
            Err(_) => {}
        }
    }
    Ok(removed)
}

/// Sync the persistent upperdir with the current host tree before the
/// forked record re-mounts it. Returns whether build-level resume is
/// possible (a usable snapshot existed and the upper was synced):
/// without a snapshot the upper is wiped and the build stages cold,
/// but safely.
pub(crate) fn sync_upper_for_resume(session: &Session, host_tree: &Path) -> io::Result<bool> {
    let Some(snapshot) = load_snapshot(&session.snapshot_file()) else {
        log::info!(
            "No host-tree snapshot in the record: build artifacts are \
             discarded, the environment is still reused"
        );
        session.wipe_build_artifacts();
        take_snapshot(host_tree, session)?;
        return Ok(false);
    };
    let current = walk_tree_meta(host_tree)?;
    let upper_entries = walk_upper(&session.upper_dir()).unwrap_or_default();
    let ops = plan_upper_sync(&snapshot, &current, &upper_entries);
    let removed = apply_upper_sync(&session.upper_dir(), &ops)?;
    if removed > 0 {
        log::debug!("Synced the record tree: {removed} stale upper entries dropped");
    }
    write_snapshot(&current, &session.snapshot_file())?;
    Ok(true)
}

/// Record fixtures shared by the `deb` test modules (this one and
/// `crate::deb`): a synthetic sessions root plus records carrying a
/// retained staged tree.
#[cfg(test)]
pub(crate) mod record_fixtures {
    use super::*;

    /// A writer copying everything it receives into a shared buffer:
    /// lets tests observe what a `Box<dyn Write + Send>` sink wrote.
    #[derive(Clone)]
    pub(crate) struct SharedWriter(pub Arc<parking_lot::Mutex<Vec<u8>>>);

    impl SharedWriter {
        /// A fresh empty capture buffer.
        pub(crate) fn new() -> Self {
            Self(Arc::new(parking_lot::Mutex::new(Vec::new())))
        }

        /// What has been written so far.
        pub(crate) fn contents(&self) -> String {
            String::from_utf8_lossy(&self.0.lock()).into_owned()
        }
    }

    impl std::io::Write for SharedWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().extend_from_slice(buf);
            Ok(buf.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    pub fn test_identity() -> SessionIdentity {
        SessionIdentity {
            package: "pkh".to_string(),
            series: "noble".to_string(),
            arch: "amd64".to_string(),
            cross: false,
        }
    }

    /// A synthetic sessions root and the chroot tarball its records were
    /// bootstrapped from.
    pub fn sessions_fixture() -> (tempfile::TempDir, PathBuf) {
        let root = tempfile::tempdir().unwrap();
        let tarball = root.path().join("noble-amd64-buildd.tar.xz");
        fs::write(&tarball, b"tarball-bytes").unwrap();
        (root, tarball)
    }

    /// An `--incremental`-shaped record: a retained staged tree (the
    /// upperdir, its host-tree snapshot and the tree journal), the sha
    /// of the tarball its environment was extracted from, and an
    /// outcome. The returned handle holds the record lock; drop it
    /// before forking or selecting from another handle.
    pub fn record_with_upper(
        root: &Path,
        tarball: &Path,
        host_tree: &Path,
        identity: &SessionIdentity,
        version: &str,
        outcome: Option<&str>,
    ) -> Session {
        let session = new_record_in(root, host_tree, identity, version).unwrap();
        fs::create_dir_all(session.upper_dir().join("src")).unwrap();
        fs::write(session.upper_dir().join("src/main.o"), b"objects").unwrap();
        fs::write(session.snapshot_file(), "f\t1\t10\ttree/src/a.c\n").unwrap();
        session.set_tarball_sha256(&sha256_file(tarball).unwrap());
        session.record_phase("patches", Some("stamp".to_string()));
        session.retain_upper(true, outcome == Some(OUTCOME_SUCCESS));
        if let Some(outcome) = outcome.filter(|o| *o != OUTCOME_SUCCESS) {
            session.set_outcome(outcome);
        }
        session
    }

    /// A metadata-only record (a plain run's shape): manifest and log,
    /// no staged tree.
    pub fn metadata_record(
        root: &Path,
        host_tree: &Path,
        identity: &SessionIdentity,
        version: &str,
        outcome: &str,
    ) -> Session {
        let session = new_record_in(root, host_tree, identity, version).unwrap();
        session.retain_upper(false, outcome == OUTCOME_SUCCESS);
        if outcome != OUTCOME_SUCCESS {
            session.set_outcome(outcome);
        }
        session
    }
}

// ---------------------------------------------------------------------------
// Removal (shared by `pkh deb` teardown and `pkh prune`)
// ---------------------------------------------------------------------------

/// Remove a record root: unmount every mount under it (the /proc bind
/// mount, overlay filesystems), then `rm -rf` it (privileged when needed:
/// the chroot holds root-owned device nodes).
pub(crate) fn remove_session_dir(root: &Path) -> io::Result<()> {
    unmount_under(root);
    crate::deb::ephemeral::privileged_remove(root)
}

#[cfg(test)]
mod tests {
    use super::*;
    use record_fixtures::{metadata_record, record_with_upper, sessions_fixture, test_identity};

    fn meta(mtime_ns: i128, size: u64) -> EntryMeta {
        EntryMeta { mtime_ns, size }
    }

    fn rel_map(entries: &[(&str, EntryMeta)]) -> BTreeMap<String, EntryMeta> {
        entries
            .iter()
            .map(|(rel, m)| (rel.to_string(), *m))
            .collect()
    }

    /// A build artifact (no host original in the snapshot) survives the
    /// sync; a host-deleted or host-modified source file does not.
    #[test]
    fn plan_keeps_artifacts_and_drops_stale_upper_files() {
        let staged = meta(1000, 10);
        let snapshot = rel_map(&[("src/a.c", staged)]);
        let current = rel_map(&[("src/b.c", staged)]); // a.c deleted, b.c added
        let upper = vec![
            UpperEntry {
                rel: "src/a.c".into(),
                kind: UpperKind::File,
            },
            UpperEntry {
                rel: "src/b.o".into(),
                kind: UpperKind::File,
            },
        ];
        assert_eq!(
            plan_upper_sync(&snapshot, &current, &upper),
            vec![UpperOp::Remove {
                rel: "src/a.c".into(),
                dir: false,
            }]
        );
    }

    /// A host-modified file must not stay shadowed by its upper copy.
    #[test]
    fn plan_drops_upper_copy_of_modified_host_file() {
        let snapshot = rel_map(&[("src/a.c", meta(1000, 10))]);
        let current = rel_map(&[("src/a.c", meta(2000, 12))]);
        let upper = vec![UpperEntry {
            rel: "src/a.c".into(),
            kind: UpperKind::File,
        }];
        assert_eq!(
            plan_upper_sync(&snapshot, &current, &upper),
            vec![UpperOp::Remove {
                rel: "src/a.c".into(),
                dir: false,
            }]
        );
    }

    /// An unchanged host file keeps its upper copy (the build's writes,
    /// e.g. quilt patches, stay applied across the resume).
    #[test]
    fn plan_keeps_upper_copy_of_unchanged_host_file() {
        let staged = meta(1000, 10);
        let snapshot = rel_map(&[("src/a.c", staged)]);
        let current = rel_map(&[("src/a.c", staged)]);
        let upper = vec![UpperEntry {
            rel: "src/a.c".into(),
            kind: UpperKind::File,
        }];
        assert!(plan_upper_sync(&snapshot, &current, &upper).is_empty());
    }

    /// Whiteouts survive only when they still hide an unchanged host file;
    /// a host re-creation (new metadata) drops them so the file shows.
    #[test]
    fn plan_keeps_accurate_whiteouts_only() {
        let staged = meta(1000, 10);
        let snapshot = rel_map(&[("src/a.c", staged)]);
        let current = rel_map(&[("src/a.c", staged)]);
        let upper = vec![UpperEntry {
            rel: "src/a.c".into(),
            kind: UpperKind::Whiteout,
        }];
        assert!(plan_upper_sync(&snapshot, &current, &upper).is_empty());

        let recreated = rel_map(&[("src/a.c", meta(5000, 99))]);
        assert_eq!(
            plan_upper_sync(&snapshot, &recreated, &upper),
            vec![UpperOp::Remove {
                rel: "src/a.c".into(),
                dir: false,
            }]
        );

        // A whiteout for a path that never existed at staging time (e.g.
        // the VCS-dir pruning) is not in the snapshot: it must be kept
        // only while the host still has the path — the .git case, where
        // current has no entry because the walk skips nothing but the
        // whiteout hides it; here the path is gone from the host, so the
        // whiteout is stale.
        let upper_git = vec![UpperEntry {
            rel: ".git/config".into(),
            kind: UpperKind::Whiteout,
        }];
        assert_eq!(
            plan_upper_sync(&snapshot, &current, &upper_git),
            vec![UpperOp::Remove {
                rel: ".git/config".into(),
                dir: false,
            }]
        );
    }

    /// A host-deleted directory removes its whole upper subtree in one op.
    #[test]
    fn plan_removes_subtree_of_deleted_host_directory() {
        let staged = meta(1000, 10);
        let snapshot = rel_map(&[("src/old", staged)]);
        let current = BTreeMap::new();
        let upper = vec![
            UpperEntry {
                rel: "src/old".into(),
                kind: UpperKind::Dir,
            },
            UpperEntry {
                rel: "src/old/x.o".into(),
                kind: UpperKind::File,
            },
        ];
        assert_eq!(
            plan_upper_sync(&snapshot, &current, &upper),
            vec![UpperOp::Remove {
                rel: "src/old".into(),
                dir: true,
            }]
        );
    }

    /// Applying ops removes the right entries; removals under an already
    /// removed ancestor are skipped, and missing paths are tolerated.
    #[test]
    fn apply_removes_entries_and_dedupes_ancestors() {
        let base = tempfile::tempdir().unwrap();
        let upper = base.path().join("upper");
        fs::create_dir_all(upper.join("src/old")).unwrap();
        fs::write(upper.join("src/a.c"), "x").unwrap();
        fs::write(upper.join("src/old/x.o"), "x").unwrap();
        fs::write(upper.join("gone.c"), "x").unwrap();

        let ops = vec![
            UpperOp::Remove {
                rel: "src/old".into(),
                dir: true,
            },
            UpperOp::Remove {
                rel: "src/old/x.o".into(),
                dir: false,
            },
            UpperOp::Remove {
                rel: "src/a.c".into(),
                dir: false,
            },
            UpperOp::Remove {
                rel: "gone.c".into(),
                dir: false,
            },
        ];
        let removed = apply_upper_sync(&upper, &ops).unwrap();
        // src/old/x.o is covered by its removed ancestor; all three others
        // (src/old, src/a.c, gone.c) are removed.
        assert_eq!(removed, 3);
        assert!(!upper.join("src/old").exists());
        assert!(!upper.join("src/a.c").exists());
        assert!(!upper.join("gone.c").exists());
    }

    /// The snapshot round-trips through its file format.
    #[test]
    fn snapshot_round_trips() {
        let base = tempfile::tempdir().unwrap();
        let path = base.path().join("host-files.list");
        let meta_map = rel_map(&[
            ("tree/src/a.c", meta(12345, 7)),
            ("orig.tar.xz", meta(1, 2)),
        ]);
        write_snapshot(&meta_map, &path).unwrap();
        assert_eq!(load_snapshot(&path), Some(meta_map));
    }

    /// The snapshot walk is best-effort below the root: an unreadable
    /// sibling directory in the staged parent (a root-owned tree the
    /// user's own setup script left) is skipped with a warning instead
    /// of aborting the build, and only the root's own unreadability
    /// propagates. Deterministic as non-root: mode bits are checked with
    /// no owner override, so a mode-000 directory blocks even its owner.
    #[test]
    fn walk_skips_unreadable_paths_but_fails_on_an_unreadable_root() {
        use std::os::unix::fs::PermissionsExt;

        let base = tempfile::tempdir().unwrap();
        let root = base.path().join("parent");
        let readable = root.join("linux/src");
        let secret = root.join("demos/sysroot-riscv64");
        fs::create_dir_all(&readable).unwrap();
        fs::create_dir_all(&secret).unwrap();
        fs::write(readable.join("a.c"), "src").unwrap();
        fs::write(secret.join("sysroot.tar"), "root-owned").unwrap();
        fs::set_permissions(&secret, fs::Permissions::from_mode(0o000)).unwrap();

        // The unreadable subtree is absent from the map; the walk itself
        // succeeds. The directory entry itself IS stat-able (its mode
        // hides its CONTENT, not its existence), so it appears as a
        // tracked empty subtree root only if it were a file — here it is
        // skipped whole, its contents untracked.
        let walked = walk_tree_meta(&root).expect("one unreadable sibling must not abort the walk");

        // Restore before asserting, so a failure still cleans up.
        fs::set_permissions(&secret, fs::Permissions::from_mode(0o755)).unwrap();

        assert!(
            walked.contains_key("linux/src/a.c"),
            "the readable tree is tracked: {walked:?}"
        );
        assert!(
            !walked.contains_key("demos/sysroot-riscv64/sysroot.tar"),
            "the unreadable subtree's content is not tracked: {walked:?}"
        );

        // An unreadable ROOT is a real error: the package tree itself is
        // invisible to this run.
        fs::set_permissions(&secret, fs::Permissions::from_mode(0o000)).unwrap();
        let err = walk_tree_meta(&secret).expect_err("walking an unreadable root must fail");
        fs::set_permissions(&secret, fs::Permissions::from_mode(0o755)).unwrap();
        assert_eq!(err.kind(), std::io::ErrorKind::PermissionDenied);
    }

    /// Identity slugs are sanitized and collision-free, and the manifest
    /// round-trips through JSON — lineage, tarball sha and the kept
    /// upper included, and their absence (older manifests) parsing as
    /// `None`/empty/false.
    #[test]
    fn slug_and_manifest_round_trip() {
        let identity = SessionIdentity {
            package: "linux/hacked".to_string(),
            series: "stonking".to_string(),
            arch: "arm64".to_string(),
            cross: true,
        };
        let slug = slug_for(&identity);
        assert!(
            slug.chars()
                .all(|c| c.is_ascii_alphanumeric() || "_-.".contains(c))
        );
        assert_eq!(slug_for(&identity), slug); // stable

        let manifest = SessionManifest {
            schema: SCHEMA_VERSION,
            pkh: "0.1.0".to_string(),
            id: "20260926T143505".to_string(),
            host_tree: "/home/me/linux".to_string(),
            identity: identity.clone(),
            created: now_rfc3339(),
            last_used: now_rfc3339(),
            last_outcome: OUTCOME_FAILED.to_string(),
            base: Some("20260926T140000".to_string()),
            tarball_sha256: "ab".to_string(),
            phases: BTreeMap::from([(
                "patches".to_string(),
                PhaseStamp {
                    at: now_rfc3339(),
                    stamp: None,
                },
            )]),
            tree_version: "7.3.0-5.6~local2".to_string(),
            kept_upper: true,
            build_resume: true,
            timings: Some(BuildTimings {
                deps_secs: 41,
                build_secs: 9020,
            }),
            profile: Some(BuildProfile {
                phases: vec![
                    PhaseTiming {
                        phase: "preparing_chroot".to_string(),
                        millis: 38_200,
                    },
                    PhaseTiming {
                        phase: "building".to_string(),
                        millis: 125_000,
                    },
                ],
            }),
        };
        let base = tempfile::tempdir().unwrap();
        save_manifest(base.path(), &manifest).unwrap();
        let loaded = open_from_dir(base.path().to_path_buf()).unwrap();
        assert_eq!(loaded.manifest(), manifest);

        // A manifest without the newer fields still parses, with their
        // defaults.
        let mut value = serde_json::to_value(&manifest).unwrap();
        let object = value.as_object_mut().unwrap();
        object.remove("base");
        object.remove("tarball_sha256");
        object.remove("kept_upper");
        object.remove("timings");
        object.remove("profile");
        let older: SessionManifest = serde_json::from_value(value).unwrap();
        assert_eq!(older.base, None);
        assert_eq!(older.tarball_sha256, "");
        assert!(!older.kept_upper);
        assert_eq!(older.timings, None);
        assert_eq!(older.profile, None);

        // A foreign schema is not discoverable.
        let mut foreign = manifest.clone();
        foreign.schema = SCHEMA_VERSION + 1;
        save_manifest(base.path(), &foreign).unwrap();
        assert!(open_from_dir(base.path().to_path_buf()).is_none());
    }

    /// `pkh deb list` renders a table with the fork lineage, the
    /// upper-kept state and the recorded build durations; the empty
    /// case explains itself.
    #[test]
    fn session_list_rendering() {
        let rows = vec![
            SessionListRow {
                id: "20260926T143505".to_string(),
                package: "linux".to_string(),
                version: "7.3.0-5.6~local2".to_string(),
                target: "stonking/arm64*".to_string(),
                base: Some("20260926T140000".to_string()),
                upper: true,
                outcome: OUTCOME_FAILED.to_string(),
                deps: "2m05s".to_string(),
                build: "2h30m".to_string(),
                age: "2h".to_string(),
            },
            SessionListRow {
                id: "20260925T090012".to_string(),
                package: "linux".to_string(),
                version: "7.3.0-5.5~local1".to_string(),
                target: "stonking/riscv64*".to_string(),
                base: None,
                upper: false,
                outcome: OUTCOME_SUCCESS.to_string(),
                deps: "-".to_string(),
                build: "-".to_string(),
                age: "1d".to_string(),
            },
        ];
        let rendered = render_session_list(Path::new("/home/me/linux"), &rows);
        assert!(
            rendered.contains("Records for /home/me/linux:"),
            "{rendered}"
        );
        assert!(rendered.contains("20260926T143505"), "{rendered}");
        assert!(rendered.contains("stonking/arm64*"), "{rendered}");
        // The lineage and upper columns: the fork's base, the metadata
        // record's dash.
        assert!(rendered.contains("20260926T140000"), "{rendered}");
        assert!(rendered.contains("UPPER"), "{rendered}");
        // The timing columns: both durations of the timed record, the
        // dashes of the record without any.
        assert!(rendered.contains("DEPS"), "{rendered}");
        assert!(rendered.contains("2m05s"), "{rendered}");
        assert!(rendered.contains("2h30m"), "{rendered}");
        assert!(
            rendered.contains("Fork with: pkh deb --incremental [<id>]"),
            "{rendered}"
        );

        let empty = render_session_list(Path::new("/home/me/linux"), &[]);
        assert!(empty.contains("No build records recorded"), "{empty}");
    }

    /// The recorded durations round-trip through the manifest and the
    /// list rows, and unmeasured builds record none: the accumulator
    /// sums both totals, an empty one is not recorded, and
    /// `pkh deb list` formats the seconds into stable units.
    #[test]
    fn timings_accumulate_render_and_persist() {
        let mut timings = BuildTimings::default();
        assert!(timings.is_empty());
        timings.add_deps(Duration::from_millis(1500));
        timings.add_deps(Duration::from_secs(3));
        timings.add_build(Duration::from_secs(65));
        assert_eq!(timings.deps_secs, 4);
        assert_eq!(timings.build_secs, 65);
        assert!(!timings.is_empty());

        let (root, _tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let record = new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap();
        assert_eq!(record.manifest().timings, None);

        record.set_timings(timings);
        let stored = record
            .manifest()
            .timings
            .expect("set_timings must record the durations");
        assert_eq!(stored, timings);
        let reloaded = open_from_dir(record.root().to_path_buf()).unwrap();
        assert_eq!(reloaded.manifest().timings, Some(timings));

        assert_eq!(format_duration(0), "0s");
        assert_eq!(format_duration(45), "45s");
        assert_eq!(format_duration(125), "2m05s");
        assert_eq!(format_duration(5400), "1h30m");
        assert_eq!(format_duration(277_200), "3d05h");
    }

    /// The per-phase profile round-trips through the manifest, and
    /// `pkh deb profile` renders each recorded phase under its human
    /// label with its share of the total, the total itself, and — for a
    /// failed build — the phase the error surfaced from; a record
    /// without a profile says so instead.
    #[test]
    fn profile_persists_and_renders() {
        let (root, _tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let record = metadata_record(root.path(), &host_tree, &identity, "1.0", OUTCOME_FAILED);
        assert_eq!(record.manifest().profile, None);
        // No profile recorded: the renderer explains rather than fails.
        assert!(
            render_record_profile(&record).contains("No phase profile recorded"),
            "{}",
            render_record_profile(&record)
        );

        let profile = BuildProfile {
            phases: vec![
                // An unknown key (an older pkh's phase since renamed)
                // renders as its raw key instead of disappearing.
                PhaseTiming {
                    phase: "retired_phase".to_string(),
                    millis: 500,
                },
                PhaseTiming {
                    phase: "preparing_chroot".to_string(),
                    millis: 10_000,
                },
                PhaseTiming {
                    phase: "updating_package_lists".to_string(),
                    millis: 5_000,
                },
                PhaseTiming {
                    phase: "installing_build_deps".to_string(),
                    millis: 75_000,
                },
                PhaseTiming {
                    phase: "building".to_string(),
                    millis: 10_000,
                },
            ],
        };
        record.set_profile(profile.clone());
        assert_eq!(record.manifest().profile, Some(profile.clone()));
        let reloaded = open_from_dir(record.root().to_path_buf()).unwrap();
        assert_eq!(reloaded.manifest().profile, Some(profile));

        let rendered = render_record_profile(&record);
        assert!(rendered.contains("Profile of build"), "{rendered}");
        assert!(rendered.contains("1.0"), "{rendered}");
        assert!(rendered.contains("failed"), "{rendered}");
        // Labels come from the known keys; the unknown key stays raw.
        assert!(rendered.contains("Preparing chroot"), "{rendered}");
        assert!(rendered.contains("Updating package lists"), "{rendered}");
        assert!(
            rendered.contains("Installing build dependencies"),
            "{rendered}"
        );
        assert!(rendered.contains("Building package"), "{rendered}");
        assert!(rendered.contains("retired_phase"), "{rendered}");
        // Shares of the 100_500 ms total, and the total itself.
        assert!(rendered.contains("9%"), "{rendered}");
        assert!(rendered.contains("74%"), "{rendered}");
        assert!(rendered.contains("500ms"), "{rendered}");
        assert!(rendered.contains("1m40s"), "{rendered}");
        // A failed build's profile names the phase it died in (the last
        // recorded one).
        assert!(
            rendered.contains("failed during \"Building package\""),
            "{rendered}"
        );

        assert_eq!(format_phase_millis(0), "0ms");
        assert_eq!(format_phase_millis(740), "740ms");
        assert_eq!(format_phase_millis(1000), "1s");
        assert_eq!(format_phase_millis(100_500), "1m40s");
    }

    // -------------------------------------------------------------------
    // Record layout, fork gate, retention, sweep and logs
    // -------------------------------------------------------------------

    /// A new record lives at `<slug>/<id>/` with its manifest and build
    /// log on disk and its lock beside the root; a concurrent handle sees
    /// it busy, and same-second creations get distinct ids.
    #[test]
    fn record_layout_and_locking() {
        let (root, _tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");

        let record = new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap();
        let slug_dir = identity_dir(root.path(), &identity);
        let record_dir = record.root();
        assert_eq!(record_dir.parent(), Some(slug_dir.as_path()));
        assert!(record_dir.join("manifest.json").is_file());
        assert!(record.log_file().is_file(), "the log exists from creation");
        assert_eq!(
            record.lock_path(),
            slug_dir.join(format!("{}.lock", record.manifest().id))
        );
        let manifest = record.manifest();
        assert_eq!(manifest.last_outcome, OUTCOME_RUNNING);
        assert_eq!(manifest.tree_version, "1.0");
        assert_eq!(manifest.base, None);
        assert_eq!(manifest.tarball_sha256, "");
        assert!(!manifest.kept_upper);

        // The handle holds the flock: another handle sees the record busy.
        assert!(lock_is_held(&record.lock_path()).unwrap());
        let other = open_from_dir(record_dir.to_path_buf()).unwrap();
        assert!(matches!(other.acquire(), Err(SessionBusy { .. })));
        drop(record);
        assert!(!lock_is_held(&other.lock_path()).unwrap());

        // Records created within the same second still get distinct ids
        // (the suffix disambiguates), and discovery finds them newest
        // first.
        drop(new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap());
        let ids: Vec<String> = identity_records_in(root.path(), &identity)
            .into_iter()
            .map(|r| r.manifest().id)
            .collect();
        assert_eq!(ids.len(), 2);
        assert_ne!(ids[0], ids[1]);
    }

    /// The ccache directory of an identity lives under the shared ccache
    /// root, keyed by the identity slug like the records.
    #[test]
    fn ccache_dir_is_keyed_by_the_identity_slug() {
        let identity = test_identity();
        assert_eq!(
            ccache_dir(&identity),
            PathBuf::from("/var/tmp/pkh/ccache").join(slug_for(&identity))
        );
    }

    /// The fork gate: the base's staged tree is carried only while its
    /// recorded tarball sha still matches the cached tarball AND the
    /// changelog version is unchanged. Everything else forks cold.
    #[test]
    fn fork_gate_carries_the_upper_only_on_sha_and_version_match() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let sha = sha256_file(&tarball).unwrap();
        let base = record_with_upper(
            root.path(),
            &tarball,
            &host_tree,
            &identity,
            "1.0",
            Some(OUTCOME_SUCCESS),
        );
        let base_id = base.manifest().id;

        // Everything matches: the staged tree, its snapshot and the tree
        // journal are carried.
        let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", &sha).unwrap();
        assert_eq!(fork.manifest().base.as_deref(), Some(base_id.as_str()));
        assert_eq!(
            fs::read_to_string(fork.upper_dir().join("src/main.o")).unwrap(),
            "objects"
        );
        assert!(fork.snapshot_file().is_file());
        assert!(fork.manifest().phases.contains_key("patches"));
        drop(fork);

        // The changelog version changed: cold fork.
        let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.1", &sha).unwrap();
        assert_eq!(fork.manifest().tree_version, "1.1");
        assert!(!fork.upper_dir().join("src/main.o").exists());
        assert!(!fork.snapshot_file().exists());
        assert!(fork.manifest().phases.is_empty());
        drop(fork);

        // The toolchain moved (the cached tarball changed): cold fork.
        let fork =
            fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", "deadbeef").unwrap();
        assert!(!fork.upper_dir().join("src/main.o").exists());
        drop(fork);

        // No cached tarball yet (the sha is unknown): cold fork.
        let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", "").unwrap();
        assert!(!fork.upper_dir().join("src/main.o").exists());
    }

    /// Forking never touches the base: writes into the fork's staged
    /// tree stay in the fork.
    #[test]
    fn fork_isolation() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let sha = sha256_file(&tarball).unwrap();
        let base = record_with_upper(root.path(), &tarball, &host_tree, &identity, "1.0", None);
        let base_upper = base.upper_dir().join("src/main.o");

        let fork = fork_record_in(root.path(), &base, &host_tree, &identity, "1.0", &sha).unwrap();
        fs::write(fork.upper_dir().join("src/main.o"), b"changed").unwrap();
        fs::write(fork.snapshot_file(), "mutated\n").unwrap();
        assert_eq!(fs::read_to_string(&base_upper).unwrap(), "objects");
        assert_eq!(
            fs::read_to_string(base.snapshot_file()).unwrap(),
            "f\t1\t10\ttree/src/a.c\n"
        );
    }

    /// The user scenario: record A succeeds, a fork B of it fails — the
    /// default selection must pick A (the newest success), not B (the
    /// newest record); with no success at all it takes the newest.
    #[test]
    fn selection_prefers_the_newest_success_over_the_newest_failure() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let sha = sha256_file(&tarball).unwrap();
        let a = record_with_upper(
            root.path(),
            &tarball,
            &host_tree,
            &identity,
            "1.0",
            Some(OUTCOME_SUCCESS),
        );
        let a_id = a.manifest().id;
        let b = fork_record_in(root.path(), &a, &host_tree, &identity, "1.0", &sha).unwrap();
        let b_id = b.manifest().id;
        b.set_outcome(OUTCOME_FAILED);
        drop((a, b));

        let candidates = fork_candidates_in(root.path(), &identity);
        assert_eq!(candidates.len(), 2);
        assert_eq!(candidates[0].manifest().id, b_id, "newest first");
        let ranked = rank_base_candidates(candidates);
        assert_eq!(ranked[0].manifest().id, a_id, "the success wins");
        assert_eq!(ranked[1].manifest().id, b_id);

        // No success anywhere (A fails too): the newest record is the
        // fallback.
        identity_records_in(root.path(), &identity)
            .into_iter()
            .find(|r| r.manifest().id == a_id)
            .expect("record A is still there")
            .set_outcome(OUTCOME_FAILED);
        let ranked = rank_base_candidates(fork_candidates_in(root.path(), &identity));
        assert_eq!(ranked[0].manifest().id, b_id);
    }

    /// Metadata-only records (plain runs, degraded ones) are not fork
    /// candidates; records are scoped per identity.
    #[test]
    fn metadata_only_records_are_not_fork_candidates() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let other = SessionIdentity {
            arch: "riscv64".to_string(),
            ..identity.clone()
        };
        let host_tree = root.path().join("tree");
        drop(metadata_record(
            root.path(),
            &host_tree,
            &identity,
            "1.0",
            OUTCOME_SUCCESS,
        ));
        drop(record_with_upper(
            root.path(),
            &tarball,
            &host_tree,
            &other,
            "1.0",
            None,
        ));
        assert!(
            fork_candidates_in(root.path(), &identity).is_empty(),
            "a metadata-only record carries nothing to fork"
        );
        assert_eq!(fork_candidates_in(root.path(), &other).len(), 1);
        assert_eq!(identity_records_in(root.path(), &identity).len(), 1);
        assert_eq!(load_all_in(root.path()).len(), 2);
    }

    /// Retaining a sixth staged tree degrades the oldest one to a
    /// metadata-only record (upperdir and snapshot deleted, manifest and
    /// log kept, `kept_upper` flipped): at most five staged trees per
    /// target at any time.
    #[test]
    fn retaining_a_sixth_upper_degrades_the_oldest() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let mut oldest = None;
        for i in 0..6 {
            let record = record_with_upper(
                root.path(),
                &tarball,
                &host_tree,
                &identity,
                &format!("1.{i}"),
                None,
            );
            oldest.get_or_insert_with(|| record.manifest().id.clone());
            drop(record);
        }
        let oldest = oldest.unwrap();
        let degraded = identity_records_in(root.path(), &identity)
            .into_iter()
            .find(|r| r.manifest().id == oldest)
            .expect("the oldest record is still there, metadata-only");
        assert!(!degraded.has_upper(), "the oldest staged tree is gone");
        assert!(!degraded.snapshot_file().exists());
        assert!(!degraded.manifest().kept_upper);
        assert!(degraded.root().join("manifest.json").is_file());
        assert!(degraded.log_file().is_file());
        assert_eq!(
            fork_candidates_in(root.path(), &identity).len(),
            KEEP_UPPERS,
            "exactly five staged trees remain"
        );
    }

    /// The end-of-run retention: metadata and the log are kept for every
    /// build; the staged tree exactly for `--incremental` ones.
    #[test]
    fn retain_upper_keeps_metadata_for_every_build() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");

        // A plain run, success or failure: metadata-only.
        for succeeded in [true, false] {
            let record =
                record_with_upper(root.path(), &tarball, &host_tree, &identity, "1.0", None);
            record.retain_upper(false, succeeded);
            assert!(record.root().join("manifest.json").is_file(), "metadata");
            assert!(record.log_file().is_file(), "the log is always kept");
            assert!(!record.has_upper(), "a plain run drops its staged tree");
            assert!(!record.snapshot_file().exists());
            let expected = if succeeded {
                OUTCOME_SUCCESS
            } else {
                OUTCOME_RUNNING // the failure outcome is the flow's to set
            };
            assert_eq!(record.manifest().last_outcome, expected);
            drop(record);
        }

        // An incremental run: the staged tree is kept, success or
        // failure.
        for outcome in [OUTCOME_SUCCESS, OUTCOME_FAILED] {
            let record = record_with_upper(
                root.path(),
                &tarball,
                &host_tree,
                &identity,
                "1.0",
                Some(outcome),
            );
            assert!(record.has_upper(), "an incremental run keeps its tree");
            assert!(record.manifest().kept_upper);
            assert!(record.root().join("manifest.json").is_file());
            assert!(record.log_file().is_file());
            drop(record);
        }
    }

    /// The retention decision follows the disk, not the request: an
    /// `--incremental` run that ended before staging created a staged
    /// tree has nothing to keep, and neither the manifest nor the
    /// end-of-run message may claim otherwise.
    #[test]
    fn upper_retention_reflects_the_disk_not_the_request() {
        use UpperRetention::{Drop, Keep, KeepNothing};

        assert_eq!(upper_retention(true, true), Keep);
        assert_eq!(upper_retention(true, false), KeepNothing);
        assert_eq!(upper_retention(false, true), Drop);
        assert_eq!(upper_retention(false, false), Drop);
    }

    /// Driving the keep path on an incremental record whose staged tree
    /// was never created: the record is kept as metadata + log, the
    /// manifest does not claim a staged tree, and the interrupt path
    /// records the same honest state.
    #[test]
    fn keep_without_a_staged_tree_claims_none() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");

        // A run that ended before staging: no tree-upper, no snapshot.
        let record = new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap();
        assert!(!record.has_upper());

        record.retain_upper(true, false);

        assert!(record.root().join("manifest.json").is_file(), "metadata");
        assert!(record.log_file().is_file(), "the log is always kept");
        assert!(
            !record.manifest().kept_upper,
            "no staged tree was created: the manifest must not claim one"
        );
        assert_eq!(record.manifest().last_outcome, OUTCOME_RUNNING);
        drop(record);

        // The interrupt path records the same honest state.
        let record = new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap();
        record.record_interrupted(true);
        assert_eq!(record.manifest().last_outcome, OUTCOME_FAILED);
        assert!(
            !record.manifest().kept_upper,
            "an interrupted pre-staging run keeps no claim of a staged tree"
        );
        drop(record);

        // And an interrupted run that DID stage claims exactly that.
        let record = record_with_upper(root.path(), &tarball, &host_tree, &identity, "1.0", None);
        record.record_interrupted(true);
        assert!(record.manifest().kept_upper);
        assert!(record.has_upper(), "the tree stays on disk");
    }

    /// The per-run chroot and overlay workdir are removed at the end of
    /// the run — nothing in a record root survives except the manifest,
    /// the log and the retained staged tree.
    #[test]
    fn run_state_is_removed_at_end_of_run() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let record = record_with_upper(root.path(), &tarball, &host_tree, &identity, "1.0", None);
        fs::create_dir_all(record.chroot_path().join("bin")).unwrap();
        fs::write(record.chroot_path().join("bin/sh"), b"").unwrap();
        fs::create_dir_all(record.work_dir().join("scratch")).unwrap();

        record.remove_run_state();

        assert!(!record.chroot_path().exists(), "the chroot is gone");
        assert!(!record.work_dir().exists(), "the overlay workdir is gone");
        assert!(record.has_upper(), "the staged tree survives");
        assert!(record.root().join("manifest.json").is_file());
    }

    /// A crashed run's leftover chroot is swept when the next run of the
    /// identity starts; a live build's chroot (lock held) is left alone.
    #[test]
    fn crashed_chroots_are_swept_on_the_next_run() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let crashed = record_with_upper(
            root.path(),
            &tarball,
            &host_tree,
            &identity,
            "1.0",
            Some(OUTCOME_FAILED),
        );
        fs::create_dir_all(crashed.chroot_path().join("bin")).unwrap();
        let live = record_with_upper(root.path(), &tarball, &host_tree, &identity, "1.0", None);
        fs::create_dir_all(live.chroot_path().join("bin")).unwrap();
        drop(crashed);

        // The next run of the identity sweeps the unlocked leftovers.
        drop(new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap());

        let swept = identity_records_in(root.path(), &identity)
            .into_iter()
            .find(|r| !lock_is_held(&r.lock_path()).unwrap())
            .expect("the crashed record is unlocked");
        assert!(!swept.chroot_path().exists(), "the leftover is swept");
        assert!(
            live.chroot_path().exists(),
            "the live build keeps its chroot"
        );
        drop(live);
    }

    /// `pkh deb log` resolves the newest record of a tree by default, an
    /// exact id or unique prefix when given, and streams its log; a
    /// terminal record's --follow just prints.
    #[test]
    fn record_log_resolution_and_streaming() {
        let (root, tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let first = metadata_record(root.path(), &host_tree, &identity, "1.0", OUTCOME_FAILED);
        fs::write(first.log_file(), "first build log\n").unwrap();
        let first_id = first.manifest().id;
        drop(first);
        // The next record must land in another second: same-second ids
        // are prefix-related (`<stamp>` / `<stamp>-02`), and the pinned
        // lookup below must resolve the first record's id unambiguously.
        std::thread::sleep(Duration::from_millis(1100));
        let newest = record_with_upper(
            root.path(),
            &tarball,
            &host_tree,
            &identity,
            "1.1",
            Some(OUTCOME_SUCCESS),
        );
        fs::write(newest.log_file(), "newest build log\n").unwrap();
        drop(newest);

        // No id: the newest record of the tree.
        let record = find_record(root.path(), &host_tree, None).unwrap();
        let mut out = Vec::new();
        print_record_log(&record, false, &mut out).unwrap();
        assert_eq!(String::from_utf8(out).unwrap(), "newest build log\n");

        // The exact id of the older record; the record is terminal, so
        // --follow just prints.
        let record = find_record(root.path(), &host_tree, Some(&first_id)).unwrap();
        let mut out = Vec::new();
        print_record_log(&record, true, &mut out).unwrap();
        assert_eq!(String::from_utf8(out).unwrap(), "first build log\n");

        // A wrong id is an error.
        assert!(find_record(root.path(), &host_tree, Some("nonesuch")).is_err());
    }

    /// A record whose build captured no line yet streams an empty log.
    #[test]
    fn record_log_of_a_silent_build_is_empty() {
        let (root, _tarball) = sessions_fixture();
        let identity = test_identity();
        let host_tree = root.path().join("tree");
        let record = new_record_in(root.path(), &host_tree, &identity, "1.0").unwrap();
        record.set_outcome(OUTCOME_SUCCESS);
        drop(record);
        let record = find_record(root.path(), &host_tree, None).unwrap();
        let mut out = Vec::new();
        print_record_log(&record, true, &mut out).unwrap();
        assert_eq!(out.len(), 0);
    }

    /// The raw writer sink routes each stream to its own writer,
    /// newline-terminated: this is what keeps --verbose (and piped
    /// runs) showing raw output under the record-log wrapper.
    #[test]
    fn raw_writer_sink_routes_streams_to_their_writers() {
        use record_fixtures::SharedWriter;

        let stdout = SharedWriter::new();
        let stderr = SharedWriter::new();
        let sink = RawWriterSink::new(Box::new(stdout.clone()), Box::new(stderr.clone()));

        sink.line(Stream::Stdout, "from stdout");
        sink.line(Stream::Stderr, "from stderr");
        sink.line(Stream::Stdout, "second line");

        assert_eq!(
            stdout.contents(),
            "from stdout\nsecond line\n",
            "stdout lines land in order, newline-terminated"
        );
        assert_eq!(
            stderr.contents(),
            "from stderr\n",
            "stderr lines land on their own writer only"
        );
    }

    /// Ages render in the coarse units the list promises.
    #[test]
    fn age_rendering() {
        let now = chrono::Utc::now();
        assert_eq!(
            age_of(&now.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)),
            "now"
        );
        let two_hours =
            (now - chrono::Duration::hours(2)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
        assert_eq!(age_of(&two_hours), "2h");
        let three_days =
            (now - chrono::Duration::days(3)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
        assert_eq!(age_of(&three_days), "3d");
        assert_eq!(age_of("not a date"), "unknown");
    }
}