composefs_ctl/

lib.rs

//! Library for `cfsctl` command line utility
//!
//! This crate also re-exports all composefs-rs library crates, so downstream
//! consumers can take a single dependency on `cfsctl` instead of listing each
//! crate individually.
//!
//! ```
//! use composefs_ctl::composefs::repository::Repository;
//! use composefs_ctl::composefs::fsverity::Sha256HashValue;
//!
//! let repo = Repository::<Sha256HashValue>::open_path(
//!     rustix::fs::CWD,
//!     "/nonexistent",
//! );
//! assert!(repo.is_err());
//! ```

pub use composefs;
pub use composefs_boot;
#[cfg(feature = "http")]
pub use composefs_http;
#[cfg(feature = "oci")]
pub use composefs_oci;

/// Shell completion helpers for dynamic value completion via [`clap_complete`].
pub mod complete;
pub mod composefs_info;
#[cfg(feature = "fuse")]
pub mod fuse;
pub mod mkcomposefs;
pub mod mountcomposefs;
/// Varlink RPC service exposing repository operations over a Unix socket.
pub mod varlink;

#[cfg(any(feature = "oci", feature = "http"))]
use std::collections::HashMap;
use std::io::{Read, Write};
use std::path::Path;
#[cfg(any(feature = "oci", feature = "http"))]
use std::sync::Mutex;
use std::{ffi::OsString, path::PathBuf};

#[cfg(feature = "oci")]
use std::{fs::create_dir_all, io::IsTerminal};

use std::sync::Arc;

use anyhow::{Context as _, Result};
use clap::{Parser, Subcommand, ValueEnum};
use clap_complete::engine::ArgValueCompleter;
use comfy_table::{Table, presets::UTF8_FULL};
#[cfg(feature = "ostree")]
use complete::complete_ostree_refs;
use complete::{complete_image_refs, complete_stream_refs};
#[cfg(feature = "oci")]
use complete::{complete_oci_digests, complete_oci_tags, complete_oci_tags_and_digests};
#[cfg(any(feature = "oci", feature = "http", feature = "ostree"))]
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use rustix::fs::{CWD, Mode, OFlags};

#[cfg(any(feature = "oci", feature = "http", feature = "ostree"))]
use composefs::progress::{
    ComponentId, ProgressEvent, ProgressReporter, ProgressUnit, SharedReporter,
};
use composefs_boot::BootOps;
use composefs_boot::cmdline::ComposefsCmdline;
#[cfg(feature = "oci")]
use composefs_boot::write_boot;

use composefs::erofs::format::FormatVersion;
#[cfg(feature = "oci")]
use composefs::shared_internals::IO_BUF_CAPACITY;
use composefs::{
    dumpfile::{dump_single_dir, dump_single_file},
    erofs::reader::erofs_to_filesystem,
    fsverity::{Algorithm, FsVerityHashValue, Sha256HashValue, Sha512HashValue},
    generic_tree::{FileSystem, Inode},
    mount::MountOptions,
    repository::{
        REPO_METADATA_FILENAME, Repository, RepositoryConfig, read_repo_algorithm, system_path,
        user_path,
    },
    tree::RegularFile,
};

/// An `indicatif`-backed [`ProgressReporter`] for use in the CLI.
///
/// Renders per-component progress bars via [`MultiProgress`].  When a component
/// completes or is skipped the bar is removed; human-readable messages are
/// printed above the bar group via [`MultiProgress::println`].
#[cfg(any(feature = "oci", feature = "http", feature = "ostree"))]
struct IndicatifReporter {
    multi: MultiProgress,
    bars: Mutex<HashMap<ComponentId, ProgressBar>>,
}

#[cfg(any(feature = "oci", feature = "http", feature = "ostree"))]
impl IndicatifReporter {
    fn new() -> Self {
        IndicatifReporter {
            multi: MultiProgress::new(),
            bars: Mutex::new(HashMap::new()),
        }
    }

    /// Build a shared reporter from this instance.
    fn into_shared(self) -> SharedReporter {
        Arc::new(self)
    }
}

#[cfg(any(feature = "oci", feature = "http", feature = "ostree"))]
impl std::fmt::Debug for IndicatifReporter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("IndicatifReporter").finish_non_exhaustive()
    }
}

#[cfg(any(feature = "oci", feature = "http", feature = "ostree"))]
impl ProgressReporter for IndicatifReporter {
    fn report(&self, event: ProgressEvent) {
        match event {
            ProgressEvent::Started { id, total, unit } => {
                let bar = if let Some(total) = total {
                    self.multi.add(ProgressBar::new(total))
                } else {
                    self.multi.add(ProgressBar::new_spinner())
                };
                let style = match unit {
                    ProgressUnit::Bytes => ProgressStyle::with_template(
                        "[eta {eta}] {bar:40.cyan/blue} {decimal_bytes:>7}/{decimal_total_bytes:7} {msg}",
                    ),
                    ProgressUnit::Items => ProgressStyle::with_template(
                        "[eta {eta}] {bar:40.cyan/blue} {pos:>7}/{len:7} objects {msg}",
                    ),
                    // Future unit variants fall back to a generic spinner.
                    _ => ProgressStyle::with_template(
                        "[eta {eta}] {bar:40.cyan/blue} {pos}/{len} {msg}",
                    ),
                };
                bar.set_style(
                    style
                        .unwrap_or_else(|_| ProgressStyle::default_bar())
                        .progress_chars("##-"),
                );
                bar.set_message(id.to_string());
                self.bars.lock().unwrap().insert(id, bar);
            }
            ProgressEvent::Progress { id, fetched, .. } => {
                if let Some(bar) = self.bars.lock().unwrap().get(&id) {
                    bar.set_position(fetched);
                }
            }
            ProgressEvent::Done { id, .. } => {
                if let Some(bar) = self.bars.lock().unwrap().remove(&id) {
                    bar.finish_and_clear();
                }
            }
            ProgressEvent::Skipped { id } => {
                if let Some(bar) = self.bars.lock().unwrap().remove(&id) {
                    bar.finish_with_message("skipped");
                }
            }
            ProgressEvent::Message(msg) => {
                let _ = self.multi.println(msg);
            }
            // `ProgressEvent` is #[non_exhaustive]: new variants added to the library
            // will be silently ignored here until cfsctl is updated to handle them.
            _ => {}
        }
    }
}

/// cfsctl
#[derive(Debug, Parser)]
#[clap(name = "cfsctl", version)]
pub struct App {
    /// Operate on repo at path
    #[clap(long, group = "repopath", value_hint = clap::ValueHint::DirPath)]
    repo: Option<PathBuf>,
    /// Operate on repo at standard user location $HOME/.var/lib/composefs
    #[clap(long, group = "repopath")]
    user: bool,
    /// Operate on repo at standard system location /sysroot/composefs
    #[clap(long, group = "repopath")]
    system: bool,

    /// What hash digest type to use for composefs repo.
    /// If omitted, auto-detected from repository metadata (meta.json).
    #[clap(long, value_enum)]
    pub hash: Option<HashType>,

    /// The EROFS format version to use when generating images.
    /// If omitted, the library default (V1) is used.
    #[clap(long, value_enum)]
    pub erofs_version: Option<ErofsVersion>,

    /// Deprecated: security mode is now auto-detected from meta.json.
    /// Use `cfsctl init --insecure` to create a repo without verity.
    /// Kept for backward compatibility.
    #[clap(long, hide = true)]
    insecure: bool,

    /// Error if the repository does not have fs-verity enabled.
    #[clap(long)]
    require_verity: bool,

    /// Don't automatically upgrade old-format repositories.
    /// When set, commands will fail on repos without meta.json instead
    /// of inferring metadata from existing objects.
    #[clap(long)]
    no_upgrade: bool,

    /// Don't open a repository. Only valid for commands that don't need one
    /// (compute-id, create-dumpfile).
    #[clap(long)]
    pub no_repo: bool,

    #[clap(subcommand)]
    cmd: Command,
}

/// The Hash algorithm used for FsVerity computation
#[derive(Debug, Copy, Clone, PartialEq, Eq, ValueEnum)]
pub enum HashType {
    /// Sha256
    Sha256,
    /// Sha512
    Sha512,
}

/// The EROFS format version used when generating images.
#[derive(Debug, Copy, Clone, PartialEq, Eq, ValueEnum)]
pub enum ErofsVersion {
    /// Format V0: compact inodes, BFS, C-compatible (composefs_version auto-detects 0 or 1).
    #[clap(name = "0")]
    V0,
    /// Format V1: same layout as V0, composefs_version always 1.
    #[clap(name = "1")]
    V1,
    /// Format V2: extended inodes, DFS (composefs_version=2).
    #[clap(name = "2")]
    V2,
}

impl From<ErofsVersion> for composefs::erofs::format::FormatVersion {
    fn from(v: ErofsVersion) -> Self {
        match v {
            ErofsVersion::V0 => Self::V0,
            ErofsVersion::V1 => Self::V1,
            ErofsVersion::V2 => Self::V2,
        }
    }
}

/// A reference to an OCI image: either a content digest or a named ref.
///
/// Digests are prefixed with `@` (e.g. `@sha256:abc123…`), while bare
/// names are refs resolved through the repository's ref tree. The `@`
/// prefix is necessary to disambiguate because ref names may contain `:`
/// — OCI digest algorithms are intentionally extensible, so we cannot
/// rely on parse heuristics to distinguish the two.
///
/// Note this differs from the podman/docker convention where `@` appears
/// between the image name and the digest (e.g. `fedora@sha256:abc…`).
/// Here, `@` is always a leading prefix on the entire argument.
///
/// At the repository level, ref names are freeform strings (the only
/// restriction is that they must not start with `@`). In practice,
/// `oci pull` defaults to tagging with the source transport reference
/// (e.g. `docker://quay.io/fedora/fedora:latest`), so most refs in a
/// repository will be container transport names — which naturally never
/// start with `@`.
#[cfg(feature = "oci")]
#[derive(Debug, Clone)]
pub enum OciReference {
    /// A content-addressable digest such as `sha256:abcdef…`.
    Digest(composefs_oci::OciDigest),
    /// A named ref resolved through the repository's ref tree, typically
    /// a container transport name (e.g. `docker://quay.io/foo:latest`).
    Named(String),
}

#[cfg(feature = "oci")]
impl std::str::FromStr for OciReference {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        if let Some(digest_str) = s.strip_prefix('@') {
            let digest: composefs_oci::OciDigest =
                digest_str.parse().context("Invalid OCI digest after '@'")?;
            Ok(Self::Digest(digest))
        } else {
            Ok(Self::Named(s.to_owned()))
        }
    }
}

#[cfg(feature = "oci")]
impl std::fmt::Display for OciReference {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Digest(d) => write!(f, "@{d}"),
            Self::Named(n) => write!(f, "{n}"),
        }
    }
}

/// CLI representation of [`composefs_oci::LocalFetchOpt`].
#[cfg(feature = "oci")]
#[derive(Debug, Clone, Copy, Default, clap::ValueEnum)]
enum LocalFetchCli {
    /// Do not use native containers-storage import; use skopeo.
    #[default]
    Disabled,
    /// Use native import with reflink/hardlink/copy fallback.
    Auto,
    /// Use native import; error if zero-copy is not possible.
    Zerocopy,
}

#[cfg(feature = "oci")]
impl From<LocalFetchCli> for composefs_oci::LocalFetchOpt {
    fn from(cli: LocalFetchCli) -> Self {
        match cli {
            LocalFetchCli::Disabled => Self::Disabled,
            LocalFetchCli::Auto => Self::IfPossible,
            LocalFetchCli::Zerocopy => Self::ZeroCopy,
        }
    }
}

/// Common options for operations using OCI config manifest streams that may transform the image rootfs
#[cfg(feature = "oci")]
#[derive(Debug, Parser)]
struct OCIConfigFilesystemOptions {
    #[clap(flatten)]
    base_config: OCIConfigOptions,
    /// Whether bootable transformation should be performed on the image rootfs
    #[clap(long)]
    bootable: bool,
}

/// Common options for operations using OCI config manifest streams
#[cfg(feature = "oci")]
#[derive(Debug, Parser)]
struct OCIConfigOptions {
    /// Ref name (e.g. myimage:latest) or @digest (e.g. @sha256:a1b2c3...)
    #[arg(add = ArgValueCompleter::new(complete_oci_tags_and_digests))]
    config_name: OciReference,
    /// verity digest for the manifest stream to be verified against
    config_verity: Option<String>,
}

#[cfg(feature = "oci")]
#[derive(Debug, Subcommand)]
enum OciCommand {
    /// Import a tar layer as a splitstream in the repository
    ImportLayer {
        /// Layer content digest, e.g. sha256:a1b2c3...
        digest: composefs_oci::OciDigest,
        /// Optional human-readable name for the layer
        name: Option<String>,
    },
    /// Dump the rootfs of a stored OCI image as a composefs dumpfile to stdout
    ///
    /// The image can be specified by ref name or @digest:
    ///   cfsctl oci dump myimage:latest
    ///   cfsctl oci dump @sha256:a1b2c3...
    Dump {
        #[clap(flatten)]
        config_opts: OCIConfigFilesystemOptions,
    },
    /// Pull an OCI image into the repository
    ///
    /// Prints the config stream digest and verity of the stored manifest.
    Pull {
        /// Source image reference, as accepted by skopeo
        image: String,
        /// Tag name to assign to the pulled image (defaults to the image reference)
        name: Option<String>,
        /// Also generate a bootable EROFS image from the pulled OCI image
        #[arg(long)]
        bootable: bool,
        /// Controls whether containers-storage: references use the native
        /// import path with zero-copy reflink/hardlink support.
        #[arg(long, value_enum, default_value_t = LocalFetchCli::Disabled)]
        local_fetch: LocalFetchCli,
    },
    /// Copy an OCI image (and its layers) from another composefs repository
    /// into this repository.
    ///
    /// The destination repository is selected by the global `--repo`/`--user`/
    /// `--system` flags. The source is `--from`.
    ///
    /// Pass `--zerocopy` to attempt reflink (then hardlink) instead of copying
    /// object data.  This requires both repositories to be on the same
    /// filesystem, to use the same hash algorithm, and the caller to have
    /// `CAP_DAC_READ_SEARCH` (i.e. root).
    /// Without `--zerocopy`, objects are always copied, which is safe on any
    /// filesystem and across repositories using different hash algorithms.
    Copy {
        /// Image to copy (tag name or `@digest`).
        image: OciReference,
        /// Path to the source composefs repository.
        #[clap(long)]
        from: PathBuf,
        /// Tag to assign to the image in the destination repository.
        #[clap(long)]
        name: Option<String>,
        /// Use reflink/hardlink zero-copy transfer (requires same filesystem, same hash algorithm, and root).
        #[clap(long)]
        zerocopy: bool,
    },
    /// List all tagged OCI images in the repository
    #[clap(name = "images")]
    ListImages {
        /// Output as JSON array
        #[clap(long)]
        json: bool,
    },
    /// Show information about an OCI image
    ///
    /// The image can be specified by ref name or @digest:
    ///   cfsctl oci inspect myimage:latest
    ///   cfsctl oci inspect @sha256:a1b2c3...
    ///
    /// By default, outputs JSON with manifest, config, and referrers.
    /// Use --manifest or --config to output just that raw JSON.
    #[clap(name = "inspect")]
    Inspect {
        /// Ref name (e.g. myimage:latest) or @digest (e.g. @sha256:a1b2c3...)
        #[arg(add = ArgValueCompleter::new(complete_oci_tags_and_digests))]
        image: OciReference,
        /// Output only the raw manifest JSON (as originally stored)
        #[clap(long, conflicts_with = "config")]
        manifest: bool,
        /// Output only the raw config JSON (as originally stored)
        #[clap(long, conflicts_with = "manifest")]
        config: bool,
    },
    /// Tag an image with a new name
    ///
    /// Example: cfsctl oci tag sha256:a1b2c3... myimage:latest
    Tag {
        /// Manifest digest, e.g. sha256:a1b2c3...
        #[arg(add = ArgValueCompleter::new(complete_oci_digests))]
        manifest_digest: composefs_oci::OciDigest,
        /// Tag name to assign (must not contain '@')
        name: String,
    },
    /// Remove a tag from an image
    Untag {
        /// Tag name to remove
        #[arg(add = ArgValueCompleter::new(complete_oci_tags))]
        name: String,
    },
    /// Inspect a stored layer
    ///
    /// By default, outputs the raw tar stream to stdout.
    /// Use --dumpfile for composefs dumpfile format, or --json for metadata.
    #[clap(name = "layer")]
    LayerInspect {
        /// Layer diff_id, e.g. sha256:a1b2c3...
        layer: composefs_oci::OciDigest,
        /// Output as composefs dumpfile format (one entry per line)
        #[clap(long, conflicts_with = "json")]
        dumpfile: bool,
        /// Output layer metadata as JSON
        #[clap(long, conflicts_with = "dumpfile")]
        json: bool,
    },
    /// Mount an OCI image's composefs EROFS at the given mountpoint
    Mount {
        /// Image reference (tag name or manifest digest)
        #[arg(add = ArgValueCompleter::new(complete_oci_tags_and_digests))]
        image: String,
        /// Target mountpoint
        #[arg(value_hint = clap::ValueHint::AnyPath)]
        mountpoint: String,
        /// Mount the bootable variant instead of the regular EROFS image
        #[arg(long)]
        bootable: bool,
        #[clap(flatten)]
        mount_opts: MountOpts,
    },
    /// Compute the composefs image ID of a stored OCI image's rootfs
    ///
    /// The image can be specified by ref name or @digest:
    ///   cfsctl oci compute-id myimage:latest
    ///   cfsctl oci compute-id @sha256:a1b2c3...
    ComputeId {
        #[clap(flatten)]
        config_opts: OCIConfigFilesystemOptions,
    },

    /// Create the composefs image of the rootfs of a stored OCI image, perform bootable transformation, commit it to the repo,
    /// then configure boot for the image by writing new boot resources and bootloader entries to boot partition. Performs
    /// state preparation for composefs-setup-root consumption as well. Note that state preparation here is not suitable for
    /// consumption by bootc.
    PrepareBoot {
        #[clap(flatten)]
        config_opts: OCIConfigOptions,
        /// boot partition mount point
        #[clap(long, default_value = "/boot", value_hint = clap::ValueHint::DirPath)]
        bootdir: PathBuf,
        /// Boot entry identifier to use. By default uses ID provided by the image or kernel version
        #[clap(long)]
        entry_id: Option<String>,
        /// additional kernel command line
        #[clap(long)]
        cmdline: Vec<String>,
    },
    /// Check integrity of OCI images in the repository
    ///
    /// Verifies manifest and config content digests, layer references, seal
    /// consistency, and delegates to the underlying repository fsck for object
    /// integrity and splitstream validation.
    Fsck {
        /// Check only the named image instead of all tagged images
        #[arg(add = ArgValueCompleter::new(complete_oci_tags))]
        image: Option<String>,
        /// Output results as JSON (always exits 0 unless the check itself fails)
        #[clap(long)]
        json: bool,
    },
    /// Serve the varlink RPC API on a Unix socket or systemd socket.
    ///
    /// Equivalent to `cfsctl varlink`: a single service answers both the
    /// `org.composefs.Repository` and `org.composefs.Oci` interfaces on one
    /// socket. Kept for discoverability under the `oci` subcommand.
    Varlink {
        /// Unix socket path to listen on (omit when using systemd socket activation).
        #[clap(long, value_hint = clap::ValueHint::AnyPath)]
        address: Option<PathBuf>,
    },
}

#[cfg(feature = "ostree")]
#[derive(Debug, Subcommand)]
enum OstreeCommand {
    PullLocal {
        #[arg(value_hint = clap::ValueHint::DirPath)]
        ostree_repo_path: PathBuf,
        /// Ostree ref name or commit ID (64-character hex)
        ostree_ref: String,
        #[clap(long)]
        base_name: Option<String>,
    },
    Pull {
        #[arg(value_hint = clap::ValueHint::Url)]
        ostree_repo_url: String,
        /// Ostree ref name or commit ID (64-character hex)
        ostree_ref: String,
        #[clap(long)]
        base_name: Option<String>,
        /// Disable static delta usage, forcing object-by-object fetching
        #[clap(long)]
        no_delta: bool,
    },
    /// Mount an ostree commit's composefs EROFS at the given mountpoint
    Mount {
        /// Ostree commit ref or commit ID
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        commit: String,
        /// Target mountpoint
        #[arg(value_hint = clap::ValueHint::AnyPath)]
        mountpoint: String,
        #[clap(flatten)]
        mount_opts: MountOpts,
    },
    /// Dump the filesystem of an ostree commit as a composefs dumpfile to stdout
    Dump {
        /// Ostree commit ref name
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        commit_name: String,
    },
    /// Compute the composefs image ID of an ostree commit
    ComputeId {
        /// Ostree commit ref name
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        commit_name: String,
    },
    /// Show the contents of an ostree commit
    Inspect {
        /// Ostree ref name, commit ID, or commit ID prefix
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        source: String,
        /// Print only the commit metadata key-value pairs
        #[clap(long)]
        metadata: bool,
    },
    /// Tag an ostree commit with a name
    ///
    /// The source can be an ostree commit checksum or an existing ref name.
    Tag {
        /// Ostree commit checksum (hex) or existing ref name
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        source: String,
        /// Tag name to assign
        name: String,
    },
    /// Remove a named ostree reference
    Untag {
        /// Tag name to remove
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        name: String,
    },
    /// Create an ostree commit from a composefs image in the repository
    ///
    /// The image is specified by its object ID or refs/ name (the same
    /// format used by `cfsctl mount` and `cfsctl image-objects`).
    Commit {
        /// Composefs image ID or refs/ name
        #[arg(add = ArgValueCompleter::new(complete_image_refs))]
        image: String,
        /// Ostree ref name to tag the commit with
        #[clap(long)]
        reference: Option<String>,
        /// One-line commit subject
        #[clap(long, default_value = "")]
        subject: String,
    },
    /// Export an ostree commit to a local ostree repository
    ///
    /// Writes all objects (files, dirtrees, dirmetas, commit) to the
    /// destination repo. File content is reflinked when possible.
    /// Only bare, bare-user, and bare-user-only repos are supported.
    Export {
        /// Ostree ref name or commit ID to export
        #[arg(add = ArgValueCompleter::new(complete_ostree_refs))]
        source: String,
        /// Path to the destination ostree repository
        #[arg(value_hint = clap::ValueHint::DirPath)]
        ostree_repo_path: PathBuf,
        /// Ref name to set in the destination repo
        #[clap(long)]
        reference: Option<String>,
    },
    /// List all ostree commits in the repository
    #[clap(name = "images")]
    ListCommits,
    /// Apply a static delta to the repository
    ApplyDelta {
        /// Path to the delta file (single-file) or superblock
        #[arg(value_hint = clap::ValueHint::FilePath)]
        delta_path: PathBuf,
    },
    /// List refs available in a remote ostree repository
    ListRefs {
        /// URL of the remote ostree repository
        #[arg(value_hint = clap::ValueHint::Url)]
        ostree_repo_url: String,
        /// Summary index subset key (defaults to system architecture)
        #[clap(long)]
        subset: Option<String>,
    },
}

/// Common options for reading a filesystem from a path
#[derive(Debug, Parser)]
struct FsReadOptions {
    /// The path to the filesystem
    #[arg(value_hint = clap::ValueHint::DirPath)]
    path: PathBuf,
    /// Transform the filesystem for boot (SELinux labels, empty /boot and /sysroot)
    #[clap(long)]
    bootable: bool,
    /// Don't copy /usr metadata to root directory (use if root already has well-defined metadata)
    #[clap(long)]
    no_propagate_usr_to_root: bool,
}

/// Common options for mount commands (shared across regular, OCI, and ostree mount).
#[derive(Debug, Parser)]
struct MountOpts {
    /// Mount mode: auto, yes (force FUSE), or no (force kernel)
    #[cfg(feature = "fuse")]
    #[arg(long, value_enum, default_value_t)]
    fuse: FuseMode,
    /// Run FUSE server in the foreground (don't daemonize)
    #[cfg(feature = "fuse")]
    #[arg(long)]
    foreground: bool,
    /// Writable upper layer directory for overlayfs
    #[arg(long, requires = "workdir", value_hint = clap::ValueHint::DirPath)]
    upperdir: Option<PathBuf>,
    /// Work directory for overlayfs (required with --upperdir)
    #[arg(long, requires = "upperdir", value_hint = clap::ValueHint::DirPath)]
    workdir: Option<PathBuf>,
    /// Mount read-write (requires --upperdir)
    #[arg(long, requires = "upperdir")]
    read_write: bool,
}

impl MountOpts {
    fn to_mount_options(&self) -> Result<composefs::mount::MountOptions> {
        get_mount_options(
            self.upperdir.as_deref(),
            self.workdir.as_deref(),
            self.read_write,
        )
    }

    fn mount_image<ObjectID: FsVerityHashValue>(
        &self,
        repo: &Arc<Repository<ObjectID>>,
        image_name: &str,
        mountpoint: &str,
    ) -> Result<()> {
        let mount_options = self.to_mount_options()?;

        #[cfg(feature = "fuse")]
        if let mode @ (MountMode::Fuse | MountMode::FuseOverlay) =
            detect_mount_mode(self.fuse, self.upperdir.is_some())
        {
            return run_fuse_mount(
                repo,
                image_name,
                mountpoint,
                mode,
                mount_options,
                self.foreground,
            );
        }

        repo.mount_at(image_name, mountpoint, &mount_options)?;
        Ok(())
    }
}

#[derive(Debug, Subcommand)]
enum Command {
    /// Initialize a new composefs repository with a metadata file.
    ///
    /// Creates the repository directory (if it doesn't exist) and writes
    /// a `meta.json` recording the digest algorithm.  By default fs-verity
    /// is enabled on `meta.json`, signaling that all objects require
    /// verity.  Use `--insecure` to skip (e.g. on tmpfs).
    Init {
        /// The fs-verity algorithm identifier.
        /// Format: fsverity-<hash>-<lg_blocksize>, e.g. fsverity-sha512-12
        #[clap(long, value_parser = clap::value_parser!(Algorithm), default_value = "fsverity-sha512-12")]
        algorithm: Algorithm,
        /// Path to the repository directory (created if it doesn't exist).
        /// If omitted, uses --repo/--user/--system location.
        #[arg(value_hint = clap::ValueHint::DirPath)]
        path: Option<PathBuf>,
        /// Do not enable fs-verity on meta.json (insecure repository).
        #[clap(long)]
        insecure: bool,
        /// Migrate an old-format repository: remove streams/ and images/
        /// (which encode the algorithm) but keep objects/, then write
        /// fresh meta.json.  Streams and images will need to be
        /// re-imported after migration.
        #[clap(long)]
        reset_metadata: bool,
        /// Default EROFS format version for images in this repository.
        /// V1 is compatible with C `mkcomposefs` 1.0.8; V2 is the legacy composefs-rs format.
        /// If omitted, falls back to the global `--erofs-version` flag, then defaults to V1.
        #[clap(long)]
        erofs_version: Option<ErofsVersion>,
    },
    /// Take a transaction lock on the repository.
    /// This prevents garbage collection from occurring.
    Transaction,
    /// Reconstitutes a split stream and writes it to stdout
    Cat {
        /// the name of the stream to cat, either a content identifier or prefixed with 'ref/'
        #[arg(add = ArgValueCompleter::new(complete_stream_refs))]
        name: String,
    },
    /// Perform garbage collection
    GC {
        /// Additional roots to keep (image or stream names)
        #[clap(long, short = 'r')]
        root: Vec<String>,
        /// Preview what would be deleted without actually deleting
        #[clap(long, short = 'n')]
        dry_run: bool,
    },
    /// Imports a composefs image (unsafe!)
    ImportImage { reference: String },
    /// List all named image references in the repository
    #[clap(name = "images", alias = "list-images")]
    Images {
        /// Output as JSON array
        #[clap(long)]
        json: bool,
        /// Show full digest instead of truncated form
        #[clap(long)]
        no_trunc: bool,
    },
    /// Commands for dealing with OCI images and layers
    #[cfg(feature = "oci")]
    Oci {
        #[clap(subcommand)]
        cmd: OciCommand,
    },
    #[cfg(feature = "ostree")]
    Ostree {
        #[clap(subcommand)]
        cmd: OstreeCommand,
    },
    /// Mounts a composefs image, possibly enforcing fsverity of the image
    Mount {
        /// the name of the image to mount, either an fs-verity hash or prefixed with 'ref/'
        #[arg(add = ArgValueCompleter::new(complete_image_refs))]
        name: String,
        /// the mountpoint
        #[arg(value_hint = clap::ValueHint::AnyPath)]
        mountpoint: String,
        #[clap(flatten)]
        mount_opts: MountOpts,
    },
    /// Read rootfs located at a path, add all files to the repo, then create the composefs image of the rootfs,
    /// commit it to the repo, and print its image object ID
    CreateImage {
        #[clap(flatten)]
        fs_opts: FsReadOptions,
        /// optional reference name for the image, use as 'ref/<name>' elsewhere
        image_name: Option<String>,
    },
    /// Read rootfs located at a path and compute the composefs image object id of the rootfs.
    /// Note that this does not create or commit the composefs image itself, and does not
    /// store any file objects in the repository.
    ComputeId {
        #[clap(flatten)]
        fs_opts: FsReadOptions,
    },
    /// Read rootfs located at a path and compute the composefs kernel argument string.
    ///
    /// Like compute-id but outputs the full kernel argument rather than the bare digest,
    /// choosing the argument name based on the EROFS format version:
    ///
    ///   V1: composefs.digest=v1-sha256-12:<hex>
    ///   V2: composefs=<hex>
    ///
    /// Use --erofs-version to select the format.
    /// The boot transformation (SELinux relabeling, empty /boot and /sysroot) is
    /// always applied — this command produces a karg for a sealed boot image.
    ///
    /// Example (in a Containerfile):
    ///   cfsctl --erofs-version 1 compute-karg /mnt/base > /etc/kernel/cmdline
    #[clap(name = "compute-karg")]
    ComputeKarg {
        /// The path to the filesystem
        #[arg(value_hint = clap::ValueHint::DirPath)]
        path: PathBuf,
        /// Don't copy /usr metadata to root directory (use if root already has well-defined metadata)
        #[clap(long)]
        no_propagate_usr_to_root: bool,
    },
    /// Read rootfs located at a path and dump full content of the rootfs to a composefs dumpfile,
    /// writing to stdout. Does not store any file objects in the repository.
    CreateDumpfile {
        #[clap(flatten)]
        fs_opts: FsReadOptions,
    },
    /// Lists all object IDs referenced by an image
    ImageObjects {
        /// the name of the image to read, either an object ID digest or prefixed with 'ref/'
        #[arg(add = ArgValueCompleter::new(complete_image_refs))]
        name: String,
    },
    /// Extract file information from a composefs image for specified files or directories
    ///
    /// By default, outputs information in composefs dumpfile format
    DumpFiles {
        /// The name of the composefs image to read from, either an object ID digest or prefixed with 'ref/'
        #[arg(add = ArgValueCompleter::new(complete_image_refs))]
        image_name: String,
        /// File or directory paths to process. If a path is a directory, its contents will be listed.
        #[arg(value_hint = clap::ValueHint::AnyPath)]
        files: Vec<PathBuf>,
        /// Show backing path information instead of dumpfile format
        /// For each file, prints either "inline" for files stored within the image,
        /// or a path relative to the object store for files stored extrenally
        #[clap(long)]
        backing_path_only: bool,
    },
    /// Check repository integrity
    ///
    /// Verifies fsverity digests of all objects, validates stream and image
    /// symlinks, and checks splitstream internal consistency. Exits with
    /// a non-zero status if corruption is found.
    Fsck {
        /// Output results as JSON (always exits 0 unless the check itself fails)
        #[clap(long)]
        json: bool,
        /// Skip per-object fs-verity verification; check only metadata and
        /// symlink structure (much faster on large repositories)
        #[clap(long)]
        metadata_only: bool,
    },
    #[cfg(feature = "http")]
    Fetch {
        #[arg(value_hint = clap::ValueHint::Url)]
        url: String,
        name: String,
    },
    /// Serve the varlink RPC API on a Unix socket or systemd socket.
    ///
    /// A single service answers both the `org.composefs.Repository` and (when
    /// the `oci` feature is enabled) `org.composefs.Oci` interfaces on one
    /// socket.
    Varlink {
        /// Unix socket path to listen on (omit when using systemd socket activation).
        #[clap(long, value_hint = clap::ValueHint::AnyPath)]
        address: Option<PathBuf>,
    },

    /// Run mkcomposefs (C-compatible image builder); hidden, also available via argv0 dispatch.
    #[clap(hide = true, name = "mkcomposefs")]
    Mkcomposefs {
        /// Arguments forwarded verbatim to mkcomposefs
        #[clap(trailing_var_arg = true, allow_hyphen_values = true)]
        args: Vec<std::ffi::OsString>,
    },

    /// Run composefs-info (C-compatible image inspector); hidden, also available via argv0 dispatch.
    #[clap(hide = true, name = "composefs-info")]
    ComposefsInfo {
        /// Arguments forwarded verbatim to composefs-info
        #[clap(trailing_var_arg = true, allow_hyphen_values = true)]
        args: Vec<std::ffi::OsString>,
    },
}

/// Acts as a proxy for the `cfsctl` CLI by executing the CLI logic programmatically
///
/// This function behaves the same as invoking the `cfsctl` binary from the
/// command line. It accepts an iterator of CLI-style arguments (excluding
/// the binary name), parses them using `clap`
pub async fn run_from_iter<I>(args: I) -> Result<()>
where
    I: IntoIterator,
    I::Item: Into<OsString> + Clone,
{
    let args = App::parse_from(
        std::iter::once(OsString::from("cfsctl")).chain(args.into_iter().map(Into::into)),
    );

    run_app(args).await
}

#[cfg(feature = "ostree")]
fn print_pull_stats(stats: &composefs_ostree::PullStats) {
    if stats.delta_parts_applied > 0 {
        println!(
            "objects {} metadata + {} files via {} delta parts",
            stats.metadata_fetched, stats.files_fetched, stats.delta_parts_applied
        );
    } else {
        println!(
            "objects {} metadata + {} files fetched",
            stats.metadata_fetched, stats.files_fetched
        );
    }
}

fn get_mount_options(
    upperdir: Option<&Path>,
    workdir: Option<&Path>,
    read_write: bool,
) -> Result<MountOptions> {
    let mut options = MountOptions::default();
    if let (Some(u), Some(w)) = (upperdir, workdir) {
        let upper_fd = rustix::fs::open(
            u,
            OFlags::PATH | OFlags::DIRECTORY | OFlags::CLOEXEC,
            Mode::empty(),
        )
        .with_context(|| format!("Opening upperdir '{}'", u.display()))?;
        let work_fd = rustix::fs::open(
            w,
            OFlags::PATH | OFlags::DIRECTORY | OFlags::CLOEXEC,
            Mode::empty(),
        )
        .with_context(|| format!("Opening workdir '{}'", w.display()))?;
        options.set_overlay(upper_fd, work_fd);
    }
    options.set_read_write(read_write);
    Ok(options)
}

#[cfg(feature = "fuse")]
use fuse::{FuseMode, MountMode, detect_mount_mode, run_fuse_mount};

#[cfg(feature = "oci")]
pub(crate) fn verity_opt<ObjectID>(opt: &Option<String>) -> Result<Option<ObjectID>>
where
    ObjectID: FsVerityHashValue,
{
    Ok(match opt {
        Some(value) => Some(FsVerityHashValue::from_hex(value)?),
        None => None,
    })
}

/// Resolve the default repository path based on the effective uid.
///
/// Root operates on the system repository; everyone else on their per-user
/// repository. Used both when no `--repo`/`--user`/`--system` is given and by
/// the socket-activated path (which has no CLI args to consult).
pub(crate) fn default_repo_path() -> Result<PathBuf> {
    if rustix::process::getuid().is_root() {
        Ok(system_path())
    } else {
        user_path()
    }
}

/// Resolve the repository path from CLI args without opening it.
///
/// Uses [`user_path`] and [`system_path`] to avoid duplicating
/// path constants.
pub(crate) fn resolve_repo_path(args: &App) -> Result<PathBuf> {
    if let Some(path) = &args.repo {
        Ok(path.clone())
    } else if args.system {
        Ok(system_path())
    } else if args.user {
        user_path()
    } else {
        default_repo_path()
    }
}

/// Determine the effective hash type for a repository.
///
/// Resolution order:
/// 1. If `meta.json` exists, use its algorithm. Error if `--hash` was
///    explicitly passed and conflicts.
/// 2. If no metadata and `upgrade` is true, infer from existing objects.
/// 3. If no metadata and `upgrade` is false, error.
///
/// Note: we read the metadata file directly here (rather than via
/// `Repository::metadata`) because this runs *before* we know which
/// generic `ObjectID` type to use — that's exactly what we're deciding.
pub(crate) fn resolve_hash_type(
    repo_path: &Path,
    cli_hash: Option<HashType>,
    upgrade: bool,
) -> Result<HashType> {
    let repo_fd = rustix::fs::open(
        repo_path,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC,
        Mode::empty(),
    )
    .with_context(|| format!("opening repository {}", repo_path.display()))?;

    let algorithm = match read_repo_algorithm(&repo_fd)? {
        Some(alg) => alg,
        None if upgrade => {
            // No meta.json — try to infer from objects (old-format repo).
            // open_upgrade will write meta.json later when the repo is opened.
            composefs::repository::infer_repo_algorithm(&repo_fd).with_context(|| {
                format!(
                    "no {REPO_METADATA_FILENAME} in {}; tried to infer algorithm from objects",
                    repo_path.display(),
                )
            })?
        }
        None => {
            anyhow::bail!(
                "{REPO_METADATA_FILENAME} not found in {}; \
                 this repository must be initialized with `cfsctl init`",
                repo_path.display(),
            );
        }
    };

    let detected = match algorithm {
        Algorithm::Sha256 { .. } => HashType::Sha256,
        Algorithm::Sha512 { .. } => HashType::Sha512,
    };

    // If the user explicitly passed --hash and it doesn't match, error
    if let Some(explicit) = cli_hash
        && explicit != detected
    {
        anyhow::bail!(
            "repository is configured for {algorithm} (from {REPO_METADATA_FILENAME}) \
             but --hash {} was specified",
            match explicit {
                HashType::Sha256 => "sha256",
                HashType::Sha512 => "sha512",
            },
        );
    }

    Ok(detected)
}

/// If the process was started *bare* via systemd socket activation, serve the
/// varlink API on the activated socket and return `Ok(true)`. Otherwise return
/// `Ok(false)` so the caller falls through to normal CLI parsing.
///
/// This runs *before* clap to support a truly argument-less invocation —
/// notably `varlinkctl exec:cfsctl`, which hands us the connected socket on fd
/// 3 but passes no subcommand for clap to parse. A client selects a repository
/// at runtime via the `OpenRepository` method.
///
/// The shortcut is taken *only* when there are no command-line arguments
/// (`argv` is just the program name). When any argument is present — e.g. a
/// systemd unit running `cfsctl varlink` — we fall through to clap; the
/// `varlink`/`oci varlink` subcommand's [`serve`](crate::varlink::serve)
/// detects and serves on the activation fd itself. We must NOT call
/// [`try_activated_listener`](crate::varlink::try_activated_listener) on that
/// path: it consumes `LISTEN_FDS`/`LISTEN_PID` (via `receive_descriptors`),
/// which would prevent `serve` from finding the fd later.
pub async fn run_if_socket_activated() -> Result<bool> {
    // Only take the pre-clap shortcut for a bare invocation (`argv[0]` only).
    // Check argv before touching the activation env so the latter is consumed
    // only when we actually intend to serve from this shortcut.
    if std::env::args_os().len() != 1 {
        return Ok(false);
    }
    let service = crate::varlink::CfsctlService::activated();
    match crate::varlink::try_activated_listener()? {
        Some(crate::varlink::ActivatedSocket::Connected(l)) => {
            crate::varlink::serve_activated(service, l).await?;
            Ok(true)
        }
        Some(crate::varlink::ActivatedSocket::Listening(listener)) => {
            crate::varlink::serve_on_listener(service, listener).await?;
            Ok(true)
        }
        None => Ok(false),
    }
}

/// Top-level dispatch: handle init specially, otherwise open repo and run.
pub async fn run_app(args: App) -> Result<()> {
    // Hidden compat subcommands: forward all trailing args to the respective tool.
    if let Command::Mkcomposefs { args: extra } = args.cmd {
        return mkcomposefs::run_from_args(extra);
    }
    if let Command::ComposefsInfo { args: extra } = args.cmd {
        return composefs_info::run_from_args(extra);
    }

    // Init is handled before opening a repo since it creates one
    if let Command::Init {
        ref algorithm,
        ref path,
        insecure,
        reset_metadata,
        erofs_version: ref init_erofs_version,
    } = args.cmd
    {
        // Prefer the subcommand-level --erofs-version; fall back to global flag; default V1.
        let erofs_version = init_erofs_version
            .or(args.erofs_version)
            .map(composefs::erofs::format::FormatVersion::from)
            .unwrap_or(composefs::erofs::format::FormatVersion::V1);
        return run_init(
            algorithm,
            path.as_deref(),
            insecure || args.insecure,
            reset_metadata,
            erofs_version,
            &args,
        );
    }

    // The varlink service opens repositories on demand via `OpenRepository`
    // (handling both hash types), so it bypasses the generic repo-open dispatch
    // below. A single `CfsctlService` answers both the `org.composefs.Repository`
    // and (when the `oci` feature is enabled) `org.composefs.Oci` interfaces, so
    // `cfsctl varlink` and `cfsctl oci varlink` serve the same combined service.
    if let Command::Varlink { ref address } = args.cmd {
        let service = crate::varlink::CfsctlService::from_app(&args);
        return crate::varlink::serve(service, address.as_deref()).await;
    }

    #[cfg(feature = "oci")]
    if let Command::Oci {
        cmd: OciCommand::Varlink { ref address },
    } = args.cmd
    {
        let service = crate::varlink::CfsctlService::from_app(&args);
        return crate::varlink::serve(service, address.as_deref()).await;
    }

    // Commands that only need verity digests (no object storage) can
    // run without opening a repository.
    if args.no_repo
        || matches!(
            args.cmd,
            Command::ComputeId { .. }
                | Command::ComputeKarg { .. }
                | Command::CreateDumpfile { .. }
        )
    {
        // If a repo path is available and --no-repo wasn't passed,
        // try to read the hash type from the repo's metadata so that
        // e.g. `cfsctl --repo <sha256-repo> compute-id` uses SHA-256
        // instead of the default SHA-512.
        let effective_hash = if !args.no_repo {
            if let Ok(repo_path) = resolve_repo_path(&args) {
                resolve_hash_type(&repo_path, args.hash, !args.no_upgrade)
                    .unwrap_or(args.hash.unwrap_or(HashType::Sha512))
            } else {
                args.hash.unwrap_or(HashType::Sha512)
            }
        } else {
            args.hash.unwrap_or(HashType::Sha512)
        };
        return match effective_hash {
            HashType::Sha256 => run_cmd_without_repo::<Sha256HashValue>(args).await,
            HashType::Sha512 => run_cmd_without_repo::<Sha512HashValue>(args).await,
        };
    }

    let repo_path = resolve_repo_path(&args)?;
    let effective_hash = resolve_hash_type(&repo_path, args.hash, !args.no_upgrade)?;

    match effective_hash {
        HashType::Sha256 => run_cmd_with_repo(open_repo::<Sha256HashValue>(&args)?, args).await,
        HashType::Sha512 => run_cmd_with_repo(open_repo::<Sha512HashValue>(&args)?, args).await,
    }
}

/// Handle `cfsctl init`
fn run_init(
    algorithm: &Algorithm,
    path: Option<&Path>,
    insecure: bool,
    reset_metadata: bool,
    erofs_version: composefs::erofs::format::FormatVersion,
    args: &App,
) -> Result<()> {
    let repo_path = if let Some(p) = path {
        p.to_path_buf()
    } else {
        resolve_repo_path(args)?
    };

    if reset_metadata {
        composefs::repository::reset_metadata(&repo_path)?;
    }

    // Ensure parent directories exist (init_path only creates the final dir).
    if let Some(parent) = repo_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("creating parent directories for {}", repo_path.display()))?;
    }

    // init_path handles idempotency: same algorithm is a no-op,
    // different algorithm is an error.
    let config = {
        let mut c = RepositoryConfig::new(*algorithm);
        c.erofs_formats = composefs::erofs::format::FormatConfig::single(erofs_version);
        if insecure { c.set_insecure() } else { c }
    };
    let created = match algorithm {
        Algorithm::Sha256 { .. } => {
            Repository::<Sha256HashValue>::init_path(CWD, &repo_path, config)?.1
        }
        Algorithm::Sha512 { .. } => {
            Repository::<Sha512HashValue>::init_path(CWD, &repo_path, config)?.1
        }
    };

    if created {
        println!(
            "Initialized composefs repository at {}",
            repo_path.display()
        );
        println!("  algorithm: {algorithm}");
        if insecure {
            println!("  verity:    not required (insecure)");
        } else {
            println!("  verity:    required");
        }
    } else {
        println!("Repository already initialized at {}", repo_path.display());
    }

    Ok(())
}

/// Open a repo at an explicit path, auto-upgrading old-format repos unless
/// `no_upgrade` is set.
///
/// This is the parameterized core shared by [`open_repo`] (which derives the
/// path and flags from [`App`]) and the varlink service (which holds these
/// values directly).
pub(crate) fn open_repo_at<ObjectID>(
    path: &Path,
    insecure: bool,
    require_verity: bool,
    no_upgrade: bool,
) -> Result<Repository<ObjectID>>
where
    ObjectID: FsVerityHashValue,
{
    let mut repo = if no_upgrade {
        Repository::open_path(CWD, path)?
    } else {
        let (repo, _upgraded) = Repository::open_upgrade(CWD, path)?;
        repo
    };
    // Hidden --insecure flag for backward compatibility; the default
    // now is to inherit the repo config, but if it's specified we
    // disable requiring verity even if the repo says to use it.
    if insecure {
        repo.set_insecure();
    }
    if require_verity {
        repo.require_verity()?;
    }
    Ok(repo)
}

/// Open a repo, auto-upgrading old-format repos unless `--no-upgrade` was passed.
pub fn open_repo<ObjectID>(args: &App) -> Result<Repository<ObjectID>>
where
    ObjectID: FsVerityHashValue,
{
    let path = resolve_repo_path(args)?;
    let mut repo = open_repo_at(&path, args.insecure, args.require_verity, args.no_upgrade)?;
    // If the user explicitly passed --erofs-version, override the stored
    // repo setting for this invocation only (does not rewrite meta.json).
    if let Some(version) = args.erofs_version {
        repo.set_erofs_version(version.into());
    }
    Ok(repo)
}

/// Copy an OCI image (and all its layers) from one repository to another using varlink connections.
#[cfg(feature = "oci")]
pub async fn copy_image(
    conn_src: &mut zlink::unix::Connection,
    conn_dest: &mut zlink::unix::Connection,
    handle_src: u64,
    handle_dest: u64,
    image: &OciReference,
    name: Option<&str>,
    zerocopy: bool,
) -> Result<crate::varlink::layer_sync::FinalizeImageReply> {
    use crate::varlink::layer_sync::LayerRef;
    use crate::varlink::oci::OciError;
    use crate::varlink::proxy::{GetLayerParams, OciProxy};
    use anyhow::ensure;
    use zlink::futures_util::StreamExt as _;

    let image_str = image.to_string();
    let inspect = conn_src
        .inspect(handle_src, &image_str)
        .await
        .context("zlink transport error calling Inspect")?
        .map_err(|e: OciError| anyhow::anyhow!("Inspect failed: {e:?}"))?;

    ensure!(
        !inspect.manifest.is_empty(),
        "inspect returned empty manifest"
    );
    ensure!(!inspect.config.is_empty(), "inspect returned empty config");

    // Extract ordered layer identifiers via the shared helper that handles
    // both container images (rootfs.diff_ids) and OCI artifacts (manifest
    // layer digests).
    let diff_ids_ordered = composefs_oci::extract_layer_ids(&inspect.manifest, &inspect.config)
        .context("extracting layer identifiers")?;

    let mut layer_refs: Vec<LayerRef> = Vec::with_capacity(diff_ids_ordered.len());

    for diff_id in &diff_ids_ordered {
        let has = conn_dest
            .has_layer(handle_dest, diff_id)
            .await
            .context("zlink transport error calling HasLayer")?
            .map_err(|e: OciError| anyhow::anyhow!("HasLayer failed: {e:?}"))?;

        let layer_verity = if has.present {
            has.layer_verity
                .context("HasLayer returned present=true but no layer_verity")?
        } else {
            let get_params = GetLayerParams {
                diff_id: Some(diff_id.to_string()),
                storage: None,
            };
            let mut get_stream = std::pin::pin!(
                conn_src
                    .get_layer(handle_src, get_params)
                    .await
                    .context("zlink transport error calling GetLayer")?
            );
            let mut all_fds: Vec<std::os::fd::OwnedFd> = Vec::new();
            let mut get_reply = None;
            while let Some(item) = get_stream.next().await {
                let (result, fds) = item.context("GetLayer stream frame error")?;
                let reply =
                    result.map_err(|e: OciError| anyhow::anyhow!("GetLayer failed: {e:?}"))?;
                get_reply = Some(reply);
                all_fds.extend(fds);
            }
            let get_reply = get_reply.context("GetLayer returned empty stream")?;
            let dir_count = get_reply.dir_count as usize;

            let pipe_and_dirfds_len = 1 + dir_count;
            let lifetime_fds = all_fds.split_off(pipe_and_dirfds_len);

            let put_reply = conn_dest
                .put_layer(handle_dest, diff_id, zerocopy, all_fds)
                .await
                .context("zlink transport error calling PutLayer")?
                .map_err(|e: OciError| anyhow::anyhow!("PutLayer failed: {e:?}"))?;
            drop(lifetime_fds);

            put_reply.layer_verity
        };

        layer_refs.push(LayerRef {
            diff_id: diff_id.clone(),
            layer_verity,
        });
    }

    let finalize = conn_dest
        .finalize_image(
            handle_dest,
            &inspect.manifest,
            &inspect.config,
            layer_refs,
            name,
        )
        .await
        .context("zlink transport error calling FinalizeImage")?
        .map_err(|e: OciError| anyhow::anyhow!("FinalizeImage failed: {e:?}"))?;

    Ok(finalize)
}

/// Resolve an [`OciReference`] to an [`OciImage`].
#[cfg(feature = "oci")]
pub(crate) fn resolve_oci_image<ObjectID: FsVerityHashValue>(
    repo: &Repository<ObjectID>,
    reference: &OciReference,
) -> Result<composefs_oci::oci_image::OciImage<ObjectID>> {
    match reference {
        OciReference::Digest(digest) => {
            composefs_oci::oci_image::OciImage::open(repo, digest, None)
        }
        OciReference::Named(name) => composefs_oci::oci_image::OciImage::open_ref(repo, name),
    }
}

/// Resolve an [`OciReference`] to a config digest and optional verity.
///
/// When resolving via a named ref, the verity override is ignored since
/// the image metadata provides the correct verity.
#[cfg(feature = "oci")]
pub(crate) fn resolve_oci_config<ObjectID: FsVerityHashValue>(
    repo: &Repository<ObjectID>,
    reference: &OciReference,
    verity_override: Option<ObjectID>,
) -> Result<(composefs_oci::OciDigest, Option<ObjectID>)> {
    match reference {
        OciReference::Digest(digest) => Ok((digest.clone(), verity_override)),
        OciReference::Named(_) => {
            let img = resolve_oci_image(repo, reference)?;
            Ok((
                img.config_digest().clone(),
                Some(img.config_verity().clone()),
            ))
        }
    }
}

#[cfg(feature = "oci")]
fn load_filesystem_from_oci_image<ObjectID: FsVerityHashValue>(
    repo: &Repository<ObjectID>,
    opts: OCIConfigFilesystemOptions,
) -> Result<FileSystem<RegularFile<ObjectID>>> {
    let verity = verity_opt(&opts.base_config.config_verity)?;
    let (config_digest, config_verity) =
        resolve_oci_config(repo, &opts.base_config.config_name, verity)?;
    let mut fs =
        composefs_oci::image::create_filesystem(repo, &config_digest, config_verity.as_ref())?;
    if opts.bootable {
        fs.transform_for_boot(repo)?;
    }
    Ok(fs)
}

async fn load_filesystem_from_ondisk_fs<ObjectID: FsVerityHashValue>(
    fs_opts: &FsReadOptions,
    repo: Option<Arc<Repository<ObjectID>>>,
) -> Result<FileSystem<RegularFile<ObjectID>>> {
    // The async API needs an OwnedFd; fs_opts.path is typically absolute
    // so the dirfd is unused for path resolution, but required by the API.
    let dirfd = rustix::fs::openat(
        CWD,
        ".",
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC,
        Mode::empty(),
    )?;
    let mut fs = if fs_opts.no_propagate_usr_to_root {
        composefs::fs::read_filesystem(dirfd, fs_opts.path.clone(), repo.clone()).await?
    } else {
        composefs::fs::read_container_root(dirfd, fs_opts.path.clone(), repo.clone()).await?
    };
    if fs_opts.bootable {
        if let Some(repo) = &repo {
            fs.transform_for_boot(repo)?;
        } else {
            let rootfd = rustix::fs::openat(
                CWD,
                &fs_opts.path,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC,
                Mode::empty(),
            )?;
            fs.transform_for_boot_from_dir(rootfd)?;
        }
    }
    Ok(fs)
}

/// Print file information from a composefs filesystem for the given paths
///
/// For each path in `files`, looks up the entry in the filesystem and either
/// outputs composefs dumpfile-format metadata or, when `backing_path_only` is
/// set, prints whether the file is stored inline or its object-relative path.
/// Directory paths have their contents listed.
pub fn dump_files<ObjectID: FsVerityHashValue>(
    repo: &Repository<ObjectID>,
    image_name: &str,
    files: &Vec<PathBuf>,
    backing_path_only: bool,
) -> Result<Vec<u8>> {
    let (img_fd, _) = repo.open_image(image_name)?;

    let mut img_buf = Vec::new();
    std::fs::File::from(img_fd).read_to_end(&mut img_buf)?;

    let fs = erofs_to_filesystem::<ObjectID>(&img_buf)?;

    let mut out = Vec::new();
    let nlink_map = fs.nlinks();

    for file_path in files {
        let (dir, file) = fs.root.split(file_path.as_os_str())?;

        let (_, file) = dir
            .entries()
            .find(|ent| ent.0 == file)
            .ok_or_else(|| anyhow::anyhow!("{} not found", file_path.display()))?;

        match &file {
            Inode::Directory(directory) => {
                if backing_path_only {
                    anyhow::bail!("{} is a directory", file_path.display());
                }

                dump_single_dir(&mut out, directory, &fs, &nlink_map, file_path.clone())?
            }

            Inode::Leaf(leaf_id, _) => {
                use composefs::generic_tree::LeafContent::*;
                use composefs::tree::RegularFile::*;

                if backing_path_only {
                    let leaf = fs.leaf(*leaf_id);
                    match &leaf.content {
                        Regular(f) => match f {
                            Inline(..) | Sparse(..) => {
                                writeln!(&mut out, "{} inline", file_path.display())?;
                            }
                            External(id, _) | ExternalNoVerity(id, _) => {
                                writeln!(
                                    &mut out,
                                    "{} {}",
                                    file_path.display(),
                                    id.to_object_pathname()
                                )?;
                            }
                        },
                        _ => {
                            writeln!(&mut out, "{} inline", file_path.display())?;
                        }
                    }

                    continue;
                }

                dump_single_file(&mut out, *leaf_id, &fs, &nlink_map, file_path.clone())?
            }
        };
    }

    Ok(out)
}

/// Run commands that don't require a repository.
pub async fn run_cmd_without_repo<ObjectID: FsVerityHashValue>(args: App) -> Result<()> {
    let erofs_version = args
        .erofs_version
        .map(composefs::erofs::format::FormatVersion::from);
    match args.cmd {
        Command::ComputeId { fs_opts } => {
            let fs = load_filesystem_from_ondisk_fs::<ObjectID>(&fs_opts, None).await?;
            let version = erofs_version.unwrap_or_default();
            let id = composefs::fsverity::compute_verity::<ObjectID>(
                &composefs::erofs::writer::mkfs_erofs_versioned(
                    &composefs::erofs::writer::ValidatedFileSystem::new(fs)?,
                    version,
                ),
            );
            println!("{}", id.to_hex());
        }
        Command::ComputeKarg {
            path,
            no_propagate_usr_to_root,
        } => {
            let fs_opts = FsReadOptions {
                path,
                bootable: true,
                no_propagate_usr_to_root,
            };
            let fs = load_filesystem_from_ondisk_fs::<ObjectID>(&fs_opts, None).await?;
            let version = erofs_version.unwrap_or_default();
            let id = composefs::fsverity::compute_verity::<ObjectID>(
                &composefs::erofs::writer::mkfs_erofs_versioned(
                    &composefs::erofs::writer::ValidatedFileSystem::new(fs)?,
                    version,
                ),
            );
            let karg = match version {
                FormatVersion::V0 | FormatVersion::V1 => {
                    ComposefsCmdline::new_v1(id, args.insecure)
                }
                FormatVersion::V2 => ComposefsCmdline::new_v2(id, args.insecure),
            };
            println!("{}", karg.to_cmdline_arg());
        }
        Command::CreateDumpfile { fs_opts } => {
            let fs = load_filesystem_from_ondisk_fs::<ObjectID>(&fs_opts, None).await?;
            fs.print_dumpfile()?;
        }
        _ => {
            anyhow::bail!("--no-repo is only supported for compute-id and create-dumpfile");
        }
    }
    Ok(())
}

/// Run with cmd
pub async fn run_cmd_with_repo<ObjectID>(repo: Repository<ObjectID>, args: App) -> Result<()>
where
    ObjectID: FsVerityHashValue,
{
    let repo = Arc::new(repo);
    #[cfg(feature = "oci")]
    let dest_path = resolve_repo_path(&args)?;
    match args.cmd {
        Command::Init { .. } => {
            // Handled in run_app before we get here
            unreachable!("init is handled before opening a repository");
        }
        Command::Transaction => {
            // just wait for ^C
            loop {
                std::thread::park();
            }
        }
        Command::Cat { name } => {
            repo.merge_splitstream(&name, None, None, &mut std::io::stdout())?;
        }
        Command::ImportImage { reference } => {
            let image_id = repo.import_image(&reference, &mut std::io::stdin())?;
            println!("{}", image_id.to_id());
        }
        #[cfg(feature = "oci")]
        Command::Oci { cmd: oci_cmd } => match oci_cmd {
            OciCommand::ImportLayer { name, ref digest } => {
                let (object_id, _stats) = composefs_oci::import_layer(
                    &repo,
                    digest,
                    name.as_deref(),
                    tokio::io::BufReader::with_capacity(IO_BUF_CAPACITY, tokio::io::stdin()),
                )
                .await?;
                println!("{}", object_id.to_id());
            }
            OciCommand::Dump { config_opts } => {
                let fs = load_filesystem_from_oci_image(&repo, config_opts)?;
                fs.print_dumpfile()?;
            }
            OciCommand::Mount {
                ref image,
                ref mountpoint,
                bootable,
                ref mount_opts,
            } => {
                let img = if image.starts_with("sha256:") {
                    let digest: composefs_oci::OciDigest =
                        image.parse().context("Parsing manifest digest")?;
                    composefs_oci::oci_image::OciImage::open(&repo, &digest, None)?
                } else {
                    composefs_oci::oci_image::OciImage::open_ref(&repo, image)?
                };
                let erofs_id = if bootable {
                    match img.boot_image_ref(repo.erofs_version()) {
                        Some(id) => id,
                        None => anyhow::bail!(
                            "No boot EROFS image linked — try pulling with --bootable"
                        ),
                    }
                } else {
                    match img.image_ref(repo.erofs_version()) {
                        Some(id) => id,
                        None => anyhow::bail!(
                            "No composefs EROFS image linked — try re-pulling the image"
                        ),
                    }
                };
                mount_opts.mount_image(&repo, &erofs_id.to_hex(), mountpoint.as_str())?;
            }
            OciCommand::ComputeId { config_opts } => {
                let fs = load_filesystem_from_oci_image(&repo, config_opts)?;
                let id = fs.compute_image_id(repo.erofs_version());
                println!("{}", id.to_hex());
            }
            OciCommand::Pull {
                ref image,
                name,
                bootable,
                local_fetch,
            } => {
                // If no explicit name provided, use the image reference as the tag
                let tag_name = name.as_deref().unwrap_or(image);

                let reporter: SharedReporter = IndicatifReporter::new().into_shared();
                let opts = composefs_oci::PullOptions {
                    local_fetch: local_fetch.into(),
                    progress: Some(reporter),
                    ..Default::default()
                };

                let result = composefs_oci::pull(&repo, image, Some(tag_name), opts).await?;

                println!("manifest {}", result.manifest_digest);
                println!("config   {}", result.config_digest);
                println!("verity   {}", result.manifest_verity.to_hex());
                println!("tagged   {tag_name}");
                println!("objects  {}", result.stats);

                if bootable {
                    let image_verity =
                        composefs_oci::generate_boot_image(&repo, &result.manifest_digest)?;
                    println!("Boot image: {}", image_verity.to_hex());
                }
            }
            OciCommand::Copy {
                ref image,
                ref from,
                ref name,
                zerocopy,
            } => {
                use crate::varlink::proxy::RepositoryProxy;

                let src_hash = resolve_hash_type(from, args.hash, !args.no_upgrade)
                    .with_context(|| format!("opening source repository {}", from.display()))?;
                let dest_hash = resolve_hash_type(&dest_path, args.hash, !args.no_upgrade)
                    .with_context(|| {
                        format!("opening destination repository {}", dest_path.display())
                    })?;

                if zerocopy && src_hash != dest_hash {
                    anyhow::bail!(
                        "--zerocopy requires matching hash algorithms; \
                         source uses {src_hash:?} but destination uses {dest_hash:?}"
                    );
                }

                let from_str = from.to_str().context("source path is not valid UTF-8")?;
                let dest_str = dest_path
                    .to_str()
                    .context("destination path is not valid UTF-8")?;

                let service_src = crate::varlink::CfsctlService::new();
                let service_dest = crate::varlink::CfsctlService::new();

                let (mut conn_src, _srv_src) = crate::varlink::spawn_in_process(service_src)
                    .context("spawning source in-process service")?;
                let (mut conn_dest, _srv_dest) = crate::varlink::spawn_in_process(service_dest)
                    .context("spawning destination in-process service")?;

                let handle_src = conn_src
                    .open_repository(Some(from_str), None, None)
                    .await
                    .context("zlink transport error calling OpenRepository on source")?
                    .map_err(|e| anyhow::anyhow!("OpenRepository failed on source: {e:?}"))?
                    .handle;

                let handle_dest = conn_dest
                    .open_repository(Some(dest_str), None, None)
                    .await
                    .context("zlink transport error calling OpenRepository on destination")?
                    .map_err(|e| anyhow::anyhow!("OpenRepository failed on destination: {e:?}"))?
                    .handle;

                let finalize_reply = copy_image(
                    &mut conn_src,
                    &mut conn_dest,
                    handle_src,
                    handle_dest,
                    image,
                    name.as_deref(),
                    zerocopy,
                )
                .await?;

                let tag_info = if let Some(n) = name {
                    format!(", tagged as {n}")
                } else {
                    String::new()
                };
                println!(
                    "Copied image {image} from {} to destination repo{}",
                    from.display(),
                    tag_info
                );
                println!("Manifest digest: {}", finalize_reply.manifest_digest);
                println!("Manifest verity: {}", finalize_reply.manifest_verity);
                println!("Config digest:   {}", finalize_reply.config_digest);
                println!("Config verity:   {}", finalize_reply.config_verity);
            }
            OciCommand::ListImages { json } => {
                let images = composefs_oci::oci_image::list_images(&repo)?;

                if json {
                    let reply = crate::varlink::ListImagesReply {
                        images: images
                            .iter()
                            .map(crate::varlink::ImageEntry::from)
                            .collect(),
                    };
                    serde_json::to_writer_pretty(std::io::stdout().lock(), &reply)?;
                    println!();
                } else if images.is_empty() {
                    println!("No images found");
                } else {
                    let mut table = Table::new();
                    table.load_preset(UTF8_FULL);
                    table.set_header(["NAME", "DIGEST", "ARCH", "LAYERS", "REFS"]);

                    for img in images {
                        let digest_str: &str = img.manifest_digest.as_ref();
                        let digest_short = digest_str.strip_prefix("sha256:").unwrap_or(digest_str);
                        let digest_display = if digest_short.len() > 12 {
                            &digest_short[..12]
                        } else {
                            digest_short
                        };
                        let arch = if img.architecture.is_empty() {
                            "artifact"
                        } else {
                            &img.architecture
                        };
                        table.add_row([
                            img.name.as_str(),
                            digest_display,
                            arch,
                            &img.layer_count.to_string(),
                            &img.referrer_count.to_string(),
                        ]);
                    }
                    println!("{table}");
                }
            }
            OciCommand::Inspect {
                ref image,
                manifest,
                config,
            } => {
                let img = resolve_oci_image(&repo, image)?;

                if manifest {
                    // Output raw manifest JSON exactly as stored
                    let manifest_json = img.read_manifest_json(&repo)?;
                    std::io::Write::write_all(&mut std::io::stdout(), &manifest_json)?;
                    println!();
                } else if config {
                    // Output raw config JSON exactly as stored
                    let config_json = img.read_config_json(&repo)?;
                    std::io::Write::write_all(&mut std::io::stdout(), &config_json)?;
                    println!();
                } else {
                    // Default: output combined JSON with manifest, config, and referrers
                    let output = crate::varlink::OciInspectReply::from_image(&repo, &img)?;
                    serde_json::to_writer_pretty(std::io::stdout().lock(), &output)?;
                    println!();
                }
            }
            OciCommand::Tag {
                ref manifest_digest,
                ref name,
            } => {
                composefs_oci::oci_image::tag_image(&repo, manifest_digest, name)?;
                println!("Tagged {manifest_digest} as {name}");
            }
            OciCommand::Untag { ref name } => {
                composefs_oci::oci_image::untag_image(&repo, name)?;
                println!("Removed tag {name}");
            }
            OciCommand::LayerInspect {
                ref layer,
                dumpfile,
                json,
            } => {
                if json {
                    let info = composefs_oci::layer_info(&repo, layer)?;
                    serde_json::to_writer_pretty(std::io::stdout().lock(), &info)?;
                    println!();
                } else if dumpfile {
                    composefs_oci::layer_dumpfile(&repo, layer, &mut std::io::stdout())?;
                } else {
                    // Default: output raw tar, but not to a tty
                    let mut out = std::io::stdout().lock();
                    if out.is_terminal() {
                        anyhow::bail!(
                            "Refusing to write tar data to terminal. \
                            Redirect to a file, pipe to tar, or use --json for metadata."
                        );
                    }
                    composefs_oci::layer_tar(&repo, layer, &mut out)?;
                }
            }

            OciCommand::PrepareBoot {
                config_opts:
                    OCIConfigOptions {
                        ref config_name,
                        ref config_verity,
                    },
                ref bootdir,
                ref entry_id,
                ref cmdline,
            } => {
                let verity = verity_opt(config_verity)?;
                let (config_digest, config_verity) =
                    resolve_oci_config(&repo, config_name, verity)?;
                let mut fs = composefs_oci::image::create_filesystem(
                    &repo,
                    &config_digest,
                    config_verity.as_ref(),
                )?;
                let entries = fs.transform_for_boot(&repo)?;
                let ids = fs.commit_images(&repo, None)?;
                let fmt_config = repo.default_format_config();
                // Prefer V1 digest; fall back to V2.
                let id = ids
                    .get(&FormatVersion::V1)
                    .or_else(|| ids.get(&FormatVersion::V2))
                    .ok_or_else(|| anyhow::anyhow!("commit_images produced no images"))?
                    .clone();

                let insecure = repo.is_insecure();
                let karg = if fmt_config.default == FormatVersion::V1
                    && !fmt_config.extra.contains(&FormatVersion::V2)
                {
                    // V1-only repo → composefs.digest=v1-...: (with optional ? for insecure)
                    ComposefsCmdline::new_v1(id, insecure)
                } else {
                    // BOTH or V2-only repo → composefs= (with optional ? for insecure)
                    ComposefsCmdline::new_v2(id, insecure)
                };

                let Some(entry) = entries.into_iter().next() else {
                    anyhow::bail!("No boot entries!");
                };

                let cmdline_refs: Vec<&str> = cmdline.iter().map(String::as_str).collect();
                write_boot::write_boot_simple(
                    &repo,
                    entry,
                    &karg,
                    bootdir,
                    None,
                    entry_id.as_deref(),
                    &cmdline_refs,
                )?;

                let state = args
                    .repo
                    .as_ref()
                    .map(|p: &PathBuf| p.parent().unwrap())
                    .unwrap_or(Path::new("/sysroot"))
                    .join("state/deploy")
                    .join(karg.digest().to_hex());

                create_dir_all(state.join("var"))?;
                create_dir_all(state.join("etc/upper"))?;
                create_dir_all(state.join("etc/work"))?;
            }
            OciCommand::Fsck { image, json } => {
                let result = if let Some(ref name) = image {
                    composefs_oci::oci_fsck_image(&repo, name).await?
                } else {
                    composefs_oci::oci_fsck(&repo).await?
                };
                if json {
                    let output = crate::varlink::OciFsckReply::from(&result);
                    serde_json::to_writer_pretty(std::io::stdout().lock(), &output)?;
                    println!();
                } else {
                    print!("{result}");
                    if !result.is_ok() {
                        anyhow::bail!("OCI integrity check failed");
                    }
                }
            }
            OciCommand::Varlink { .. } => {
                unreachable!("oci varlink is handled before opening a repository");
            }
        },
        #[cfg(feature = "ostree")]
        Command::Ostree { cmd: ostree_cmd } => match ostree_cmd {
            OstreeCommand::PullLocal {
                ref ostree_repo_path,
                ref ostree_ref,
                base_name,
            } => {
                let ostree_repo =
                    composefs_ostree::LocalRepo::open_path(&repo, CWD, ostree_repo_path)?;
                let reporter: SharedReporter = IndicatifReporter::new().into_shared();
                let opts = composefs_ostree::PullOptions {
                    base_reference: base_name.as_deref(),
                    progress: Some(reporter),
                    ..Default::default()
                };
                let (verity, stats) =
                    composefs_ostree::pull(&repo, ostree_repo, ostree_ref, opts).await?;

                let image_id = composefs_ostree::get_image_ref(&repo, &stats.commit_id)?;
                println!("commit  {}", stats.commit_id);
                println!("verity  {}", verity.to_hex());
                println!("image   {}", image_id.to_hex());
                if !composefs_ostree::is_commit_id(ostree_ref) {
                    println!("tagged  {ostree_ref}");
                }
                print_pull_stats(&stats);
            }
            OstreeCommand::Pull {
                ref ostree_repo_url,
                ref ostree_ref,
                base_name,
                no_delta,
            } => {
                let ostree_repo = composefs_ostree::RemoteRepo::new(&repo, ostree_repo_url)?;
                let reporter: SharedReporter = IndicatifReporter::new().into_shared();
                let opts = composefs_ostree::PullOptions {
                    base_reference: base_name.as_deref(),
                    progress: Some(reporter),
                    disable_deltas: no_delta,
                };
                let (verity, stats) =
                    composefs_ostree::pull(&repo, ostree_repo, ostree_ref, opts).await?;

                let image_id = composefs_ostree::get_image_ref(&repo, &stats.commit_id)?;
                println!("commit  {}", stats.commit_id);
                println!("verity  {}", verity.to_hex());
                println!("image   {}", image_id.to_hex());
                if !composefs_ostree::is_commit_id(ostree_ref) {
                    println!("tagged  {ostree_ref}");
                }
                print_pull_stats(&stats);
            }
            OstreeCommand::Mount {
                ref commit,
                ref mountpoint,
                ref mount_opts,
            } => {
                let image_id = composefs_ostree::get_image_ref(&repo, commit)?;
                mount_opts.mount_image(&repo, &image_id.to_hex(), mountpoint.as_str())?;
            }
            OstreeCommand::Dump { ref commit_name } => {
                let fs = composefs_ostree::create_filesystem(&repo, commit_name)?;
                fs.print_dumpfile()?;
            }
            OstreeCommand::ComputeId { ref commit_name } => {
                let image_id = composefs_ostree::ensure_ostree_erofs(&repo, commit_name)?;
                println!("{}", image_id.to_hex());
            }
            OstreeCommand::Inspect {
                ref source,
                metadata,
            } => {
                composefs_ostree::inspect(&repo, source, metadata)?;
            }
            OstreeCommand::Tag {
                ref source,
                ref name,
            } => {
                composefs_ostree::tag(&repo, source, name)?;
                println!("Tagged {source} as {name}");
            }
            OstreeCommand::Untag { ref name } => {
                composefs_ostree::untag(&repo, name)?;
            }
            OstreeCommand::Commit {
                ref image,
                ref reference,
                ref subject,
            } => {
                use std::time::{SystemTime, UNIX_EPOCH};

                let (img_fd, _) = repo.open_image(image)?;
                let mut img_buf = Vec::new();
                std::fs::File::from(img_fd).read_to_end(&mut img_buf)?;
                let fs = composefs::erofs::reader::erofs_to_filesystem(&img_buf)?;

                let timestamp = SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap_or_default()
                    .as_secs();
                let mut commit_meta = composefs_ostree::ostree::CommitMetadata::default()
                    .subject(subject.as_str())
                    .timestamp(timestamp);
                if let Some(ref_name) = reference {
                    commit_meta = commit_meta.add_metadata(
                        "ostree.ref-binding",
                        composefs_ostree::ostree::MetadataValue::StringArray(vec![
                            ref_name.clone(),
                        ]),
                    );
                }

                let (verity, commit_id) = composefs_ostree::commit_filesystem(
                    &repo,
                    &fs,
                    commit_meta,
                    reference.as_deref(),
                )?;
                println!("commit  {commit_id}");
                println!("verity  {}", verity.to_hex());
                if let Some(ref_name) = reference {
                    println!("tagged  {ref_name}");
                }
            }
            OstreeCommand::Export {
                ref source,
                ref ostree_repo_path,
                ref reference,
            } => {
                let dest = composefs_ostree::LocalRepo::open_path(&repo, CWD, ostree_repo_path)?;
                let commit_id =
                    composefs_ostree::export_commit(&repo, source, &dest, reference.as_deref())?;
                println!("commit  {commit_id}");
                if let Some(ref_name) = reference {
                    println!("tagged  {ref_name}");
                }
            }
            OstreeCommand::ListCommits => {
                let commits = composefs_ostree::list_commits(&repo)?;
                if commits.is_empty() {
                    println!("No ostree commits found");
                } else {
                    let mut table = Table::new();
                    table.load_preset(UTF8_FULL);
                    table.set_header(["NAME", "COMMIT"]);
                    for c in commits {
                        table.add_row([c.name.as_str(), &c.commit_id]);
                    }
                    println!("{table}");
                }
            }
            OstreeCommand::ApplyDelta { ref delta_path } => {
                let (verity, stats) = composefs_ostree::apply_delta_offline(&repo, delta_path)?;
                let image_id = composefs_ostree::get_image_ref(&repo, &stats.commit_id)?;
                println!("commit  {}", stats.commit_id);
                println!("verity  {}", verity.to_hex());
                println!("image   {}", image_id.to_hex());
                println!(
                    "objects {} metadata + {} files applied",
                    stats.metadata_fetched, stats.files_fetched
                );
            }
            OstreeCommand::ListRefs {
                ref ostree_repo_url,
                ref subset,
            } => {
                let mut ostree_repo = composefs_ostree::RemoteRepo::new(&repo, ostree_repo_url)?;
                if let Some(s) = subset {
                    ostree_repo = ostree_repo.with_summary_subset(s);
                }
                let refs = ostree_repo.list_remote_refs().await?;
                if refs.is_empty() {
                    println!("No refs found");
                } else {
                    let mut table = Table::new();
                    table.load_preset(UTF8_FULL);
                    table.set_header(["REF", "COMMIT"]);
                    for (name, checksum) in &refs {
                        table.add_row([name.as_str(), &hex::encode(checksum)]);
                    }
                    println!("{table}");
                }
            }
        },
        Command::CreateImage {
            fs_opts,
            ref image_name,
        } => {
            let fs = load_filesystem_from_ondisk_fs(&fs_opts, Some(Arc::clone(&repo))).await?;
            let id = fs.commit_image(&repo, image_name.as_deref())?;
            println!("{}", id.to_id());
        }
        Command::ComputeId { .. }
        | Command::ComputeKarg { .. }
        | Command::CreateDumpfile { .. } => {
            // Handled in run_app before opening the repo
            unreachable!(
                "compute-id, compute-karg, and create-dumpfile are dispatched without a repo"
            );
        }
        Command::Mount {
            name,
            mountpoint,
            ref mount_opts,
        } => {
            mount_opts.mount_image(&repo, &name, &mountpoint)?;
        }
        Command::Images { json, no_trunc } => {
            let reply =
                varlink::run_list_image_refs(&repo).map_err(|e| anyhow::anyhow!("{e:?}"))?;

            if json {
                serde_json::to_writer_pretty(std::io::stdout().lock(), &reply)?;
                println!();
            } else if reply.images.is_empty() {
                println!("No images found");
            } else {
                let mut table = Table::new();
                table.load_preset(UTF8_FULL);
                table.set_header(["NAME", "DIGEST"]);

                for entry in &reply.images {
                    let digest_display = if !no_trunc && entry.digest.len() > 12 {
                        &entry.digest[..12]
                    } else {
                        &entry.digest
                    };
                    table.add_row([entry.name.as_str(), digest_display]);
                }
                println!("{table}");
            }
        }
        Command::ImageObjects { name } => {
            let objects = repo.objects_for_image(&name)?;
            for object in objects {
                println!("{}", object.to_id());
            }
        }
        Command::GC { root, dry_run } => {
            let roots: Vec<&str> = root.iter().map(|s| s.as_str()).collect();
            let result = if dry_run {
                repo.gc_dry_run(&roots)?
            } else {
                repo.gc(&roots)?
            };
            if dry_run {
                println!("Dry run (no files deleted):");
            }
            println!(
                "Objects: {} removed ({} bytes)",
                result.objects_removed, result.objects_bytes
            );
            if result.images_pruned > 0 || result.streams_pruned > 0 {
                println!(
                    "Pruned symlinks: {} images, {} streams",
                    result.images_pruned, result.streams_pruned
                );
            }
        }
        Command::DumpFiles {
            image_name,
            files,
            backing_path_only,
        } => {
            let out = dump_files(&repo, &image_name, &files, backing_path_only)?;

            if !out.is_empty() {
                let out_str = std::str::from_utf8(&out).unwrap();
                print!("{}", out_str);
            }
        }
        Command::Fsck {
            json,
            metadata_only,
        } => {
            let result = if metadata_only {
                repo.fsck_metadata_only().await?
            } else {
                repo.fsck().await?
            };
            if json {
                let output = crate::varlink::FsckReply::from(&result);
                serde_json::to_writer_pretty(std::io::stdout().lock(), &output)?;
                println!();
            } else {
                print!("{result}");
                if !result.is_ok() {
                    anyhow::bail!("repository integrity check failed");
                }
            }
        }
        Command::Varlink { .. } => {
            // Handled in run_app before opening the repo.
            unreachable!("varlink is handled before opening a repository");
        }
        #[cfg(feature = "http")]
        Command::Fetch { url, name } => {
            let reporter: SharedReporter = IndicatifReporter::new().into_shared();
            let (digest, verity) = composefs_http::download(
                &url,
                &name,
                Arc::clone(&repo),
                composefs_http::DownloadOptions {
                    progress: Some(reporter),
                },
            )
            .await?;
            println!("content {digest}");
            println!("verity {}", verity.to_hex());
        }
        Command::Mkcomposefs { .. } | Command::ComposefsInfo { .. } => {
            // Dispatched in run_app before a repository is opened
            unreachable!("mkcomposefs/composefs-info are dispatched before opening a repository");
        }
    }
    Ok(())
}

#[cfg(test)]
#[cfg(any(feature = "oci", feature = "http"))]
mod tests {
    use super::*;
    use composefs::progress::{ProgressEvent, ProgressUnit};

    // ── IndicatifReporter ────────────────────────────────────────────────────

    /// A complete valid lifecycle (Started → Progress → Done) must not panic,
    /// even without a real terminal (indicatif handles headless gracefully).
    #[test]
    fn test_indicatif_reporter_valid_lifecycle() {
        let reporter = IndicatifReporter::new();
        // Message before any component
        reporter.report(ProgressEvent::Message("starting pull".into()));
        // Byte-tracked component
        reporter.report(ProgressEvent::Started {
            id: "sha256:abc".into(),
            total: Some(1_000_000),
            unit: ProgressUnit::Bytes,
        });
        reporter.report(ProgressEvent::Progress {
            id: "sha256:abc".into(),
            fetched: 500_000,
            total: Some(1_000_000),
        });
        reporter.report(ProgressEvent::Done {
            id: "sha256:abc".into(),
            transferred: 1_000_000,
        });
        // Item-counted component (HTTP objects)
        reporter.report(ProgressEvent::Started {
            id: "objects:stream".into(),
            total: Some(200),
            unit: ProgressUnit::Items,
        });
        reporter.report(ProgressEvent::Progress {
            id: "objects:stream".into(),
            fetched: 100,
            total: Some(200),
        });
        reporter.report(ProgressEvent::Done {
            id: "objects:stream".into(),
            transferred: 200,
        });
        // Skipped component
        reporter.report(ProgressEvent::Started {
            id: "sha256:cached".into(),
            total: None,
            unit: ProgressUnit::Bytes,
        });
        reporter.report(ProgressEvent::Skipped {
            id: "sha256:cached".into(),
        });
    }

    /// Progress/Done events for an ID that was never `Started` must not panic.
    ///
    /// This guards against error-recovery paths where a `Started` event may
    /// have been suppressed or the reporter was attached after the operation
    /// began.
    #[test]
    fn test_indicatif_reporter_unknown_id_no_panic() {
        let reporter = IndicatifReporter::new();
        // Progress for unknown ID — should silently ignore
        reporter.report(ProgressEvent::Progress {
            id: "ghost".into(),
            fetched: 42,
            total: None,
        });
        // Done for unknown ID — should silently ignore
        reporter.report(ProgressEvent::Done {
            id: "ghost".into(),
            transferred: 42,
        });
        // Skipped for unknown ID — should silently ignore
        reporter.report(ProgressEvent::Skipped { id: "ghost".into() });
    }

    /// A spinner-style bar (unknown total) must not panic.
    #[test]
    fn test_indicatif_reporter_spinner_lifecycle() {
        let reporter = IndicatifReporter::new();
        // Started with unknown total → spinner
        reporter.report(ProgressEvent::Started {
            id: "layer:unknown-size".into(),
            total: None,
            unit: ProgressUnit::Bytes,
        });
        reporter.report(ProgressEvent::Progress {
            id: "layer:unknown-size".into(),
            fetched: 1024,
            total: None,
        });
        reporter.report(ProgressEvent::Done {
            id: "layer:unknown-size".into(),
            transferred: 2048,
        });
    }

    /// Multiple concurrent components must not interfere with each other.
    #[test]
    fn test_indicatif_reporter_multiple_concurrent_components() {
        let reporter = IndicatifReporter::new();
        // Start two layers in parallel
        reporter.report(ProgressEvent::Started {
            id: "layer:a".into(),
            total: Some(100),
            unit: ProgressUnit::Bytes,
        });
        reporter.report(ProgressEvent::Started {
            id: "layer:b".into(),
            total: Some(200),
            unit: ProgressUnit::Bytes,
        });
        // Interleaved progress
        reporter.report(ProgressEvent::Progress {
            id: "layer:a".into(),
            fetched: 50,
            total: Some(100),
        });
        reporter.report(ProgressEvent::Progress {
            id: "layer:b".into(),
            fetched: 100,
            total: Some(200),
        });
        // Layer B finishes first
        reporter.report(ProgressEvent::Done {
            id: "layer:b".into(),
            transferred: 200,
        });
        // Layer A finishes
        reporter.report(ProgressEvent::Done {
            id: "layer:a".into(),
            transferred: 100,
        });
    }
}