vole-document 0.1.0-alpha.15

//! `vole-document` command-line interface.
//!
//! File output is written atomically: a temporary sibling is written, fsynced,
//! then renamed into place. A partial or failed operation never replaces the
//! destination.

use std::fs;
#[cfg(feature = "rans")]
use std::fs::File;
use std::io::Write;
#[cfg(feature = "rans")]
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::time::{SystemTime, UNIX_EPOCH};

use vole_document::adapter::pdf;
#[cfg(feature = "store")]
use vole_document::container::Descriptor;
use vole_document::container::UNIVERSE;
#[cfg(feature = "rans")]
use vole_document::container::header::{FEATURE_SEEK_DIRECTORY, HEADER_LEN, Header};
use vole_document::dra::Op;
use vole_document::encode::candidates::CandidateKind;
use vole_document::error::{Error, Result};
#[cfg(feature = "field")]
use vole_document::field::{
    Field, FieldId, FieldStore,
    cache::DerivedCache,
    edit as field_edit,
    explain::explain,
    ingest as field_ingest,
    observe::{ObserveRequest, ObserveStats, Representation, Selector, observe},
    provenance::{AnswerValue, FieldAnswer},
    share,
};
use vole_document::limits::Limits;
#[cfg(feature = "rans")]
use vole_document::materialize::observation::{ObservationReport, ObservationSelector};
#[cfg(feature = "store")]
use vole_document::store::{EmbeddedStore, account, externalize, gc};
use vole_document::{encode, integrity, materialize};

const USAGE_HEAD: &str = "\
vole-document — byte-exact procedural document storage

USAGE:
    vole-document encode [--force KIND] INPUT  OUTPUT.voldoc
    vole-document decode     INPUT.voldoc     OUTPUT
    vole-document materialize INPUT.voldoc    OUTPUT
    vole-document verify     INPUT.voldoc
    vole-document inspect    INPUT.voldoc
    vole-document pdf-inspect INPUT
    vole-document pdf-make-samples DIR
    vole-document pdf-make-large DIR [OBJECTS]
";

/// The `deflate-stats` line is advertised only when the replay stack is built in.
#[cfg(feature = "deflate-replay")]
const USAGE_DEFLATE_STATS: &str = "    vole-document deflate-stats INPUT...\n";
#[cfg(not(feature = "deflate-replay"))]
const USAGE_DEFLATE_STATS: &str = "";

/// The `view` line is advertised only when the entropy decoder is built in.
#[cfg(feature = "rans")]
const USAGE_VIEW: &str = "\
    vole-document view      INPUT.voldoc [OUTPUT] --byte-range A:L | --pdf-object N:G |\n\
        --pdf-stream N:G | --pdf-revision I [--stats]\n";
#[cfg(not(feature = "rans"))]
const USAGE_VIEW: &str = "";

/// The store lines are advertised only when the object store is built in.
#[cfg(feature = "store")]
const USAGE_STORE: &str = "\
    vole-document decode --store STORE_DIR INPUT.voldoc OUTPUT\n\
    vole-document store put     INPUT.voldoc STORE_DIR\n\
    vole-document store account STORE_DIR ROOT...\n\
    vole-document store gc      STORE_DIR ROOT...\n";
#[cfg(not(feature = "store"))]
const USAGE_STORE: &str = "";

/// The field observation verbs are advertised only when the field is built in.
#[cfg(feature = "field")]
const USAGE_FIELD: &str = "\
    vole-document field-ingest INPUT.voldoc --store DIR [--entropyfs]
    vole-document field-edit --store DIR --field HEX --page N --content FILE [--entropyfs]
    vole-document observe --store DIR --field HEX [--entropyfs] (--page N | --object N | --stream N |
        --revision N | --byte-range A..B) --kind metadata|text|structure|operators|
        encoded|decoded|exact|preview|full
    vole-document find    --store DIR --field HEX --text PATTERN [--entropyfs]
    vole-document explain --store DIR --field HEX <selector> --kind KIND [--analyze] [--entropyfs]
    vole-document preview --store DIR --field HEX --page N [--json] [--entropyfs]
    vole-document materialize --store DIR --field HEX --exact --output FILE [--entropyfs]
    vole-document cache  --store DIR [--clear] [--entropyfs]
    vole-document field-store-stats --store DIR [--entropyfs]
    (--entropyfs needs a build with the entropyfs-store feature)
";
#[cfg(not(feature = "field"))]
const USAGE_FIELD: &str = "";

/// The fine-unit sharing verbs are advertised only when the field is built in.
#[cfg(feature = "field")]
const USAGE_SHARE: &str = "\
    vole-document share report INPUT.voldoc...\n\
    vole-document share-account --store DIR INPUT.voldoc...\n\
    vole-document share externalize --store DIR INPUT.voldoc OUTPUT.voldoc\n";
#[cfg(not(feature = "field"))]
const USAGE_SHARE: &str = "";

const USAGE_TAIL: &str = "\
    vole-document capabilities

KIND (for encode --force): raw | rle | byte-rans | pdf-physical | pdf-channels |
    pdf-layout | pdf-layout-rans | pdf-deflate-replay | pdf-deflate-replay-rans |
    pdf-deflate-replay-rans-indexed
    Forces the complete-cost court to consider only that candidate family, for
    honest per-mechanism ablation. Fails when the input does not propose it.

EXIT CODES:
    0 ok   2 usage   3 io   4 invalid-container   5 unsupported-version
    6 unsupported-feature   7 integrity-mismatch   8 resource-limit
    9 invalid-graph  10 invalid-model  15 coverage-violation
    16 reconstruction-mismatch  70 internal-invariant
";

/// The full usage text, with the replay-gated command line included only when
/// the feature is present.
fn usage() -> String {
    format!(
        "{USAGE_HEAD}{USAGE_VIEW}{USAGE_DEFLATE_STATS}{USAGE_STORE}{USAGE_FIELD}{USAGE_SHARE}{USAGE_TAIL}"
    )
}

fn main() -> ExitCode {
    let args: Vec<String> = std::env::args().collect();
    match run(&args) {
        Ok(()) => ExitCode::SUCCESS,
        Err(e) => {
            eprintln!("error: {e}");
            let code = e.exit_code();
            ExitCode::from(u8::try_from(code).unwrap_or(70))
        }
    }
}

fn run(args: &[String]) -> Result<()> {
    // Hidden internal replay-worker mode: handled before any normal parsing,
    // because it reads a framed request from stdin and never returns. It is not
    // advertised in USAGE.
    #[cfg(feature = "deflate-replay")]
    {
        use vole_document::codec::deflate::{
            REPLAY_WORKER_SUBCOMMAND, install_default_replay_worker, run_worker_stdio,
        };
        if args.get(1).map(String::as_str) == Some(REPLAY_WORKER_SUBCOMMAND) {
            run_worker_stdio();
        }
        // Isolate DEFLATE replay by default: point the library at this binary.
        // `std::env::set_var` is `unsafe` under Rust 2024 (and this crate forbids
        // `unsafe`), so the path is installed via a safe library setter; an
        // explicit `VOLE_REPLAY_WORKER` still takes precedence.
        if std::env::var_os("VOLE_REPLAY_WORKER").is_none()
            && let Ok(exe) = std::env::current_exe()
        {
            install_default_replay_worker(exe);
        }
    }

    let cmd = match args.get(1).map(String::as_str) {
        Some(c) => c,
        None => {
            print!("{}", usage());
            return Err(Error::usage("no subcommand given"));
        }
    };
    let limits = Limits::DEFAULT;

    match cmd {
        "-h" | "--help" | "help" => {
            print!("{}", usage());
            Ok(())
        }
        "capabilities" => cmd_capabilities(),
        "encode" => cmd_encode_args(args, limits),
        "decode" | "materialize" => {
            #[cfg(feature = "field")]
            if args
                .iter()
                .any(|a| a.as_str() == "--field" || a.starts_with("--field="))
            {
                return cmd_field_materialize(args, limits);
            }
            cmd_decode_args(args, limits)
        }
        "verify" => {
            let input = arg(args, 2, "INPUT.voldoc")?;
            cmd_verify(&input, limits)
        }
        "inspect" => {
            let input = arg(args, 2, "INPUT.voldoc")?;
            cmd_inspect(&input, limits)
        }
        #[cfg(feature = "rans")]
        "view" => cmd_view_args(args, limits),
        #[cfg(feature = "store")]
        "store" => cmd_store_args(args, limits),
        "pdf-inspect" => {
            let input = arg(args, 2, "INPUT")?;
            cmd_pdf_inspect(&input, limits)
        }
        "pdf-make-samples" => {
            let dir = arg(args, 2, "DIR")?;
            cmd_pdf_make_samples(&dir)
        }
        "pdf-make-large" => {
            let dir = arg(args, 2, "DIR")?;
            let objects = args.get(3).map(String::as_str);
            cmd_pdf_make_large(&dir, objects)
        }
        #[cfg(feature = "deflate-replay")]
        "deflate-stats" => {
            let inputs: Vec<PathBuf> = args
                .get(2..)
                .unwrap_or(&[])
                .iter()
                .map(PathBuf::from)
                .collect();
            if inputs.is_empty() {
                return Err(Error::usage(
                    "deflate-stats requires at least one INPUT (a PDF)",
                ));
            }
            cmd_deflate_stats(&inputs, limits)
        }
        #[cfg(feature = "field")]
        "field-ingest" => cmd_field_ingest(args, limits),
        #[cfg(feature = "field")]
        "field-edit" => cmd_field_edit(args, limits),
        #[cfg(feature = "field")]
        "observe" => cmd_field_observe(args, limits),
        #[cfg(feature = "field")]
        "find" => cmd_field_find(args, limits),
        #[cfg(feature = "field")]
        "explain" => cmd_field_explain(args, limits),
        #[cfg(feature = "field")]
        "preview" => cmd_field_preview(args, limits),
        #[cfg(feature = "field")]
        "cache" => cmd_field_cache(args),
        #[cfg(feature = "field")]
        "field-store-stats" => cmd_field_store_stats(args),
        #[cfg(feature = "field")]
        "share" => cmd_share_args(args, limits),
        #[cfg(feature = "field")]
        "share-account" => cmd_share_account(args, limits),
        other => Err(Error::usage(format!(
            "unknown subcommand {other:?}\n\n{}",
            usage()
        ))),
    }
}

fn arg(args: &[String], idx: usize, name: &str) -> Result<PathBuf> {
    args.get(idx)
        .map(PathBuf::from)
        .ok_or_else(|| Error::usage(format!("missing argument {name}")))
}

fn cmd_encode(
    input: &Path,
    output: &Path,
    limits: Limits,
    force: Option<CandidateKind>,
) -> Result<()> {
    let source = fs::read(input)?;
    if source.len() as u64 > limits.max_input_bytes {
        return Err(Error::resource_limit(
            "input exceeds configured input limit",
        ));
    }
    let (bytes, report) = encode::encode_with(&source, limits, force)?;
    write_atomic(output, &bytes)?;

    println!("{}", report_json(&report));
    Ok(())
}

/// Parse the `encode` subcommand arguments: an optional `--force KIND` flag and
/// the two positional paths `INPUT` and `OUTPUT.voldoc`.
///
/// The flag accepts either `--force KIND` or `--force=KIND` and may appear
/// before or after the positionals, so existing `encode IN OUT` invocations are
/// unchanged.
fn cmd_encode_args(args: &[String], limits: Limits) -> Result<()> {
    let mut force: Option<CandidateKind> = None;
    let mut positional: Vec<&str> = Vec::new();
    let mut i = 2;
    while i < args.len() {
        let a = args[i].as_str();
        if a == "--force" {
            let kind = args
                .get(i + 1)
                .ok_or_else(|| Error::usage("--force requires a KIND argument"))?;
            force = Some(parse_force_kind(kind)?);
            i += 2;
        } else if let Some(kind) = a.strip_prefix("--force=") {
            force = Some(parse_force_kind(kind)?);
            i += 1;
        } else {
            positional.push(a);
            i += 1;
        }
    }
    let input = positional
        .first()
        .map(PathBuf::from)
        .ok_or_else(|| Error::usage("missing argument INPUT"))?;
    let output = positional
        .get(1)
        .map(PathBuf::from)
        .ok_or_else(|| Error::usage("missing argument OUTPUT.voldoc"))?;
    if positional.len() > 2 {
        return Err(Error::usage(format!(
            "unexpected extra argument {:?}",
            positional[2]
        )));
    }
    cmd_encode(&input, &output, limits, force)
}

/// Map a `--force` KIND spelling to a [`CandidateKind`].
fn parse_force_kind(s: &str) -> Result<CandidateKind> {
    match s {
        "raw" => Ok(CandidateKind::Raw),
        "rle" => Ok(CandidateKind::Rle),
        "byte-rans" => Ok(CandidateKind::ByteRans),
        "pdf-physical" => Ok(CandidateKind::PdfPhysical),
        "pdf-channels" => Ok(CandidateKind::PdfChannels),
        "pdf-layout" => Ok(CandidateKind::PdfLayout),
        "pdf-layout-rans" => Ok(CandidateKind::PdfLayoutRans),
        "pdf-deflate-replay" => Ok(CandidateKind::PdfDeflateReplay),
        "pdf-deflate-replay-rans" => Ok(CandidateKind::PdfDeflateReplayRans),
        "pdf-deflate-replay-rans-indexed" => Ok(CandidateKind::PdfDeflateReplayRansIndexed),
        other => Err(Error::usage(format!(
            "unknown --force kind {other:?}; expected one of raw, rle, byte-rans, pdf-physical, pdf-channels, pdf-layout, pdf-layout-rans, pdf-deflate-replay, pdf-deflate-replay-rans, pdf-deflate-replay-rans-indexed"
        ))),
    }
}

/// Parse and dispatch the `view` subcommand: serve a narrow observation of a
/// `.voldoc` that carries an `OBSERVATION_INDEX`.
///
/// Exactly one selector flag is required. `--stats` with no `OUTPUT` prints the
/// stats object only (no bytes); otherwise bytes go to `OUTPUT` atomically or to
/// stdout, with the stats object on stdout (or stderr when bytes occupy stdout).
#[cfg(feature = "rans")]
fn cmd_view_args(args: &[String], limits: Limits) -> Result<()> {
    let mut selector: Option<ObservationSelector> = None;
    let mut stats = false;
    let mut positional: Vec<&str> = Vec::new();
    let mut i = 2;
    while i < args.len() {
        let a = args[i].as_str();
        if a == "--stats" {
            stats = true;
            i += 1;
        } else if let Some(v) = a.strip_prefix("--byte-range=") {
            set_selector(&mut selector, parse_byte_range(v)?)?;
            i += 1;
        } else if let Some(v) = a.strip_prefix("--pdf-object=") {
            set_selector(&mut selector, parse_pdf_object(v)?)?;
            i += 1;
        } else if let Some(v) = a.strip_prefix("--pdf-stream=") {
            set_selector(&mut selector, parse_pdf_stream(v)?)?;
            i += 1;
        } else if let Some(v) = a.strip_prefix("--pdf-revision=") {
            set_selector(&mut selector, parse_pdf_revision(v)?)?;
            i += 1;
        } else if a == "--byte-range" {
            let v = args
                .get(i + 1)
                .ok_or_else(|| Error::usage("--byte-range requires A:L"))?;
            set_selector(&mut selector, parse_byte_range(v)?)?;
            i += 2;
        } else if a == "--pdf-object" {
            let v = args
                .get(i + 1)
                .ok_or_else(|| Error::usage("--pdf-object requires N:G"))?;
            set_selector(&mut selector, parse_pdf_object(v)?)?;
            i += 2;
        } else if a == "--pdf-stream" {
            let v = args
                .get(i + 1)
                .ok_or_else(|| Error::usage("--pdf-stream requires N:G"))?;
            set_selector(&mut selector, parse_pdf_stream(v)?)?;
            i += 2;
        } else if a == "--pdf-revision" {
            let v = args
                .get(i + 1)
                .ok_or_else(|| Error::usage("--pdf-revision requires I"))?;
            set_selector(&mut selector, parse_pdf_revision(v)?)?;
            i += 2;
        } else {
            positional.push(a);
            i += 1;
        }
    }

    let selector = selector.ok_or_else(|| {
        Error::usage(
            "view requires exactly one of --byte-range, --pdf-object, --pdf-stream, --pdf-revision",
        )
    })?;
    let input = positional
        .first()
        .map(PathBuf::from)
        .ok_or_else(|| Error::usage("missing argument INPUT.voldoc"))?;
    if positional.len() > 2 {
        return Err(Error::usage(format!(
            "unexpected extra argument {:?}",
            positional[2]
        )));
    }
    let output = positional.get(1).map(PathBuf::from);
    cmd_view(&input, output.as_deref(), selector, stats, limits)
}

/// Record the one selector a `view` invocation may carry.
#[cfg(feature = "rans")]
fn set_selector(
    slot: &mut Option<ObservationSelector>,
    selector: ObservationSelector,
) -> Result<()> {
    if slot.is_some() {
        return Err(Error::usage(
            "view accepts exactly one selector; more than one was given",
        ));
    }
    *slot = Some(selector);
    Ok(())
}

#[cfg(feature = "rans")]
fn parse_byte_range(value: &str) -> Result<ObservationSelector> {
    let (offset, len) = value
        .split_once(':')
        .ok_or_else(|| Error::usage("--byte-range must be A:L (e.g. 1024:4096)"))?;
    let offset: u64 = offset
        .parse()
        .map_err(|_| Error::usage(format!("--byte-range offset {offset:?} is not a u64")))?;
    let len: u64 = len
        .parse()
        .map_err(|_| Error::usage(format!("--byte-range length {len:?} is not a u64")))?;
    Ok(ObservationSelector::ByteRange { offset, len })
}

#[cfg(feature = "rans")]
fn parse_pdf_object(value: &str) -> Result<ObservationSelector> {
    let (object, generation) = parse_n_g(value, "--pdf-object")?;
    Ok(ObservationSelector::PdfIndirectObject { object, generation })
}

#[cfg(feature = "rans")]
fn parse_pdf_stream(value: &str) -> Result<ObservationSelector> {
    let (object, generation) = parse_n_g(value, "--pdf-stream")?;
    Ok(ObservationSelector::PdfEncodedStream { object, generation })
}

#[cfg(feature = "rans")]
fn parse_pdf_revision(value: &str) -> Result<ObservationSelector> {
    let index: u32 = value
        .parse()
        .map_err(|_| Error::usage(format!("--pdf-revision {value:?} is not a u32")))?;
    Ok(ObservationSelector::PdfRevision { index })
}

#[cfg(feature = "rans")]
fn parse_n_g(value: &str, flag: &str) -> Result<(u32, u16)> {
    let (n, g) = value
        .split_once(':')
        .ok_or_else(|| Error::usage(format!("{flag} must be N:G (e.g. 4:0)")))?;
    let object: u32 = n
        .parse()
        .map_err(|_| Error::usage(format!("{flag} object {n:?} is not a u32")))?;
    let generation: u16 = g
        .parse()
        .map_err(|_| Error::usage(format!("{flag} generation {g:?} is not a u16")))?;
    Ok((object, generation))
}

#[cfg(feature = "rans")]
fn cmd_view(
    input: &Path,
    output: Option<&Path>,
    selector: ObservationSelector,
    stats: bool,
    limits: Limits,
) -> Result<()> {
    // Peek only the fixed 64-byte header first. A descriptor that advertises the
    // seek feature is served by the seek reader over the same handle, so the
    // process never `fs::read`s the whole descriptor: the bytes-read claim is a
    // real on-disk-I/O claim, not a post-hoc approximation. A descriptor without
    // one keeps the Phase-7 in-memory path (which must parse the whole file).
    let mut file = File::open(input)?;
    let mut hdr = [0u8; HEADER_LEN];
    file.read_exact(&mut hdr)
        .map_err(|_| Error::invalid_container("truncated header"))?;
    let header = Header::decode(&hdr)?;
    if header.optional_features & FEATURE_SEEK_DIRECTORY != 0 {
        // `materialize_observation_seeked` seeks to offset 0 itself for the
        // header, so rewind our peek rather than leaving the position at 64.
        file.seek(SeekFrom::Start(0))
            .map_err(|e| Error::io(format!("seek to start failed: {e}")))?;
        let report = vole_document::materialize::seek::materialize_observation_seeked(
            file, selector, limits,
        )?;
        let json = observation_json(&selector, header.declared_source_len, &report);
        return emit_view(output, stats, &report, &json);
    }
    drop(file);
    let encoded = fs::read(input)?;
    let parsed = vole_document::container::Descriptor::parse(&encoded, limits)?;
    let report = vole_document::materialize::observation::materialize_observation(
        &parsed, selector, limits,
    )?;
    let json = observation_json(&selector, parsed.descriptor.source_len, &report);
    emit_view(output, stats, &report, &json)
}

/// Write the served bytes (or print stats) exactly as the `view` CLI documents.
#[cfg(feature = "rans")]
fn emit_view(
    output: Option<&Path>,
    stats: bool,
    report: &ObservationReport,
    json: &str,
) -> Result<()> {
    match output {
        Some(path) => {
            write_atomic(path, &report.bytes)?;
            println!("{json}");
        }
        // Stats-only: no bytes are emitted, so stdout stays a single JSON line.
        None if stats => println!("{json}"),
        None => {
            std::io::stdout()
                .write_all(&report.bytes)
                .map_err(Error::from)?;
            eprintln!("{json}");
        }
    }
    Ok(())
}

/// One JSON line describing a served observation and its measured cost.
#[cfg(feature = "rans")]
fn observation_json(
    selector: &ObservationSelector,
    source_len: u64,
    report: &ObservationReport,
) -> String {
    let s = &report.stats;
    format!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"selector\":\"{}\",",
            "\"source_len\":{},",
            "\"range_start\":{},",
            "\"range_len\":{},",
            "\"bytes\":{},",
            "\"ops_evaluated\":{},",
            "\"ops_total\":{},",
            "\"objects_fetched\":{},",
            "\"objects_total\":{},",
            "\"channels_decoded\":{},",
            "\"channels_total\":{},",
            "\"entropy_bytes_decoded\":{},",
            "\"descriptor_bytes_traversed\":{},",
            "\"bytes_read\":{},",
            "\"integrity_verified\":{},",
            "\"output_bytes\":{},",
            "\"work_amplification\":{:.6}",
            "}}"
        ),
        selector_label(selector),
        source_len,
        report.range.0,
        report.range.1.saturating_sub(report.range.0),
        report.bytes.len(),
        s.ops_evaluated,
        s.ops_total,
        s.objects_fetched,
        s.objects_total,
        s.channels_decoded,
        s.channels_total,
        s.entropy_bytes_decoded,
        s.descriptor_bytes_traversed,
        s.bytes_read,
        s.integrity_verified,
        s.output_bytes,
        s.work_amplification(),
    )
}

#[cfg(feature = "rans")]
fn selector_label(selector: &ObservationSelector) -> String {
    match selector {
        ObservationSelector::ByteRange { offset, len } => format!("byte-range:{offset}:{len}"),
        ObservationSelector::PdfIndirectObject { object, generation } => {
            format!("pdf-object:{object}:{generation}")
        }
        ObservationSelector::PdfEncodedStream { object, generation } => {
            format!("pdf-stream:{object}:{generation}")
        }
        ObservationSelector::PdfRevision { index } => format!("pdf-revision:{index}"),
    }
}

fn cmd_decode(input: &Path, output: &Path, limits: Limits) -> Result<()> {
    let encoded = fs::read(input)?;
    let (bytes, parsed) = materialize::decode_to_bytes(&encoded, limits)?;
    write_atomic(output, &bytes)?;
    println!(
        "{{\"ok\":true,\"source_len\":{},\"sha256\":\"{}\",\"graph_ops\":{},\"objects\":{}}}",
        bytes.len(),
        integrity::to_hex(&parsed.descriptor.source_sha256),
        parsed.descriptor.program.ops.len(),
        parsed.descriptor.objects.len()
    );
    Ok(())
}

/// Parse `decode`/`materialize` arguments: an optional `--store STORE_DIR`
/// (store-backed descriptors), then the positional `INPUT.voldoc` and `OUTPUT`.
///
/// A standalone `decode INPUT OUTPUT` invocation is unchanged. Without the
/// `store` cargo feature, `--store` is refused with `UnsupportedFeature` rather
/// than ignored.
fn cmd_decode_args(args: &[String], limits: Limits) -> Result<()> {
    let mut store_dir: Option<PathBuf> = None;
    let mut positional: Vec<&str> = Vec::new();
    let mut i = 2;
    while i < args.len() {
        let a = args[i].as_str();
        if a == "--store" {
            let dir = args
                .get(i + 1)
                .ok_or_else(|| Error::usage("--store requires a STORE_DIR argument"))?;
            store_dir = Some(PathBuf::from(dir));
            i += 2;
        } else if let Some(dir) = a.strip_prefix("--store=") {
            store_dir = Some(PathBuf::from(dir));
            i += 1;
        } else {
            positional.push(a);
            i += 1;
        }
    }
    let input = positional
        .first()
        .map(PathBuf::from)
        .ok_or_else(|| Error::usage("missing argument INPUT.voldoc"))?;
    let output = positional
        .get(1)
        .map(PathBuf::from)
        .ok_or_else(|| Error::usage("missing argument OUTPUT"))?;
    if positional.len() > 2 {
        return Err(Error::usage(format!(
            "unexpected extra argument {:?}",
            positional[2]
        )));
    }
    if let Some(dir) = store_dir {
        #[cfg(feature = "store")]
        {
            return cmd_decode_with_store(&dir, &input, &output, limits);
        }
        #[cfg(not(feature = "store"))]
        {
            let _ = dir;
            return Err(Error::unsupported_feature(
                "this build was compiled without the `store` feature",
            ));
        }
    }
    cmd_decode(&input, &output, limits)
}

/// Materialize a store-backed descriptor, resolving `EXTERNAL_REF` objects
/// through an [`EmbeddedStore`] rooted at `store_dir`.
#[cfg(feature = "store")]
fn cmd_decode_with_store(
    store_dir: &Path,
    input: &Path,
    output: &Path,
    limits: Limits,
) -> Result<()> {
    let encoded = fs::read(input)?;
    let store = EmbeddedStore::open(store_dir)?;
    let (bytes, parsed) = materialize::decode_to_bytes_with(&encoded, &store, limits)?;
    write_atomic(output, &bytes)?;
    println!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"source_len\":{},",
            "\"sha256\":\"{}\",",
            "\"graph_ops\":{},",
            "\"objects\":{},",
            "\"store\":\"{}\"",
            "}}"
        ),
        bytes.len(),
        integrity::to_hex(&parsed.descriptor.source_sha256),
        parsed.descriptor.program.ops.len(),
        parsed.descriptor.objects.len(),
        store_dir.display()
    );
    Ok(())
}

fn cmd_verify(input: &Path, limits: Limits) -> Result<()> {
    let encoded = fs::read(input)?;
    let report = materialize::verify(&encoded, limits)?;
    println!(
        "{{\"ok\":true,\"source_len\":{},\"sha256\":\"{}\",\"objects\":{},\"graph_ops\":{}}}",
        report.source_len, report.sha256_hex, report.object_count, report.graph_ops
    );
    Ok(())
}

/// `store put | account | gc` argument dispatch (Phase 9.2).
///
/// Every `ROOT...` is a `ROOT.voldoc` descriptor. `put` writes the store-backed
/// descriptor to `STORE_DIR/<input-stem>.voldoc`; `account` and `gc` read their
/// roots in place.
#[cfg(feature = "store")]
fn cmd_store_args(args: &[String], limits: Limits) -> Result<()> {
    match args.get(2).map(String::as_str) {
        Some("put") => {
            let input = arg(args, 3, "INPUT.voldoc")?;
            let store_dir = arg(args, 4, "STORE_DIR")?;
            if let Some(extra) = args.get(5) {
                return Err(Error::usage(format!("unexpected extra argument {extra:?}")));
            }
            cmd_store_put(&input, &store_dir, limits)
        }
        Some("account") => {
            let store_dir = arg(args, 3, "STORE_DIR")?;
            let roots = root_args(args, 4)?;
            cmd_store_account(&store_dir, &roots, limits)
        }
        Some("gc") => {
            let store_dir = arg(args, 3, "STORE_DIR")?;
            let roots = root_args(args, 4)?;
            cmd_store_gc(&store_dir, &roots, limits)
        }
        Some(other) => Err(Error::usage(format!(
            "unknown store subcommand {other:?}; expected put | account | gc"
        ))),
        None => Err(Error::usage(
            "store requires a subcommand: put | account | gc",
        )),
    }
}

#[cfg(feature = "store")]
fn root_args(args: &[String], start: usize) -> Result<Vec<PathBuf>> {
    let roots: Vec<PathBuf> = args
        .get(start..)
        .unwrap_or(&[])
        .iter()
        .map(PathBuf::from)
        .collect();
    if roots.is_empty() {
        return Err(Error::usage("expected at least one ROOT.voldoc"));
    }
    Ok(roots)
}

/// Externalize every object of `input` into `store_dir` and write the
/// store-backed descriptor to `STORE_DIR/<input-stem>.voldoc`.
///
/// An input that is already store-backed resolves its references through the
/// destination store before re-putting every object, so the operation is
/// idempotent and every object ends up in `store_dir` (a reference the store
/// cannot satisfy fails closed).
#[cfg(feature = "store")]
fn cmd_store_put(input: &Path, store_dir: &Path, limits: Limits) -> Result<()> {
    let encoded = fs::read(input)?;
    let mut descriptor = Descriptor::parse(&encoded, limits)?.descriptor;
    let stem = input
        .file_stem()
        .and_then(|s| s.to_str())
        .ok_or_else(|| Error::usage(format!("input {input:?} has no usable file stem")))?;
    let output = store_dir.join(format!("{stem}.voldoc"));

    let mut store = EmbeddedStore::open(store_dir)?;
    let resolver = store.clone();
    externalize(&mut descriptor, &resolver, &mut store)?;
    let (bytes, _cost) = descriptor.serialize()?;
    write_atomic(&output, &bytes)?;
    let stats = store.stats()?;
    println!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"output\":\"{}\",",
            "\"objects\":{},",
            "\"stored_objects\":{},",
            "\"stored_bytes\":{},",
            "\"root_bytes\":{}",
            "}}"
        ),
        output.display(),
        descriptor.objects.len(),
        stats.object_count,
        stats.stored_bytes,
        bytes.len()
    );
    Ok(())
}

/// Print the three accounting universes (standalone / unique-reachable /
/// amortized) over the roots, plus the per-root amortized split.
#[cfg(feature = "store")]
fn cmd_store_account(store_dir: &Path, roots: &[PathBuf], limits: Limits) -> Result<()> {
    let store = EmbeddedStore::open(store_dir)?;
    let mut descriptors = Vec::with_capacity(roots.len());
    for path in roots {
        let encoded = fs::read(path)?;
        descriptors.push(Descriptor::parse(&encoded, limits)?.descriptor);
    }
    let report = account(&descriptors, &store)?;
    let stats = store.stats()?;
    let per_root: Vec<String> = report
        .roots
        .iter()
        .zip(roots)
        .map(|(r, path)| {
            format!(
                concat!(
                    "{{",
                    "\"root\":\"{}\",",
                    "\"root_bytes\":{},",
                    "\"standalone_bytes\":{},",
                    "\"reachable_objects\":{},",
                    "\"amortized_bytes\":{}",
                    "}}"
                ),
                path.display(),
                r.root_bytes,
                r.standalone_bytes,
                r.reachable_objects,
                r.amortized_bytes
            )
        })
        .collect();
    println!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"roots\":{},",
            "\"standalone_bytes\":{},",
            "\"unique_reachable_bytes\":{},",
            "\"amortized_bytes\":{},",
            "\"unique_objects\":{},",
            "\"unique_object_bytes\":{},",
            "\"stored_bytes\":{},",
            "\"dangling\":{},",
            "\"per_root\":[{}]",
            "}}"
        ),
        report.roots.len(),
        report.standalone_bytes,
        report.unique_reachable_bytes,
        report.amortized_bytes,
        report.unique_objects,
        report.unique_object_bytes,
        stats.stored_bytes,
        report.dangling.len(),
        per_root.join(","),
    );
    Ok(())
}

/// Run mark-and-sweep GC over the roots and print the report. Every `ROOT` must
/// be a store-backed descriptor (one with external references); a standalone
/// root would otherwise mark nothing and let GC sweep the whole store.
#[cfg(feature = "store")]
fn cmd_store_gc(store_dir: &Path, roots: &[PathBuf], limits: Limits) -> Result<()> {
    let store = EmbeddedStore::open(store_dir)?;
    let mut descriptors = Vec::with_capacity(roots.len());
    for path in roots {
        let encoded = fs::read(path)?;
        descriptors.push(Descriptor::parse(&encoded, limits)?.descriptor);
    }
    for (path, d) in roots.iter().zip(&descriptors) {
        if !d
            .objects
            .iter()
            .any(|o| matches!(o, vole_document::container::ObjectSource::External { .. }))
        {
            return Err(Error::usage(format!(
                "ROOT {} is not store-backed (no EXTERNAL_REF); refusing to GC",
                path.display()
            )));
        }
    }
    let report = gc(&descriptors, &store)?;
    let dangling: Vec<String> = report
        .dangling
        .iter()
        .map(|id| format!("\"{}\"", id.to_hex()))
        .collect();
    println!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"reachable\":{},",
            "\"swept\":{},",
            "\"bytes_reclaimed\":{},",
            "\"dangling\":{},",
            "\"dangling_ids\":[{}]",
            "}}"
        ),
        report.reachable,
        report.swept,
        report.bytes_reclaimed,
        report.dangling.len(),
        dangling.join(",")
    );
    Ok(())
}

fn cmd_inspect(input: &Path, limits: Limits) -> Result<()> {
    let encoded = fs::read(input)?;
    let parsed = vole_document::container::Descriptor::parse(&encoded, limits)?;
    let d = &parsed.descriptor;
    let ops: Vec<String> = d.program.ops.iter().map(describe_op).collect();
    let ops_json = ops
        .iter()
        .map(|s| format!("\"{s}\""))
        .collect::<Vec<_>>()
        .join(",");
    let object_lens: Vec<String> = d.objects.iter().map(|o| o.len().to_string()).collect();
    println!(
        concat!(
            "{{",
            "\"universe\":\"{}\",",
            "\"universe_id\":\"{}\",",
            "\"format_version\":\"{}.{}\",",
            "\"exactness_profile\":\"EXACT_BYTES\",",
            "\"source_format\":{},",
            "\"format_basis\":\"{}\",",
            "\"source_len\":{},",
            "\"source_sha256\":\"{}\",",
            "\"object_count\":{},",
            "\"object_lens\":[{}],",
            "\"graph_ops\":{},",
            "\"program\":[{}],",
            "\"encoded_len\":{},",
            "\"cost\":{}",
            "}}"
        ),
        d.universe,
        integrity::to_hex(&parsed.universe_id),
        vole_document::container::header::FORMAT_MAJOR,
        vole_document::container::header::FORMAT_MINOR,
        d.source_format,
        d.format_basis,
        d.source_len,
        integrity::to_hex(&d.source_sha256),
        d.objects.len(),
        object_lens.join(","),
        d.program.ops.len(),
        ops_json,
        encoded.len(),
        parsed.cost.to_json(),
    );
    Ok(())
}

/// Machine-readable structural view of a PDF input for the differential oracle.
///
/// Emits one JSON line. `is_pdf` mirrors [`pdf::detect`]: a `%PDF-` header, at
/// least one indirect object, and at least one `%%EOF`. Non-PDFs are reported
/// with `"is_pdf":false` and exit 0; a real scan failure (I/O, resource limit,
/// coverage) is returned as a typed error with a nonzero exit code.
fn cmd_pdf_inspect(input: &Path, limits: Limits) -> Result<()> {
    let bytes = fs::read(input)?;
    let is_pdf = pdf::detect(&bytes, limits);
    let physical = pdf::scan(&bytes, limits)?;

    let objects: Vec<String> = physical
        .objects
        .iter()
        .map(|o| {
            format!(
                "{{\"number\":{},\"generation\":{},\"role\":\"{}\"}}",
                o.number,
                o.generation,
                role_name(o.role)
            )
        })
        .collect();
    let startxref: Vec<String> = physical.startxref.iter().map(u64::to_string).collect();

    println!(
        concat!(
            "{{",
            "\"file\":\"{}\",",
            "\"is_pdf\":{},",
            "\"span_count\":{},",
            "\"object_count\":{},",
            "\"objects\":[{}],",
            "\"revision_count\":{},",
            "\"startxref\":[{}],",
            "\"eof_count\":{}",
            "}}"
        ),
        json_escape(&input.display().to_string()),
        is_pdf,
        physical.spans.len(),
        physical.objects.len(),
        objects.join(","),
        physical.revisions.len(),
        startxref.join(","),
        physical.eofs.len(),
    );
    Ok(())
}

/// Write the deterministic Phase-3 sample corpus to `dir`, one file per entry.
///
/// Every sample is assembled with correct `/Length` and `startxref` by
/// construction; see [`pdf::samples`]. Writing is atomic per file so a failed
/// run cannot leave a partially written sample in place.
fn cmd_pdf_make_samples(dir: &Path) -> Result<()> {
    fs::create_dir_all(dir)?;
    let samples = pdf::samples::sample_pdfs();
    let mut names: Vec<String> = Vec::with_capacity(samples.len());
    for (name, bytes) in &samples {
        write_atomic(&dir.join(name), bytes)?;
        names.push(format!("\"{}\"", json_escape(name)));
    }
    println!(
        "{{\"ok\":true,\"dir\":\"{}\",\"count\":{},\"samples\":[{}]}}",
        json_escape(&dir.display().to_string()),
        names.len(),
        names.join(",")
    );
    Ok(())
}

/// Number of `FlateDecode` streams (and pages) the large generator emits by
/// default.
const LARGE_DEFAULT_OBJECTS: u64 = 800;

/// Source-corpus size floor for the partial-materialization query court
/// (contract §5.1, ≥ 32 MiB). The per-stream plaintext length is scaled so the
/// generated PDF reaches this regardless of the object count.
const LARGE_TARGET_BYTES: u64 = 32 * 1024 * 1024;

/// Upper bound on the requested object count, so a hostile argument cannot make
/// the generator allocate without bound.
const LARGE_MAX_OBJECTS: u64 = 100_000;

/// Write a large, deterministic, multi-object classic-xref PDF to `DIR/large.pdf`.
///
/// `OBJECTS` (default 800) is the number of pages; each page carries its own
/// `/FlateDecode` content stream, so there are `OBJECTS` lone-`FlateDecode`
/// streams and `2*OBJECTS + 3` indirect objects (catalog, pages, font, and one
/// page + one content object per page). Per-stream plaintext is generated from a
/// stream-indexed LCG so every stream's bytes are *distinct* (no accidental
/// cross-stream sharing), and each stream is a real, self-contained zlib
/// (RFC 1950) stream whose `/Length` and every xref offset are correct by
/// construction. The per-stream plaintext length is scaled so the total source is
/// at least 32 MiB.
///
/// This subcommand is encode-time corpus tooling: it lives in the binary, adds no
/// library or runtime dependency, and its output is regenerable and gitignored.
fn cmd_pdf_make_large(dir: &Path, objects_arg: Option<&str>) -> Result<()> {
    let objects: u64 = match objects_arg {
        Some(s) => s
            .parse()
            .map_err(|_| Error::usage(format!("OBJECTS {s:?} is not a non-negative integer")))?,
        None => LARGE_DEFAULT_OBJECTS,
    };
    if objects == 0 || objects > LARGE_MAX_OBJECTS {
        return Err(Error::usage(format!(
            "OBJECTS must be between 1 and {LARGE_MAX_OBJECTS}"
        )));
    }
    let bytes = pdf::large_pdf(objects, LARGE_TARGET_BYTES);
    fs::create_dir_all(dir)?;
    let name = "large.pdf";
    write_atomic(&dir.join(name), &bytes)?;
    let indirect_objects = 2 * objects + 3;
    println!(
        "{{\"ok\":true,\"dir\":\"{}\",\"file\":\"{}\",\"objects\":{},\"streams\":{},\"indirect_objects\":{},\"source_len\":{},\"sha256\":\"{}\"}}",
        json_escape(&dir.display().to_string()),
        name,
        objects,
        objects,
        indirect_objects,
        bytes.len(),
        integrity::to_hex(&integrity::sha256(&bytes)),
    );
    Ok(())
}

/// Machine-readable exact-DEFLATE-replay correction-ratio view of each input.
///
/// Emits one JSON line per input: per-`FlateDecode`-stream `compressed_bytes`,
/// `plaintext_bytes`, `correction_bytes`, `rans_plaintext_bytes`, the ratios
/// `correction/compressed`, `(plaintext+correction)/compressed`, and
/// `(rans_plaintext+correction)/compressed`, plus the aggregate. A non-PDF is
/// reported with `"is_pdf":false` and no streams (exit 0). Ratios are rendered
/// as fixed-point decimals computed with integer arithmetic only; they are
/// diagnostics and never a persisted representation.
///
/// Two aggregate complete-cost figures are reported so shared plaintext is not
/// overcounted: `replayed_rans_full_bytes` is the **naive per-stream** sum
/// (each stream pays its own rANS plaintext cost plus its own correction), while
/// `replayed_rans_dedup_bytes` charges each **unique** plaintext (rANS) and each
/// **unique** correction blob once, mirroring what the shared-channel
/// `PDF_DEFLATE_REPLAY_RANS` candidate actually stores. The deduped figure is
/// therefore ≤ the naive one, with equality exactly when no plaintext or
/// correction repeats.
#[cfg(feature = "deflate-replay")]
fn cmd_deflate_stats(inputs: &[PathBuf], limits: Limits) -> Result<()> {
    for input in inputs {
        let bytes = fs::read(input)?;
        let stats = pdf::deflate_stats(&bytes, limits)?;
        println!("{}", deflate_stats_json(input, &stats));
    }
    Ok(())
}

#[cfg(feature = "deflate-replay")]
fn deflate_stats_json(input: &Path, stats: &pdf::DeflateStats) -> String {
    let streams: Vec<String> = stats.streams.iter().map(stream_stats_json).collect();
    let s = &stats.summary;
    let replayed_full = s.plaintext_bytes.saturating_add(s.correction_bytes);
    let replayed_rans_full = s.rans_plaintext_bytes.saturating_add(s.correction_bytes);
    let replayed_rans_dedup = s.replayed_rans_dedup_bytes;
    format!(
        concat!(
            "{{",
            "\"file\":\"{}\",",
            "\"is_pdf\":{},",
            "\"streams\":[{}],",
            "\"summary\":{{",
            "\"flate_streams\":{},",
            "\"replayed\":{},",
            "\"declined\":{},",
            "\"compressed_bytes\":{},",
            "\"plaintext_bytes\":{},",
            "\"correction_bytes\":{},",
            "\"rans_plaintext_bytes\":{},",
            "\"correction_over_compressed\":{},",
            "\"replayed_full_bytes\":{},",
            "\"replayed_rans_full_bytes\":{},",
            "\"replayed_rans_dedup_bytes\":{}",
            "}}",
            "}}"
        ),
        json_escape(&input.display().to_string()),
        stats.is_pdf,
        streams.join(","),
        s.flate_streams,
        s.replayed,
        s.declined,
        s.compressed_bytes,
        s.plaintext_bytes,
        s.correction_bytes,
        s.rans_plaintext_bytes,
        ratio6(s.correction_bytes, s.compressed_bytes),
        replayed_full,
        replayed_rans_full,
        replayed_rans_dedup,
    )
}

#[cfg(feature = "deflate-replay")]
fn stream_stats_json(s: &pdf::StreamStats) -> String {
    let reason = match s.decline_reason {
        Some(r) => format!("\"{r}\""),
        None => "null".to_string(),
    };
    let raw_ratio = match s.correction_bytes {
        Some(c) => ratio6(c, s.compressed_bytes),
        None => "null".to_string(),
    };
    let plain_ratio = match (s.plaintext_bytes, s.correction_bytes) {
        (Some(p), Some(c)) => ratio6(p.saturating_add(c), s.compressed_bytes),
        _ => "null".to_string(),
    };
    let rans_ratio = match (s.rans_plaintext_bytes, s.correction_bytes) {
        (Some(r), Some(c)) => ratio6(r.saturating_add(c), s.compressed_bytes),
        _ => "null".to_string(),
    };
    format!(
        concat!(
            "{{",
            "\"object\":{},",
            "\"generation\":{},",
            "\"compressed_bytes\":{},",
            "\"replayed\":{},",
            "\"decline_reason\":{},",
            "\"plaintext_bytes\":{},",
            "\"correction_bytes\":{},",
            "\"rans_plaintext_bytes\":{},",
            "\"raw_ratio\":{},",
            "\"plain_ratio\":{},",
            "\"rans_ratio\":{}",
            "}}"
        ),
        s.object,
        s.generation,
        s.compressed_bytes,
        s.replayed,
        reason,
        opt_u64(s.plaintext_bytes),
        opt_u64(s.correction_bytes),
        opt_u64(s.rans_plaintext_bytes),
        raw_ratio,
        plain_ratio,
        rans_ratio,
    )
}

#[cfg(feature = "deflate-replay")]
fn opt_u64(v: Option<u64>) -> String {
    match v {
        Some(n) => n.to_string(),
        None => "null".to_string(),
    }
}

/// Fixed-point ratio with six decimals, computed with integer arithmetic only
/// (truncating division). Diagnostics only; never decides a persisted value.
#[cfg(feature = "deflate-replay")]
fn ratio6(num: u64, den: u64) -> String {
    if den == 0 {
        return "0.000000".to_string();
    }
    let scaled = u128::from(num) * 1_000_000;
    let q = scaled / u128::from(den);
    format!("{}.{:06}", q / 1_000_000, q % 1_000_000)
}

/// Stable string name for an object role, as used in the oracle JSON.
fn role_name(role: pdf::ObjRole) -> &'static str {
    match role {
        pdf::ObjRole::Generic => "Generic",
        pdf::ObjRole::XRefStream => "XRefStream",
        pdf::ObjRole::ObjectStream => "ObjectStream",
    }
}

/// Minimal JSON string escaping for the `file` field (paths may contain quotes).
fn json_escape(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        match c {
            '"' => out.push_str("\\\""),
            '\\' => out.push_str("\\\\"),
            '\n' => out.push_str("\\n"),
            '\r' => out.push_str("\\r"),
            '\t' => out.push_str("\\t"),
            c if (c as u32) < 0x20 => out.push_str(&format!("\\u{:04x}", c as u32)),
            c => out.push(c),
        }
    }
    out
}

fn cmd_capabilities() -> Result<()> {
    println!(
        concat!(
            "{{",
            "\"crate\":\"vole-document\",",
            "\"format_major\":{},",
            "\"format_minor\":{},",
            "\"dra_version\":{},",
            "\"exactness_profiles\":[\"EXACT_BYTES\"],",
            "\"source_formats\":[\"OPAQUE\",\"PDF\"],",
            "\"universe\":\"{}\",",
            "\"dra_ops\":[\"EMIT_OBJECT\",\"INLINE\",\"REPEAT_LAST\",\"DECODE_CHANNEL\",\"INTERLEAVE_CHANNELS\",\"MARK_OFFSET\",\"EMIT_OFFSET\",\"PACK_SEGMENTS\",\"PACKED_CHANNELS\",\"DEFLATE_REPLAY\"],",
            "\"entropy_channels\":[\"ORDER0_BYTE_RANS\"],",
            "\"features\":{{\"rans\":{},\"deflate_replay\":{}}}",
            "}}"
        ),
        vole_document::container::header::FORMAT_MAJOR,
        vole_document::container::header::FORMAT_MINOR,
        vole_document::dra::program::DRA_VERSION,
        UNIVERSE,
        cfg!(feature = "rans"),
        cfg!(feature = "deflate-replay"),
    );
    Ok(())
}

fn describe_op(op: &Op) -> String {
    match op {
        Op::EmitObject { object_id } => format!("EMIT_OBJECT({object_id})"),
        Op::Inline { bytes } => format!("INLINE({})", bytes.len()),
        Op::RepeatLast { count } => format!("REPEAT_LAST({count})"),
        Op::DecodeChannel { channel_id } => format!("DECODE_CHANNEL({channel_id})"),
        Op::InterleaveChannels {
            first_payload_channel,
            payload_channel_count,
            ..
        } => format!("INTERLEAVE_CHANNELS({first_payload_channel},{payload_channel_count})"),
        Op::MarkOffset { slot } => format!("MARK_OFFSET({slot})"),
        Op::EmitOffset { slot, width } => format!("EMIT_OFFSET({slot},{width})"),
        Op::PackSegments { data_object, items } => {
            format!("PACK_SEGMENTS({data_object},{})", items.len())
        }
        Op::PackedChannels {
            data_channel,
            plan_channel,
            declared_output_len,
        } => format!("PACKED_CHANNELS({data_channel},{plan_channel},{declared_output_len})"),
        Op::DeflateReplay {
            replay_codec,
            source_kind,
            source_id,
            corrections_object,
            declared_output_len,
        } => format!(
            "DEFLATE_REPLAY({replay_codec},{source_kind},{source_id},{corrections_object},{declared_output_len})"
        ),
    }
}

fn report_json(r: &encode::EncodeReport) -> String {
    format!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"candidate\":\"{}\",",
            "\"candidates_evaluated\":{},",
            "\"source_len\":{},",
            "\"encoded_len\":{},",
            "\"ratio\":{:.6},",
            "\"sha256\":\"{}\",",
            "\"graph_ops\":{},",
            "\"cost\":{}",
            "}}"
        ),
        r.kind.name(),
        r.candidates_evaluated,
        r.source_len,
        r.encoded_len,
        r.compression_ratio(),
        r.sha256_hex,
        r.graph_ops,
        r.cost.to_json(),
    )
}

/// Parsed arguments for the Phase-11 field observation verbs.
#[cfg(feature = "field")]
#[derive(Default)]
struct FieldArgs {
    store: Option<PathBuf>,
    field: Option<String>,
    page: Option<u32>,
    object: Option<u32>,
    stream: Option<u32>,
    revision: Option<u32>,
    byte_range: Option<(u64, u64)>,
    kind: Option<String>,
    text: Option<String>,
    output: Option<PathBuf>,
    content: Option<PathBuf>,
    analyze: bool,
    json: bool,
    no_cache: bool,
    entropyfs: bool,
    positional: Vec<String>,
}

#[cfg(feature = "field")]
fn field_arg_value(
    args: &[String],
    i: &mut usize,
    flag: &str,
    inline: Option<&str>,
) -> Result<String> {
    match inline {
        Some(v) => {
            *i += 1;
            Ok(v.to_string())
        }
        None => {
            let v = args
                .get(*i + 1)
                .ok_or_else(|| Error::usage(format!("{flag} requires a value")))?;
            *i += 2;
            Ok(v.clone())
        }
    }
}

#[cfg(feature = "field")]
fn parse_field_args(args: &[String]) -> Result<FieldArgs> {
    let mut out = FieldArgs::default();
    let mut i = 2;
    while i < args.len() {
        let a = args[i].as_str();
        let (flag, inline) = match a.split_once('=') {
            Some((f, v)) => (f, Some(v)),
            None => (a, None),
        };
        match flag {
            "--analyze" => {
                out.analyze = true;
                i += 1;
            }
            "--json" => {
                out.json = true;
                i += 1;
            }
            "--exact" => i += 1,
            "--no-cache" => {
                out.no_cache = true;
                i += 1;
            }
            "--entropyfs" => {
                out.entropyfs = true;
                i += 1;
            }
            "--store" => {
                out.store = Some(PathBuf::from(field_arg_value(
                    args, &mut i, "--store", inline,
                )?));
            }
            "--field" => out.field = Some(field_arg_value(args, &mut i, "--field", inline)?),
            "--page" => {
                out.page = Some(parse_field_u32(
                    &field_arg_value(args, &mut i, "--page", inline)?,
                    "--page",
                )?);
            }
            "--object" => {
                out.object = Some(parse_field_u32(
                    &field_arg_value(args, &mut i, "--object", inline)?,
                    "--object",
                )?);
            }
            "--stream" => {
                out.stream = Some(parse_field_u32(
                    &field_arg_value(args, &mut i, "--stream", inline)?,
                    "--stream",
                )?);
            }
            "--revision" => {
                out.revision = Some(parse_field_u32(
                    &field_arg_value(args, &mut i, "--revision", inline)?,
                    "--revision",
                )?);
            }
            "--byte-range" => {
                out.byte_range = Some(parse_field_range(&field_arg_value(
                    args,
                    &mut i,
                    "--byte-range",
                    inline,
                )?)?);
            }
            "--kind" => out.kind = Some(field_arg_value(args, &mut i, "--kind", inline)?),
            "--text" => out.text = Some(field_arg_value(args, &mut i, "--text", inline)?),
            "--output" => {
                out.output = Some(PathBuf::from(field_arg_value(
                    args, &mut i, "--output", inline,
                )?));
            }
            "--content" => {
                out.content = Some(PathBuf::from(field_arg_value(
                    args,
                    &mut i,
                    "--content",
                    inline,
                )?));
            }
            other if !other.starts_with("--") => {
                out.positional.push(other.to_string());
                i += 1;
            }
            other => return Err(Error::usage(format!("unknown field argument {other:?}"))),
        }
    }
    Ok(out)
}

#[cfg(feature = "field")]
fn parse_field_u32(value: &str, flag: &str) -> Result<u32> {
    value
        .parse()
        .map_err(|_| Error::usage(format!("{flag} value {value:?} is not a u32")))
}

#[cfg(feature = "field")]
fn parse_field_range(value: &str) -> Result<(u64, u64)> {
    let (a, b) = value
        .split_once("..")
        .ok_or_else(|| Error::usage("--byte-range must be A..B (e.g. 0..128)"))?;
    let start: u64 = a
        .parse()
        .map_err(|_| Error::usage(format!("--byte-range start {a:?} is not a u64")))?;
    let end: u64 = b
        .parse()
        .map_err(|_| Error::usage(format!("--byte-range end {b:?} is not a u64")))?;
    if end < start {
        return Err(Error::usage("--byte-range end precedes its start"));
    }
    Ok((start, end - start))
}

#[cfg(feature = "field")]
fn field_selector(out: &FieldArgs) -> Result<Selector> {
    let mut chosen: Vec<Selector> = Vec::new();
    if let Some(n) = out.page {
        chosen.push(Selector::Page(n));
    }
    if let Some(n) = out.object {
        chosen.push(Selector::Object(n));
    }
    if let Some(n) = out.stream {
        chosen.push(Selector::Stream(n));
    }
    if let Some(n) = out.revision {
        chosen.push(Selector::Revision(n));
    }
    if let Some((offset, len)) = out.byte_range {
        chosen.push(Selector::ByteRange { offset, len });
    }
    if let Some(text) = &out.text {
        chosen.push(Selector::TextMatch(text.clone()));
    }
    match chosen.len() {
        0 => Err(Error::usage("exactly one selector flag is required")),
        1 => chosen
            .into_iter()
            .next()
            .ok_or_else(|| Error::usage("no selector")),
        _ => Err(Error::usage(
            "exactly one selector flag is required; more than one was given",
        )),
    }
}

#[cfg(feature = "field")]
fn field_representation(kind: &str) -> Result<Representation> {
    Ok(match kind {
        "metadata" => Representation::Metadata,
        "text" => Representation::Text,
        "structure" => Representation::Structure,
        "operators" => Representation::Operators,
        "encoded" => Representation::EncodedBytes,
        "decoded" => Representation::DecodedBytes,
        "exact" => Representation::ExactBytes,
        "preview" => Representation::Preview,
        "full" => Representation::FullDocument,
        other => return Err(Error::usage(format!("unknown --kind {other:?}"))),
    })
}

#[cfg(feature = "field")]
fn observe_request(
    out: &FieldArgs,
    selector: Selector,
    representation: Representation,
) -> ObserveRequest {
    let mut req = ObserveRequest::new(selector, representation);
    req.use_cache = !out.no_cache;
    req
}

#[cfg(feature = "field")]
fn field_answer_json(answer: &FieldAnswer, stats: &ObserveStats, field: &FieldId) -> String {
    let value = match &answer.value {
        AnswerValue::Bytes(b) => {
            let sha = integrity::sha256(b);
            let hex_part = if b.len() <= 8192 {
                format!(",\"value_hex\":\"{}\"", integrity::to_hex(b))
            } else {
                String::new()
            };
            format!(
                "\"bytes_len\":{},\"bytes_sha256\":\"{}\"{}",
                b.len(),
                integrity::to_hex(&sha),
                hex_part
            )
        }
        AnswerValue::Text(t) => format!("\"text\":\"{}\"", json_escape(t)),
        AnswerValue::Json(j) => format!("\"value\":{j}"),
        AnswerValue::None => "\"value\":null".to_string(),
    };
    let span = match answer.source_span {
        Some((a, b)) => format!("[{a},{b}]"),
        None => "null".to_string(),
    };
    let deps = answer
        .dependency_ids
        .iter()
        .map(|d| format!("\"{}\"", d.to_hex()))
        .collect::<Vec<_>>()
        .join(",");
    format!(
        concat!(
            "{{",
            "\"field\":\"{}\",",
            "\"selector\":\"{}\",",
            "\"representation\":\"{}\",",
            "\"basis\":\"{}\",",
            "\"exact\":{},",
            "\"integrity_scope\":\"{}\",",
            "\"source_span\":{},",
            "\"dependency_ids\":[{}],",
            "{},",
            "\"stats\":{{\"index_nodes_read\":{},\"seed_nodes_fetched\":{},\"seed_nodes_materialized\":{},\"seed_nodes_executed\":{},\"seed_nodes_reused\":{},\"cache_bytes_written\":{},\"descriptor_bytes_read\":{},\"descriptor_read_mode\":\"{}\",\"manifest_bytes_read\":{},\"index_bytes_read\":{},\"seed_bytes_read\":{},\"bytes_read\":{},\"bytes_returned\":{},\"deepened\":{},\"wall_micros\":{}}}",
            "}}"
        ),
        field.to_hex(),
        json_escape(&answer.selector),
        json_escape(&answer.representation),
        answer.basis.name(),
        answer.exact,
        answer.integrity_scope.name(),
        span,
        deps,
        value,
        stats.index_nodes_read,
        stats.seed_nodes_fetched,
        stats.seed_nodes_materialized,
        stats.seed_nodes_executed,
        stats.seed_nodes_reused,
        stats.cache_bytes_written,
        stats.descriptor_bytes_read,
        stats.descriptor_read_mode.name(),
        stats.manifest_bytes_read,
        stats.index_bytes_read,
        stats.seed_bytes_read,
        stats.bytes_read,
        stats.bytes_returned,
        stats.deepened,
        stats.wall_micros,
    )
}

/// Open the field store selected by `--entropyfs` (or the filesystem default).
///
/// The engine-backed backend requires a build with the `entropyfs-store`
/// feature; asking for it without that feature is a typed `UnsupportedFeature`,
/// never a silent fallback to the filesystem backend.
#[cfg(feature = "field")]
fn open_field_store(store_dir: &Path, entropyfs: bool) -> Result<FieldStore> {
    #[cfg(feature = "entropyfs-store")]
    if entropyfs {
        return FieldStore::open_entropyfs(store_dir);
    }
    if entropyfs {
        return Err(Error::unsupported_feature(
            "--entropyfs requires a build with the entropyfs-store feature",
        ));
    }
    FieldStore::open(store_dir)
}

/// `field-store-stats --store DIR [--entropyfs]`: report advisory engine
/// accounting (blob count and bytes) for an EntropyFS-backed field store, so a
/// court can witness that the seed DAG is many individual engine blobs.
#[cfg(feature = "field")]
fn cmd_field_store_stats(args: &[String]) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("field-store-stats requires --store DIR"))?;
    let store = open_field_store(store_dir, out.entropyfs)?;
    #[cfg(feature = "entropyfs-store")]
    {
        match store.engine_stats()? {
            Some(s) => println!(
                "{{\"backend\":\"entropyfs\",\"blob_count\":{},\"logical_bytes\":{},\"physical_used_bytes\":{}}}",
                s.blob_count, s.logical_bytes, s.physical_used_bytes
            ),
            None => println!("{{\"backend\":\"fs\"}}"),
        }
    }
    #[cfg(not(feature = "entropyfs-store"))]
    {
        let _ = store;
        println!("{{\"backend\":\"fs\"}}");
    }
    Ok(())
}

#[cfg(feature = "field")]
fn cmd_field_ingest(args: &[String], limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let input = out
        .positional
        .first()
        .ok_or_else(|| Error::usage("field-ingest requires INPUT.voldoc"))?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("field-ingest requires --store DIR"))?;
    let bytes = fs::read(input)?;
    let mut store = open_field_store(store_dir, out.entropyfs)?;
    let r = field_ingest::ingest_pdf(&mut store, &bytes, limits)?;
    store.sync()?;
    let index_root = match r.index_root {
        Some(id) => format!("\"{}\"", id.to_hex()),
        None => "null".to_string(),
    };
    println!(
        concat!(
            "{{",
            "\"field\":\"{}\",",
            "\"root_node\":\"{}\",",
            "\"index_root\":{},",
            "\"node_count\":{},",
            "\"index_node_count\":{},",
            "\"source_len\":{},",
            "\"object_nodes\":{},",
            "\"stream_nodes\":{},",
            "\"decoded_stream_nodes\":{},",
            "\"page_nodes\":{},",
            "\"revision_nodes\":{},",
            "\"declined_streams\":{}",
            "}}"
        ),
        r.field.to_hex(),
        r.root_node.to_hex(),
        index_root,
        r.node_count,
        r.index_node_count,
        r.source_len,
        r.object_nodes,
        r.stream_nodes,
        r.decoded_stream_nodes,
        r.page_nodes,
        r.revision_nodes,
        r.declined_streams,
    );
    Ok(())
}

#[cfg(feature = "field")]
fn cmd_field_edit(args: &[String], _limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("field-edit requires --store DIR"))?;
    let field_hex = out
        .field
        .as_deref()
        .ok_or_else(|| Error::usage("field-edit requires --field HEX"))?;
    let page = out
        .page
        .ok_or_else(|| Error::usage("field-edit requires --page N"))?;
    let content_path = out
        .content
        .as_deref()
        .ok_or_else(|| Error::usage("field-edit requires --content FILE"))?;
    let content = fs::read(content_path)?;
    let mut store = open_field_store(store_dir, out.entropyfs)?;
    let id = FieldId::from_hex(field_hex)?;
    let r = field_edit::replace_page_content(&mut store, &id, page, &content)?;
    store.sync()?;
    println!(
        concat!(
            "{{",
            "\"field\":\"{}\",",
            "\"previous\":\"{}\",",
            "\"page\":{},",
            "\"page_content\":\"{}\",",
            "\"content_literal\":\"{}\",",
            "\"index_root\":\"{}\",",
            "\"index_entries\":{},",
            "\"index_entries_reused\":{},",
            "\"index_entries_replaced\":{},",
            "\"seed_nodes_new\":{},",
            "\"seed_nodes_reused\":{},",
            "\"index_nodes_reused\":{},",
            "\"index_nodes_new\":{},",
            "\"bytes_newly_persisted\":{},",
            "\"descriptor_bytes_read\":{},",
            "\"manifest_bytes_read\":{},",
            "\"index_bytes_read\":{},",
            "\"seed_bytes_read\":{}",
            "}}"
        ),
        r.field.to_hex(),
        r.previous.to_hex(),
        r.page,
        r.page_content.to_hex(),
        r.content_literal.to_hex(),
        r.index_root.to_hex(),
        r.index_entries,
        r.index_entries_reused,
        r.index_entries_replaced,
        r.seed_nodes_new,
        r.seed_nodes_reused,
        r.index_nodes_reused,
        r.index_nodes_new,
        r.bytes_newly_persisted,
        r.descriptor_bytes_read,
        r.manifest_bytes_read,
        r.index_bytes_read,
        r.seed_bytes_read,
    );
    Ok(())
}

#[cfg(feature = "field")]
fn cmd_field_observe(args: &[String], limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("observe requires --store DIR"))?;
    let field_hex = out
        .field
        .as_deref()
        .ok_or_else(|| Error::usage("observe requires --field HEX"))?;
    let selector = field_selector(&out)?;
    let kind = out
        .kind
        .as_deref()
        .ok_or_else(|| Error::usage("observe requires --kind KIND"))?;
    let representation = field_representation(kind)?;
    let mut store = open_field_store(store_dir, out.entropyfs)?;
    let id = FieldId::from_hex(field_hex)?;
    let req = observe_request(&out, selector, representation);
    let (answer, stats, field) = observe(&mut store, &id, &req, limits)?;
    store.sync()?;
    println!("{}", field_answer_json(&answer, &stats, &field));
    Ok(())
}

#[cfg(feature = "field")]
fn cmd_field_find(args: &[String], limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("find requires --store DIR"))?;
    let field_hex = out
        .field
        .as_deref()
        .ok_or_else(|| Error::usage("find requires --field HEX"))?;
    let text = out
        .text
        .clone()
        .ok_or_else(|| Error::usage("find requires --text PATTERN"))?;
    let mut store = open_field_store(store_dir, out.entropyfs)?;
    let id = FieldId::from_hex(field_hex)?;
    let req = observe_request(&out, Selector::TextMatch(text), Representation::Text);
    let (answer, stats, field) = observe(&mut store, &id, &req, limits)?;
    store.sync()?;
    println!("{}", field_answer_json(&answer, &stats, &field));
    Ok(())
}

#[cfg(feature = "field")]
fn cmd_field_explain(args: &[String], limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("explain requires --store DIR"))?;
    let field_hex = out
        .field
        .as_deref()
        .ok_or_else(|| Error::usage("explain requires --field HEX"))?;
    let selector = field_selector(&out)?;
    let kind = out
        .kind
        .as_deref()
        .ok_or_else(|| Error::usage("explain requires --kind KIND"))?;
    let representation = field_representation(kind)?;
    let mut store = open_field_store(store_dir, out.entropyfs)?;
    let id = FieldId::from_hex(field_hex)?;
    let req = observe_request(&out, selector, representation);
    if out.analyze {
        let planned_json = {
            let manifest = store.get_field(&id)?;
            explain(&manifest, &store, &req)?.json
        };
        let (answer, stats, promoted) = observe(&mut store, &id, &req, limits)?;
        println!(
            "{{\"plan\":{},\"actual\":{}}}",
            planned_json,
            explain_actual_json(&stats, &answer, &promoted)
        );
    } else {
        let manifest = store.get_field(&id)?;
        let plan = explain(&manifest, &store, &req)?;
        println!("{}", plan.json);
    }
    store.sync()?;
    Ok(())
}

/// The executed-observation evidence object for `explain --analyze`, including
/// the promoted field id, the per-class physical byte counts (review fix #1),
/// and the reuse counters (11.8).
#[cfg(feature = "field")]
fn explain_actual_json(stats: &ObserveStats, answer: &FieldAnswer, field: &FieldId) -> String {
    format!(
        concat!(
            "{{",
            "\"field\":\"{}\",",
            "\"index_nodes_read\":{},",
            "\"seed_nodes_fetched\":{},",
            "\"seed_nodes_materialized\":{},",
            "\"seed_nodes_executed\":{},",
            "\"seed_nodes_reused\":{},",
            "\"cache_bytes_written\":{},",
            "\"descriptor_bytes_read\":{},",
            "\"descriptor_read_mode\":\"{}\",",
            "\"manifest_bytes_read\":{},",
            "\"index_bytes_read\":{},",
            "\"seed_bytes_read\":{},",
            "\"bytes_read\":{},",
            "\"bytes_returned\":{},",
            "\"deepened\":{},",
            "\"wall_micros\":{},",
            "\"basis\":\"{}\",",
            "\"exact\":{}",
            "}}"
        ),
        field.to_hex(),
        stats.index_nodes_read,
        stats.seed_nodes_fetched,
        stats.seed_nodes_materialized,
        stats.seed_nodes_executed,
        stats.seed_nodes_reused,
        stats.cache_bytes_written,
        stats.descriptor_bytes_read,
        stats.descriptor_read_mode.name(),
        stats.manifest_bytes_read,
        stats.index_bytes_read,
        stats.seed_bytes_read,
        stats.bytes_read,
        stats.bytes_returned,
        stats.deepened,
        stats.wall_micros,
        answer.basis.name(),
        answer.exact,
    )
}

#[cfg(feature = "field")]
fn cmd_field_preview(args: &[String], limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("preview requires --store DIR"))?;
    let field_hex = out
        .field
        .as_deref()
        .ok_or_else(|| Error::usage("preview requires --field HEX"))?;
    let page = out
        .page
        .ok_or_else(|| Error::usage("preview requires --page N"))?;
    let as_json = out.json;
    let mut store = open_field_store(store_dir, out.entropyfs)?;
    let id = FieldId::from_hex(field_hex)?;
    let req = observe_request(&out, Selector::Page(page), Representation::Preview);
    let (answer, stats, field) = observe(&mut store, &id, &req, limits)?;
    store.sync()?;
    if !as_json && let AnswerValue::Bytes(bytes) = &answer.value {
        std::io::stdout().write_all(bytes).map_err(Error::from)?;
        return Ok(());
    }
    println!("{}", field_answer_json(&answer, &stats, &field));
    Ok(())
}

/// `share-account --store DIR INPUT.voldoc...`: store every inline fine unit of
/// the cohort in `<DIR>/share` and print the [`share::ShareReport`] as JSON
/// (Phase 11.14).
///
/// The store is populated so the fine-unit sharing rests on **stored bytes**, not
/// a paper calculation: because the store is content-addressed, `store_bytes`
/// (distinct stored blobs) must equal `unique_bytes`, an independent check on the
/// in-memory report. The descriptor still carries every unit inline; `unique_bytes`
/// excludes all record/root framing and is a lower bound on any store form.
/// Deterministic for a fixed input set (content addressing + sorted `by_kind`).
#[cfg(feature = "field")]
fn cmd_share_account(args: &[String], limits: Limits) -> Result<()> {
    let mut store_dir: Option<PathBuf> = None;
    let mut inputs: Vec<PathBuf> = Vec::new();
    let mut i = 2;
    while i < args.len() {
        let a = args[i].as_str();
        let (flag, inline) = match a.split_once('=') {
            Some((f, v)) => (f, Some(v)),
            None => (a, None),
        };
        match flag {
            "--store" => {
                store_dir = Some(PathBuf::from(field_arg_value(
                    args, &mut i, "--store", inline,
                )?));
            }
            other if other.starts_with("--") => {
                return Err(Error::usage(format!(
                    "unknown share-account argument {other:?}"
                )));
            }
            other => {
                inputs.push(PathBuf::from(other));
                i += 1;
            }
        }
    }
    let store_dir = store_dir.ok_or_else(|| Error::usage("share-account requires --store DIR"))?;
    if inputs.is_empty() {
        return Err(Error::usage(
            "share-account requires at least one INPUT.voldoc",
        ));
    }
    let mut descriptors = Vec::with_capacity(inputs.len());
    for path in &inputs {
        let bytes = fs::read(path)?;
        descriptors.push(Descriptor::parse(&bytes, limits)?.descriptor);
    }
    // Store every inline fine unit so the accounting rests on stored bytes.
    let mut units_offered: u64 = 0;
    for d in &descriptors {
        units_offered += share::store_units(&store_dir, d)?;
    }
    let report = share::cohort_report(&descriptors)?;
    let store_bytes = share::share_store(&store_dir)?.stats()?.stored_bytes;
    let kinds: Vec<String> = report
        .by_kind
        .iter()
        .map(|(k, t, u)| {
            format!(
                "{{\"kind\":\"{}\",\"total\":{t},\"unique\":{u}}}",
                json_escape(k)
            )
        })
        .collect();
    println!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"files\":{},",
            "\"units_offered\":{},",
            "\"store_bytes\":{},",
            "\"total_bytes\":{},",
            "\"unique_bytes\":{},",
            "\"unit_count\":{},",
            "\"unique_count\":{},",
            "\"by_kind\":[{}]",
            "}}"
        ),
        inputs.len(),
        units_offered,
        store_bytes,
        report.total_bytes,
        report.unique_bytes,
        report.unit_count,
        report.unique_count,
        kinds.join(",")
    );
    Ok(())
}

/// `share report INPUT.voldoc...`: measure fine-unit sharing over a cohort of
/// descriptors (Phase 11.14). Read-only; no store is touched.
#[cfg(feature = "field")]
fn cmd_share_args(args: &[String], limits: Limits) -> Result<()> {
    match args.get(2).map(String::as_str) {
        Some("report") => {
            let inputs: Vec<PathBuf> = args
                .get(3..)
                .unwrap_or(&[])
                .iter()
                .map(PathBuf::from)
                .collect();
            if inputs.is_empty() {
                return Err(Error::usage(
                    "share report requires at least one INPUT.voldoc",
                ));
            }
            cmd_share_report(&inputs, limits)
        }
        Some("externalize") => {
            let mut store_dir: Option<PathBuf> = None;
            let mut positional: Vec<&str> = Vec::new();
            let mut i = 3;
            while i < args.len() {
                let a = args[i].as_str();
                let (flag, inline) = match a.split_once('=') {
                    Some((f, v)) => (f, Some(v)),
                    None => (a, None),
                };
                match flag {
                    "--store" => {
                        store_dir = Some(PathBuf::from(field_arg_value(
                            args, &mut i, "--store", inline,
                        )?));
                    }
                    other => positional.push(other),
                }
            }
            let store_dir =
                store_dir.ok_or_else(|| Error::usage("share externalize requires --store DIR"))?;
            let input = positional
                .first()
                .map(PathBuf::from)
                .ok_or_else(|| Error::usage("share externalize requires INPUT.voldoc"))?;
            let output = positional
                .get(1)
                .map(PathBuf::from)
                .ok_or_else(|| Error::usage("share externalize requires OUTPUT.voldoc"))?;
            if positional.len() > 2 {
                return Err(Error::usage(format!(
                    "unexpected extra argument {:?}",
                    positional[2]
                )));
            }
            cmd_share_externalize(&store_dir, &input, &output, limits)
        }
        other => Err(Error::usage(format!(
            "unknown share subcommand {:?}",
            other.unwrap_or("")
        ))),
    }
}

/// Print fine-unit sharing for each descriptor and for the cohort as a whole.
#[cfg(feature = "field")]
fn cmd_share_report(inputs: &[PathBuf], limits: Limits) -> Result<()> {
    let mut descriptors = Vec::with_capacity(inputs.len());
    for path in inputs {
        let bytes = fs::read(path)?;
        descriptors.push(Descriptor::parse(&bytes, limits)?.descriptor);
    }
    let report = share::cohort_report(&descriptors)?;
    let per_file: Vec<String> = inputs
        .iter()
        .zip(&descriptors)
        .map(|(path, d)| {
            let r = share::cohort_report(std::slice::from_ref(d))?;
            Ok(format!(
                concat!(
                    "{{",
                    "\"file\":\"{}\",",
                    "\"total_bytes\":{},",
                    "\"unique_bytes\":{},",
                    "\"unit_count\":{},",
                    "\"unique_count\":{}",
                    "}}"
                ),
                json_escape(&path.display().to_string()),
                r.total_bytes,
                r.unique_bytes,
                r.unit_count,
                r.unique_count
            ))
        })
        .collect::<Result<Vec<_>>>()?;
    println!(
        "{{\"ok\":true,\"files\":{},\"report\":{},\"per_file\":[{}]}}",
        inputs.len(),
        report.to_json(),
        per_file.join(",")
    );
    Ok(())
}

/// Externalize the object table into `<DIR>/share` (wire-real) and store every
/// fine unit there (measured), writing the store-backed descriptor to `OUTPUT`.
#[cfg(feature = "field")]
fn cmd_share_externalize(
    store_dir: &Path,
    input: &Path,
    output: &Path,
    limits: Limits,
) -> Result<()> {
    let encoded = fs::read(input)?;
    let mut descriptor = Descriptor::parse(&encoded, limits)?.descriptor;
    let offered = share::store_units(store_dir, &descriptor)?;
    let objects = share::externalize_objects(store_dir, &mut descriptor)?;
    let (bytes, _cost) = descriptor.serialize()?;
    write_atomic(output, &bytes)?;
    let report = share::cohort_report(std::slice::from_ref(&descriptor))?;
    let share_bytes = share::share_store(store_dir)?.stats()?.stored_bytes;
    println!(
        concat!(
            "{{",
            "\"ok\":true,",
            "\"output\":\"{}\",",
            "\"objects\":{},",
            "\"units_offered\":{},",
            "\"root_bytes\":{},",
            "\"share_store_bytes\":{},",
            "\"report\":{}",
            "}}"
        ),
        json_escape(&output.display().to_string()),
        objects,
        offered,
        bytes.len(),
        share_bytes,
        report.to_json()
    );
    Ok(())
}

/// `cache --store DIR [--clear]`: report — and optionally reclaim — the
/// disposable derived-cache universe (ADR-0027). Never touches the store or the
/// descriptor.
#[cfg(feature = "field")]
fn cmd_field_cache(args: &[String]) -> Result<()> {
    let mut store_dir: Option<PathBuf> = None;
    let mut clear = false;
    let mut entropyfs = false;
    let mut i = 2;
    while i < args.len() {
        let a = args[i].as_str();
        let (flag, inline) = match a.split_once('=') {
            Some((f, v)) => (f, Some(v)),
            None => (a, None),
        };
        match flag {
            "--clear" => {
                clear = true;
                i += 1;
            }
            "--entropyfs" => {
                entropyfs = true;
                i += 1;
            }
            "--store" => {
                store_dir = Some(PathBuf::from(field_arg_value(
                    args, &mut i, "--store", inline,
                )?));
            }
            other => return Err(Error::usage(format!("unknown cache argument {other:?}"))),
        }
    }
    let store_dir = store_dir.ok_or_else(|| Error::usage("cache requires --store DIR"))?;
    let store = open_field_store(&store_dir, entropyfs)?;
    let cache = DerivedCache::open(store.root().join("cache"))?;
    if clear {
        let reclaimed = cache.clear()?;
        println!(
            "{{\"cache_bytes\":{},\"reclaimed\":{}}}",
            cache.total_bytes()?,
            reclaimed
        );
    } else {
        println!("{{\"cache_bytes\":{}}}", cache.total_bytes()?);
    }
    Ok(())
}

#[cfg(feature = "field")]
fn cmd_field_materialize(args: &[String], limits: Limits) -> Result<()> {
    let out = parse_field_args(args)?;
    let store_dir = out
        .store
        .as_deref()
        .ok_or_else(|| Error::usage("materialize requires --store DIR"))?;
    let field_hex = out
        .field
        .as_deref()
        .ok_or_else(|| Error::usage("materialize requires --field HEX"))?;
    let output = out
        .output
        .as_deref()
        .ok_or_else(|| Error::usage("materialize requires --output FILE"))?;
    let store = open_field_store(store_dir, out.entropyfs)?;
    let id = FieldId::from_hex(field_hex)?;
    let field = Field::open(&store, &id, limits)?;
    let bytes = field.materialize_exact(limits)?;
    write_atomic(output, &bytes)?;
    println!(
        "{{\"source_len\":{},\"sha256\":\"{}\"}}",
        bytes.len(),
        integrity::to_hex(&integrity::sha256(&bytes))
    );
    Ok(())
}

/// Write `bytes` to `path` atomically: temp sibling, fsync, rename, fsync dir.
fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
    let dir = match path.parent() {
        Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
        _ => PathBuf::from("."),
    };
    let name = path
        .file_name()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_else(|| "out".to_string());
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let tmp = dir.join(format!(".{name}.voldoc-tmp-{}-{nanos}", std::process::id()));

    {
        let mut f = fs::File::create(&tmp)?;
        f.write_all(bytes)?;
        f.sync_all()?;
    }
    if let Err(e) = fs::rename(&tmp, path) {
        let _ = fs::remove_file(&tmp);
        return Err(Error::from(e));
    }
    // Best-effort directory durability (Linux).
    if let Ok(d) = fs::File::open(&dir) {
        let _ = d.sync_all();
    }
    Ok(())
}